cleaning of the files event system

packages/meta-sort-core/src/container-plugins/ContainerPluginScheduler.ts

@@ -19,13 +19,9 @@ import type {
 } from './types.js';
 import type { TaskQueueType } from '../plugin-engine/types.js';
 import type { IKVClient } from '../kv/IKVClient.js';
-import { hasStreamSupport, type ExtendedContainerTask } from '../types/ExtendedInterfaces.js';
+import { type ExtendedContainerTask } from '../types/ExtendedInterfaces.js';
 import { config } from '../config/EnvConfig.js';
 
-// Redis Streams for reliable event delivery
-const EVENTS_STREAM = 'meta-sort:events';
-const STREAM_MAXLEN = 10000;
-
 /**
  * Scheduler events
  */
@@ -193,6 +189,7 @@ export class ContainerPluginScheduler extends EventEmitter {
 
     /**
      * Mark a plugin as completed for a file and notify waiters
+     * Note: Event publishing to meta-fuse is handled by meta-core (file:events stream)
      */
     private markPluginCompleted(fileHash: string, pluginId: string, filePath?: string): void {
         // Track completion
@@ -210,37 +207,6 @@ export class ContainerPluginScheduler extends EventEmitter {
             }
             this.dependencyWaiters.delete(key);
         }
-
-        // Publish to Redis for external subscribers (e.g., meta-stremio)
-        this.publishPluginComplete(fileHash, pluginId, filePath || '');
-    }
-
-    /**
-     * Publish plugin completion event to Redis Streams
-     *
-     * External services (like meta-fuse, meta-stremio) consume this stream
-     * to be notified when specific plugins complete for a file.
-     */
-    private async publishPluginComplete(fileHash: string, pluginId: string, filePath: string): Promise<void> {
-        if (!this.kvClient || !hasStreamSupport(this.kvClient)) {
-            return;
-        }
-
-        try {
-            const payload = JSON.stringify({
-                fileHash,
-                pluginId,
-                filePath,
-                timestamp: Date.now()
-            });
-            await this.kvClient.xadd(EVENTS_STREAM, STREAM_MAXLEN, {
-                type: 'plugin:complete',
-                payload,
-                timestamp: Date.now().toString()
-            });
-        } catch (error) {
-            // Silent fail - stream publish is optional
-        }
     }
 
     /**

packages/meta-sort-core/src/events/EventSubscriber.ts

@@ -1,270 +0,0 @@
-/**
- * EventSubscriber
- *
- * Subscribes to file events from meta-core via SSE (Server-Sent Events).
- * When events are received, they are forwarded to the StreamingPipeline.
- *
- * This enables meta-sort to run without direct filesystem access - it receives
- * file discovery events from meta-core and processes files via WebDAV.
- */
-
-import { EventSource } from 'eventsource';
-
-// File event types matching meta-core's watcher
-interface FileEvent {
-  type: 'add' | 'change' | 'delete' | 'rename';
-  path: string;           // Relative to FILES_PATH
-  size?: number;
-  timestamp: number;
-  partialHash?: string;   // Hash of first 64KB
-  oldPath?: string;       // For rename events
-}
-
-// Pipeline interface (minimal, matching StreamingPipeline)
-interface Pipeline {
-  handleFileAdded(filePath: string): void;
-  handleFileChanged(filePath: string): void;
-  handleFileDeleted(filePath: string): void;
-}
-
-export interface EventSubscriberOptions {
-  /** URL of meta-core API (e.g., http://meta-core or http://localhost:8083) */
-  metaCoreUrl: string;
-
-  /** Pipeline to receive events */
-  pipeline: Pipeline;
-
-  /** Base path for files (e.g., /files). Paths from meta-core are relative and need this prefix. */
-  filesPath?: string;
-
-  /** Whether to request an initial scan when connecting */
-  requestInitialScan?: boolean;
-
-  /** Reconnect delay in milliseconds */
-  reconnectDelayMs?: number;
-
-  /** Maximum reconnect attempts (0 = unlimited) */
-  maxReconnectAttempts?: number;
-}
-
-export class EventSubscriber {
-  private options: Required<EventSubscriberOptions>;
-  private eventSource: EventSource | null = null;
-  private isRunning = false;
-  private reconnectAttempts = 0;
-  private reconnectTimer: NodeJS.Timeout | null = null;
-
-  constructor(options: EventSubscriberOptions) {
-    this.options = {
-      filesPath: '/files',
-      requestInitialScan: true,
-      reconnectDelayMs: 5000,
-      maxReconnectAttempts: 0, // Unlimited by default
-      ...options
-    };
-  }
-
-  /**
-   * Start subscribing to events from meta-core
-   */
-  async start(): Promise<void> {
-    if (this.isRunning) {
-      console.log('[EventSubscriber] Already running');
-      return;
-    }
-
-    this.isRunning = true;
-    console.log(`[EventSubscriber] Connecting to ${this.options.metaCoreUrl}/api/events/subscribe`);
-
-    await this.connect();
-  }
-
-  /**
-   * Stop subscribing to events
-   */
-  async stop(): Promise<void> {
-    this.isRunning = false;
-
-    if (this.reconnectTimer) {
-      clearTimeout(this.reconnectTimer);
-      this.reconnectTimer = null;
-    }
-
-    if (this.eventSource) {
-      this.eventSource.close();
-      this.eventSource = null;
-    }
-
-    console.log('[EventSubscriber] Stopped');
-  }
-
-  // NOTE: triggerScan() has been removed - meta-core handles scanning automatically
-  // To trigger a scan manually, call meta-core's /api/scan/trigger endpoint directly
-
-  /**
-   * Get watcher status from meta-core
-   */
-  async getStatus(): Promise<{
-    status: string;
-    scanning: boolean;
-    lastScan: number;
-    fileCount: number;
-  } | null> {
-    try {
-      const response = await fetch(`${this.options.metaCoreUrl}/api/scan/status`);
-
-      if (!response.ok) {
-        return null;
-      }
-
-      return await response.json();
-    } catch (error) {
-      console.error('[EventSubscriber] Failed to get status:', error);
-      return null;
-    }
-  }
-
-  /**
-   * Poll for events since a timestamp (fallback for SSE)
-   */
-  async pollEvents(sinceMs: number, limit = 100): Promise<FileEvent[]> {
-    try {
-      const url = `${this.options.metaCoreUrl}/api/events/poll?since=${sinceMs}&limit=${limit}`;
-      const response = await fetch(url);
-
-      if (!response.ok) {
-        return [];
-      }
-
-      const data = await response.json();
-      return data.events || [];
-    } catch (error) {
-      console.error('[EventSubscriber] Failed to poll events:', error);
-      return [];
-    }
-  }
-
-  private async connect(): Promise<void> {
-    if (!this.isRunning) return;
-
-    const url = `${this.options.metaCoreUrl}/api/events/subscribe`;
-
-    try {
-      this.eventSource = new EventSource(url);
-
-      this.eventSource.onopen = () => {
-        console.log('[EventSubscriber] Connected to meta-core SSE');
-        this.reconnectAttempts = 0;
-        // NOTE: Initial scan is handled automatically by meta-core (Architecture V3)
-      };
-
-      this.eventSource.onerror = (error) => {
-        console.error('[EventSubscriber] SSE connection error:', error);
-        this.scheduleReconnect();
-      };
-
-      // Handle file events
-      this.eventSource.addEventListener('file', (event: MessageEvent) => {
-        try {
-          const fileEvent: FileEvent = JSON.parse(event.data);
-          this.handleEvent(fileEvent);
-        } catch (error) {
-          console.error('[EventSubscriber] Failed to parse event:', error);
-        }
-      });
-
-      // Handle connected event
-      this.eventSource.addEventListener('connected', () => {
-        console.log('[EventSubscriber] Received connected event');
-      });
-
-    } catch (error) {
-      console.error('[EventSubscriber] Failed to connect:', error);
-      this.scheduleReconnect();
-    }
-  }
-
-  private handleEvent(event: FileEvent): void {
-    const { pipeline, filesPath } = this.options;
-
-    // Convert relative path to absolute path
-    const toAbsolutePath = (relativePath: string): string => {
-      // If path is already absolute, return as-is
-      if (relativePath.startsWith('/')) {
-        return relativePath;
-      }
-      // Otherwise, prepend filesPath
-      return `${filesPath}/${relativePath}`;
-    };
-
-    const absolutePath = toAbsolutePath(event.path);
-
-    switch (event.type) {
-      case 'add':
-        pipeline.handleFileAdded(absolutePath);
-        break;
-
-      case 'change':
-        pipeline.handleFileChanged(absolutePath);
-        break;
-
-      case 'delete':
-        pipeline.handleFileDeleted(absolutePath);
-        break;
-
-      case 'rename':
-        // Handle rename as delete + add
-        if (event.oldPath) {
-          pipeline.handleFileDeleted(toAbsolutePath(event.oldPath));
-        }
-        pipeline.handleFileAdded(absolutePath);
-        break;
-
-      default:
-        console.warn(`[EventSubscriber] Unknown event type: ${event.type}`);
-    }
-  }
-
-  private scheduleReconnect(): void {
-    if (!this.isRunning) return;
-
-    // Close existing connection
-    if (this.eventSource) {
-      this.eventSource.close();
-      this.eventSource = null;
-    }
-
-    // Check reconnect limits
-    if (this.options.maxReconnectAttempts > 0 &&
-        this.reconnectAttempts >= this.options.maxReconnectAttempts) {
-      console.error('[EventSubscriber] Max reconnect attempts reached, giving up');
-      this.isRunning = false;
-      return;
-    }
-
-    this.reconnectAttempts++;
-    const delay = this.options.reconnectDelayMs * Math.min(this.reconnectAttempts, 6); // Max 30s
-
-    console.log(`[EventSubscriber] Reconnecting in ${delay}ms (attempt ${this.reconnectAttempts})`);
-
-    this.reconnectTimer = setTimeout(() => {
-      this.reconnectTimer = null;
-      this.connect();
-    }, delay);
-  }
-}
-
-/**
- * Create an EventSubscriber with default configuration
- */
-export function createEventSubscriber(
-  metaCoreUrl: string,
-  pipeline: Pipeline,
-  options?: Partial<EventSubscriberOptions>
-): EventSubscriber {
-  return new EventSubscriber({
-    metaCoreUrl,
-    pipeline,
-    ...options
-  });
-}

packages/meta-sort-core/src/events/FileEventConsumer.ts

@@ -0,0 +1,157 @@
+/**
+ * FileEventConsumer
+ *
+ * Consumes file events from meta-core via Redis Stream.
+ * When events are received, they are forwarded to the StreamingPipeline.
+ *
+ * This enables meta-sort to run without direct filesystem access - it receives
+ * file discovery events from meta-core and processes files via WebDAV.
+ */
+
+import type { RedisKVClient, StreamMessage } from '../kv/RedisClient.js';
+
+// Pipeline interface (minimal, matching StreamingPipeline)
+interface Pipeline {
+    handleFileAdded(filePath: string): void;
+    handleFileChanged(filePath: string): void;
+    handleFileDeleted(filePath: string): void;
+}
+
+// Redis Stream configuration
+const EVENTS_STREAM = 'file:events';
+const CONSUMER_GROUP = 'meta-sort-processor';
+
+export interface FileEventConsumerOptions {
+    /** Redis KV client */
+    kvClient: RedisKVClient;
+
+    /** Pipeline to receive events */
+    pipeline: Pipeline;
+
+    /** Base path for files (e.g., /files). Paths from meta-core are relative and need this prefix. */
+    filesPath?: string;
+
+    /** Block timeout in milliseconds for stream reads */
+    blockMs?: number;
+}
+
+export class FileEventConsumer {
+    private kvClient: RedisKVClient;
+    private pipeline: Pipeline;
+    private filesPath: string;
+    private blockMs: number;
+    private isRunning = false;
+
+    constructor(options: FileEventConsumerOptions) {
+        this.kvClient = options.kvClient;
+        this.pipeline = options.pipeline;
+        this.filesPath = options.filesPath ?? '/files';
+        this.blockMs = options.blockMs ?? 5000;
+    }
+
+    /**
+     * Start consuming events from Redis Stream
+     */
+    async start(): Promise<void> {
+        if (this.isRunning) {
+            console.log('[FileEventConsumer] Already running');
+            return;
+        }
+
+        this.isRunning = true;
+        console.log(`[FileEventConsumer] Starting stream consumer for ${EVENTS_STREAM}...`);
+
+        // Initialize consumer group
+        await this.kvClient.initStreamConsumer(EVENTS_STREAM, CONSUMER_GROUP);
+
+        // Process any pending entries from crashed consumers
+        await this.kvClient.processPendingEntries(
+            EVENTS_STREAM,
+            CONSUMER_GROUP,
+            30000, // 30 second idle threshold
+            async (message) => this.handleMessage(message)
+        );
+
+        // Start consumer loop in background (don't await - runs indefinitely)
+        this.kvClient.startStreamConsumer(
+            EVENTS_STREAM,
+            CONSUMER_GROUP,
+            async (message) => this.handleMessage(message),
+            this.blockMs
+        ).catch(error => {
+            console.error('[FileEventConsumer] Stream consumer error:', error);
+        });
+
+        console.log('[FileEventConsumer] Stream consumer started');
+    }
+
+    /**
+     * Stop consuming events
+     */
+    stop(): void {
+        if (!this.isRunning) {
+            return;
+        }
+
+        this.isRunning = false;
+        this.kvClient.stopStreamConsumer();
+        console.log('[FileEventConsumer] Stopped');
+    }
+
+    /**
+     * Handle a stream message
+     */
+    private async handleMessage(message: StreamMessage): Promise<void> {
+        // Convert relative path to absolute path
+        const toAbsolutePath = (relativePath: string): string => {
+            // If path is already absolute, return as-is
+            if (relativePath.startsWith('/')) {
+                return relativePath;
+            }
+            // Otherwise, prepend filesPath
+            return `${this.filesPath}/${relativePath}`;
+        };
+
+        const absolutePath = toAbsolutePath(message.path);
+
+        switch (message.type) {
+            case 'add':
+                this.pipeline.handleFileAdded(absolutePath);
+                break;
+
+            case 'change':
+                this.pipeline.handleFileChanged(absolutePath);
+                break;
+
+            case 'delete':
+                this.pipeline.handleFileDeleted(absolutePath);
+                break;
+
+            case 'rename':
+                // Handle rename as delete + add
+                if (message.oldPath) {
+                    this.pipeline.handleFileDeleted(toAbsolutePath(message.oldPath));
+                }
+                this.pipeline.handleFileAdded(absolutePath);
+                break;
+
+            default:
+                console.warn(`[FileEventConsumer] Unknown event type: ${message.type}`);
+        }
+    }
+}
+
+/**
+ * Create a FileEventConsumer with default configuration
+ */
+export function createFileEventConsumer(
+    kvClient: RedisKVClient,
+    pipeline: Pipeline,
+    options?: Partial<FileEventConsumerOptions>
+): FileEventConsumer {
+    return new FileEventConsumer({
+        kvClient,
+        pipeline,
+        ...options
+    });
+}

packages/meta-sort-core/src/index.ts

@@ -11,7 +11,8 @@ import {KVManager} from "./kv/KVManager.js";
 import {getPluginManagerInstance, initializePluginManager} from "./logic/fileProcessor/FileProcessorPiscina.js";
 import {ContainerManager} from "./container-plugins/ContainerManager.js";
 import {ContainerPluginScheduler} from "./container-plugins/ContainerPluginScheduler.js";
-import {EventSubscriber} from "./events/EventSubscriber.js";
+import {FileEventConsumer} from "./events/FileEventConsumer.js";
+import {RedisKVClient} from "./kv/RedisClient.js";
 
 try {
     // Get current user information
@@ -39,8 +40,8 @@ let kvManager: KVManager | null = null;
 let containerManager: ContainerManager | null = null;
 let containerPluginScheduler: ContainerPluginScheduler | null = null;
 
-// Event Subscriber for receiving events from meta-core (Architecture V3)
+// File Event Consumer for receiving events from meta-core via Redis Stream
-let eventSubscriber: EventSubscriber | null = null;
+let fileEventConsumer: FileEventConsumer | null = null;
 
 // Calculate concurrency for pipeline stages
 const cpuCount = os.cpus().length;
@@ -102,10 +103,10 @@ process.on('SIGINT', async () => {
         await apiServer.stop();
     }
 
-    // Stop Event Subscriber if running
+    // Stop File Event Consumer if running
-    if (eventSubscriber) {
+    if (fileEventConsumer) {
-        console.log('[Shutdown] Stopping Event Subscriber...');
+        console.log('[Shutdown] Stopping File Event Consumer...');
-        await eventSubscriber.stop();
+        fileEventConsumer.stop();
     }
 
     // Stop Container Manager if running
@@ -246,24 +247,21 @@ process.on('SIGINT', async () => {
             console.log(`[Cache] midhash256 cache: ${hits} hits, ${misses} misses (${hitRate}% hit rate)`);
         }
 
-        // Connect to meta-core for file events via SSE (Architecture V3)
+        // Connect to meta-core for file events via Redis Stream
-        // Get meta-core URL from leader discovery
+        // KV client must be available
-        const leaderInfo = kvManager?.getLeaderInfo();
+        if (!kvClient) {
-        if (!leaderInfo) {
+            console.error('[Startup] ERROR: KV client not available. Cannot consume file events.');
-            console.error('[Startup] ERROR: No leader info available. Cannot connect to meta-core.');
+            console.error('[Startup] Make sure meta-core is running and Redis is accessible.');
-            console.error('[Startup] Make sure meta-core is running and kv-leader.info exists.');
             process.exit(1);
         }
-        const metaCoreUrl = leaderInfo.apiUrl;
+        console.log('[Startup] Starting File Event Consumer (Redis Stream)...');
-        console.log(`[Startup] Connecting to meta-core at ${metaCoreUrl}...`);
+        fileEventConsumer = new FileEventConsumer({
-        eventSubscriber = new EventSubscriber({
+            kvClient: kvClient as RedisKVClient,
-            metaCoreUrl: metaCoreUrl,
             pipeline: pipeline,
-            requestInitialScan: true, // Request initial scan from meta-core
+            filesPath: config.FILES_PATH,
-            reconnectDelayMs: 5000,
         });
-        await eventSubscriber.start();
+        await fileEventConsumer.start();
-        console.log('[Startup] EventSubscriber connected to meta-core');
+        console.log('[Startup] FileEventConsumer started, consuming from file:events stream');
 
     } catch (error) {
         console.error('[Startup] Error during startup:', error);

packages/meta-sort-core/src/kv/RedisClient.ts

@@ -13,6 +13,7 @@
  */
 
 import * as IORedis from 'ioredis';
+import * as os from 'os';
 import type { IKVClient, KeyValuePair } from './IKVClient.js';
 import {
     flattenMetadata,
@@ -25,6 +26,25 @@ import {
 const Redis = (IORedis as any).default ?? IORedis;
 type RedisInstance = InstanceType<typeof Redis>;
 
+/**
+ * Stream message from Redis Streams
+ * File events from meta-core: add, change, delete, rename
+ */
+export interface StreamMessage {
+    id: string;
+    type: 'add' | 'change' | 'delete' | 'rename';
+    path: string;
+    size?: string;
+    partialHash?: string;
+    oldPath?: string;
+    timestamp: string;
+}
+
+/**
+ * Stream consumer callback
+ */
+export type StreamMessageHandler = (message: StreamMessage) => Promise<void>;
+
 interface RedisClientConfig {
     url: string;
     timeout?: number;
@@ -37,10 +57,18 @@ export class RedisKVClient implements IKVClient {
     private prefix: string;
     private isConnected: boolean = false;
 
+    // Stream consumer state
+    private streamConsumerRunning = false;
+    private streamConsumerAbort: AbortController | null = null;
+    private consumerName: string;
+
     constructor(config: RedisClientConfig) {
         this.timeout = config.timeout || 30000;
         this.prefix = config.prefix || '';
 
+        // Unique consumer name for this instance
+        this.consumerName = `${os.hostname()}-${process.pid}`;
+
         // Parse Redis URL and create client
         this.redis = new Redis(config.url, {
             commandTimeout: this.timeout,
@@ -371,11 +399,226 @@ export class RedisKVClient implements IKVClient {
         return result as string;
     }
 
+    // ========================================================================
+    // Redis Streams Consumer Methods
+    // ========================================================================
+
+    /**
+     * Initialize stream consumer group
+     * Creates the consumer group at position 0 to read all historical events
+     *
+     * @param stream - Stream name (e.g., 'file:events')
+     * @param group - Consumer group name (e.g., 'meta-sort-processor')
+     */
+    async initStreamConsumer(stream: string, group: string): Promise<void> {
+        try {
+            // Create consumer group at position 0 (read all historical events)
+            // MKSTREAM creates the stream if it doesn't exist
+            await this.redis.xgroup('CREATE', stream, group, '0', 'MKSTREAM');
+            console.log(`[Redis] Created consumer group '${group}' for stream '${stream}'`);
+        } catch (error: any) {
+            // BUSYGROUP means group already exists - that's fine
+            if (error.message?.includes('BUSYGROUP')) {
+                console.log(`[Redis] Consumer group '${group}' already exists`);
+            } else {
+                throw error;
+            }
+        }
+    }
+
+    /**
+     * Process pending entries from crashed consumers
+     * Uses XAUTOCLAIM to claim entries that have been idle too long
+     *
+     * @param stream - Stream name
+     * @param group - Consumer group name
+     * @param minIdleTime - Minimum idle time in ms before claiming (default: 30000)
+     * @param onMessage - Handler for each message
+     */
+    async processPendingEntries(
+        stream: string,
+        group: string,
+        minIdleTime: number = 30000,
+        onMessage: StreamMessageHandler
+    ): Promise<number> {
+        let processed = 0;
+        let cursor = '0-0';
+
+        try {
+            while (true) {
+                // XAUTOCLAIM claims idle entries and returns them
+                const result = await this.redis.xautoclaim(
+                    stream,
+                    group,
+                    this.consumerName,
+                    minIdleTime,
+                    cursor,
+                    'COUNT',
+                    100
+                ) as [string, Array<[string, string[]]>, string[]];
+
+                const [nextCursor, entries] = result;
+                cursor = nextCursor;
+
+                if (!entries || entries.length === 0) {
+                    break;
+                }
+
+                for (const [id, fields] of entries) {
+                    try {
+                        const message = this.parseStreamEntry(id, fields);
+                        if (message) {
+                            await onMessage(message);
+                            // ACK the message after successful processing
+                            await this.redis.xack(stream, group, id);
+                            processed++;
+                        }
+                    } catch (error: any) {
+                        console.error(`[Redis] Error processing pending entry ${id}:`, error.message);
+                    }
+                }
+
+                // If cursor is 0-0, we've processed all pending entries
+                if (cursor === '0-0') {
+                    break;
+                }
+            }
+
+            if (processed > 0) {
+                console.log(`[Redis] Processed ${processed} pending stream entries`);
+            }
+        } catch (error: any) {
+            console.error('[Redis] Error processing pending entries:', error.message);
+        }
+
+        return processed;
+    }
+
+    /**
+     * Start stream consumer loop
+     * Reads messages using XREADGROUP and calls the handler
+     *
+     * @param stream - Stream name
+     * @param group - Consumer group name
+     * @param onMessage - Handler for each message
+     * @param blockMs - Block timeout in ms (default: 5000)
+     */
+    async startStreamConsumer(
+        stream: string,
+        group: string,
+        onMessage: StreamMessageHandler,
+        blockMs: number = 5000
+    ): Promise<void> {
+        if (this.streamConsumerRunning) {
+            console.warn('[Redis] Stream consumer already running');
+            return;
+        }
+
+        this.streamConsumerRunning = true;
+        this.streamConsumerAbort = new AbortController();
+
+        console.log(`[Redis] Starting stream consumer: ${stream} (group: ${group}, consumer: ${this.consumerName})`);
+
+        // Consumer loop
+        while (this.streamConsumerRunning && !this.streamConsumerAbort.signal.aborted) {
+            try {
+                // XREADGROUP blocks waiting for new messages
+                // '>' means only read new messages (not already delivered to this consumer)
+                const result = await (this.redis.call(
+                    'XREADGROUP',
+                    'GROUP', group,
+                    this.consumerName,
+                    'BLOCK', blockMs,
+                    'COUNT', 10,
+                    'STREAMS', stream,
+                    '>'
+                ) as Promise<Array<[string, Array<[string, string[]]>]> | null>);
+
+                if (!result || result.length === 0) {
+                    continue; // Timeout, loop again
+                }
+
+                // Process each stream's messages
+                for (const [streamName, entries] of result) {
+                    for (const [id, fields] of entries) {
+                        try {
+                            const message = this.parseStreamEntry(id, fields);
+                            if (message) {
+                                await onMessage(message);
+                                // ACK the message after successful processing
+                                await this.redis.xack(stream, group, id);
+                            }
+                        } catch (error: any) {
+                            console.error(`[Redis] Error processing stream entry ${id}:`, error.message);
+                            // Don't ACK on error - message will be reprocessed
+                        }
+                    }
+                }
+            } catch (error: any) {
+                if (this.streamConsumerAbort?.signal.aborted) {
+                    break;
+                }
+                console.error('[Redis] Stream consumer error:', error.message);
+                // Brief pause before retrying
+                await new Promise(resolve => setTimeout(resolve, 1000));
+            }
+        }
+
+        console.log('[Redis] Stream consumer stopped');
+    }
+
+    /**
+     * Stop the stream consumer loop
+     */
+    stopStreamConsumer(): void {
+        if (!this.streamConsumerRunning) {
+            return;
+        }
+
+        console.log('[Redis] Stopping stream consumer...');
+        this.streamConsumerRunning = false;
+        this.streamConsumerAbort?.abort();
+        this.streamConsumerAbort = null;
+    }
+
+    /**
+     * Parse a stream entry into a StreamMessage
+     */
+    private parseStreamEntry(id: string, fields: string[]): StreamMessage | null {
+        // Fields come as flat array: ['type', 'add', 'path', '...', 'timestamp', '123']
+        const fieldMap: Record<string, string> = {};
+        for (let i = 0; i < fields.length; i += 2) {
+            fieldMap[fields[i]] = fields[i + 1];
+        }
+
+        if (!fieldMap.type) {
+            console.warn(`[Redis] Invalid stream entry ${id}: missing type`);
+            return null;
+        }
+
+        if (!fieldMap.path) {
+            console.warn(`[Redis] Invalid stream entry ${id}: missing path for ${fieldMap.type} event`);
+            return null;
+        }
+
+        return {
+            id,
+            type: fieldMap.type as StreamMessage['type'],
+            path: fieldMap.path,
+            size: fieldMap.size,
+            partialHash: fieldMap.partialHash,
+            oldPath: fieldMap.oldPath,
+            timestamp: fieldMap.timestamp || '0',
+        };
+    }
+
     // ========================================================================
     // Lifecycle
     // ========================================================================
 
     async close(): Promise<void> {
+        // Stop stream consumer first
+        this.stopStreamConsumer();
         await this.redis.quit();
         this.isConnected = false;
     }

packages/meta-sort-core/src/logic/WatchedFileProcessor.ts

@@ -12,12 +12,8 @@ import {VirtualFileSystem} from "../api/VirtualFileSystem.js";
 import {performanceMetrics} from "../metrics/PerformanceMetrics.js";
 import {UnifiedProcessingStateManager} from "./UnifiedProcessingStateManager.js";
 import type {IKVClient} from "../kv/IKVClient.js";
-import { hasStreamSupport, type ExtendedFileAnalyzer, type ExtendedHashMeta } from "../types/ExtendedInterfaces.js";
+import { type ExtendedFileAnalyzer, type ExtendedHashMeta } from "../types/ExtendedInterfaces.js";
 import * as webdav from '../webdav/WebdavClient.js';
-
-// Redis Streams for reliable event delivery
-const EVENTS_STREAM = 'meta-sort:events';
-const STREAM_MAXLEN = 10000;
 import PQueue from "p-queue";
 import os from 'os';
 import { TaskScheduler, createTaskScheduler, MetadataNodeKVStore } from "../plugin-engine/index.js";
@@ -205,7 +201,8 @@ export class WatchedFileProcessor implements FileProcessorInterface {
     }
 
     /**
-     * Set up TaskScheduler event listeners for VFS integration and Redis pub/sub
+     * Set up TaskScheduler event listeners for VFS integration
+     * Note: Event publishing to meta-fuse is handled by meta-core (file:events stream)
      */
     private setupTaskSchedulerEvents(): void {
         if (!this.taskScheduler) return;
@@ -218,9 +215,6 @@ export class WatchedFileProcessor implements FileProcessorInterface {
             if (task.pluginId === 'filename-parser') {
                 this.updateVFSForFile(task.fileHash, task.filePath);
             }
-
-            // Publish plugin completion event to Redis
-            this.publishPluginComplete(task.fileHash, task.pluginId, task.filePath);
         });
 
         // When all tasks for a file complete
@@ -238,32 +232,6 @@ export class WatchedFileProcessor implements FileProcessorInterface {
         });
     }
 
-    /**
-     * Publish plugin completion event to Redis Streams
-     */
-    private async publishPluginComplete(fileHash: string, pluginId: string, filePath: string): Promise<void> {
-        const kvClient = this.getKVClient();
-        if (!kvClient || !hasStreamSupport(kvClient)) {
-            return;
-        }
-
-        try {
-            const payload = JSON.stringify({
-                fileHash,
-                pluginId,
-                filePath,
-                timestamp: Date.now()
-            });
-            await kvClient.xadd(EVENTS_STREAM, STREAM_MAXLEN, {
-                type: 'plugin:complete',
-                payload,
-                timestamp: Date.now().toString()
-            });
-        } catch (error) {
-            // Silent fail - stream publish is optional
-        }
-    }
-
     /**
      * Update VFS for a file with current metadata
      */

packages/meta-sort-core/src/logic/pipeline/StreamingPipeline.ts

@@ -2,22 +2,6 @@ import * as path from 'path';
 import PQueue from 'p-queue';
 import {PipelineConfig} from './PipelineConfig.js';
 import {performanceMetrics} from '../../metrics/PerformanceMetrics.js';
-import { hasStreamSupport } from '../../types/ExtendedInterfaces.js';
-
-// Redis Streams for reliable event delivery to meta-fuse
-// Single stream with type field for all event types (batch, reset, plugin:complete)
-const EVENTS_STREAM = 'meta-sort:events';
-
-// Maximum stream length (approximate, ~10000 entries provides hours of history)
-const STREAM_MAXLEN = 10000;
-
-// Batch update interval (5 seconds)
-const BATCH_INTERVAL_MS = 5000;
-
-interface BatchChange {
-    action: 'add' | 'update' | 'remove';
-    hashId: string;
-}
 
 /**
  * Streaming Pipeline for file processing
@@ -26,7 +10,7 @@ interface BatchChange {
  *
  * With midhash256 being instant (< 1s), files appear in VFS immediately
  *
- * Publishes batched updates to meta-fuse via Redis Streams every 5 seconds.
+ * Note: Event publishing to meta-fuse is handled by meta-core (file:events stream)
  */
 export class StreamingPipeline {
     private validationQueue: PQueue;
@@ -39,10 +23,6 @@ export class StreamingPipeline {
     private fastProcessedCount = 0;
     private backgroundProcessedCount = 0;
 
-    // Batch buffer for stream notifications
-    private batchBuffer: BatchChange[] = [];
-    private batchTimer: NodeJS.Timeout | null = null;
-
     constructor(config: PipelineConfig) {
         this.config = config;
 
@@ -65,72 +45,6 @@ export class StreamingPipeline {
         });
 
         console.log(`[Pipeline] Initialized with concurrency: validation=${config.validationConcurrency}, fastQueue=${config.fastQueueConcurrency}, backgroundQueue=${config.backgroundQueueConcurrency}`);
-
-        // Start batch publish timer
-        this.startBatchTimer();
-    }
-
-    /**
-     * Start the batch publish timer
-     * Publishes accumulated changes to meta-fuse every 5 seconds
-     */
-    private startBatchTimer(): void {
-        this.batchTimer = setInterval(() => {
-            this.flushBatch().catch(err => {
-                console.error('[Pipeline] Failed to flush batch:', err.message);
-            });
-        }, BATCH_INTERVAL_MS);
-    }
-
-    /**
-     * Stop the batch publish timer
-     */
-    stopBatchTimer(): void {
-        if (this.batchTimer) {
-            clearInterval(this.batchTimer);
-            this.batchTimer = null;
-        }
-    }
-
-    /**
-     * Queue a change for batched notification
-     */
-    private queueChange(action: 'add' | 'update' | 'remove', hashId: string): void {
-        // Deduplicate: remove existing entries for same hashId
-        this.batchBuffer = this.batchBuffer.filter(c => c.hashId !== hashId);
-        this.batchBuffer.push({ action, hashId });
-    }
-
-    /**
-     * Flush the batch buffer and publish to Redis Streams
-     */
-    private async flushBatch(): Promise<void> {
-        if (this.batchBuffer.length === 0) {
-            return;
-        }
-
-        const changes = [...this.batchBuffer];
-        this.batchBuffer = [];
-
-        try {
-            const kvClient = this.config.kvClient;
-            if (kvClient && hasStreamSupport(kvClient)) {
-                const payload = JSON.stringify({
-                    timestamp: Date.now(),
-                    changes
-                });
-                await kvClient.xadd(EVENTS_STREAM, STREAM_MAXLEN, {
-                    type: 'batch',
-                    payload,
-                    timestamp: Date.now().toString()
-                });
-                console.log(`[Pipeline] Published batch update to stream: ${changes.length} changes`);
-            }
-        } catch (error: any) {
-            // Put changes back in buffer on failure
-            this.batchBuffer = [...changes, ...this.batchBuffer];
-            console.warn(`[Pipeline] Failed to publish batch: ${error.message}`);
-        }
     }
 
     /**
@@ -233,11 +147,6 @@ export class StreamingPipeline {
                     // Add to VFS immediately (file appears with permanent midhash256 ID)
                     this.config.virtualFileSystem.addFile(virtualPath, filePath, metadata);
 
-                    // Queue for pub/sub notification to meta-fuse
-                    if (metadata.cid_midhash256) {
-                        this.queueChange('add', metadata.cid_midhash256);
-                    }
-
                     // Notify Stremio addon to add video (incremental update like FUSE)
                     if (metadata.cid_midhash256) {
                         this.notifyStremioAdd(metadata.cid_midhash256).catch(err => {
@@ -346,20 +255,12 @@ export class StreamingPipeline {
      * Handle file deleted event (from chokidar)
      */
     async handleFileDeleted(filePath: string): Promise<void> {
-        // Get hashId before removal for pub/sub notification
-        const metadata = this.config.fileProcessor.getDatabase().get(filePath);
-        const hashId = (metadata as any)?.cid_midhash256;
-
         // Remove from state and call cleanup
         if (this.config.fileProcessor.deleteFile) {
             this.config.fileProcessor.deleteFile(filePath);
         }
         this.config.stateManager.removeFile(filePath);
-
+        // Note: meta-core handles event publishing to meta-fuse via file:events stream
-        // Queue for pub/sub notification to meta-fuse
-        if (hashId) {
-            this.queueChange('remove', hashId);
-        }
     }
 
     /**
@@ -412,7 +313,7 @@ export class StreamingPipeline {
     /**
      * Reset pipeline counters and clear state manager
      * Called when triggering a fresh scan to reset statistics
-     * Publishes reset event to notify meta-fuse to clear its VFS
+     * Note: meta-fuse listens to file:events stream from meta-core for updates
      */
     async reset(): Promise<void> {
         console.log('[Pipeline] Resetting counters and state...');
@@ -423,30 +324,9 @@ export class StreamingPipeline {
         this.fastProcessedCount = 0;
         this.backgroundProcessedCount = 0;
 
-        // Clear batch buffer
-        this.batchBuffer = [];
-
         // Clear state manager (discovered, processing, done states)
         this.config.stateManager.clear();
 
-        // Publish reset event to Redis Streams to notify meta-fuse
-        if (this.config.kvClient && hasStreamSupport(this.config.kvClient)) {
-            try {
-                const payload = JSON.stringify({
-                    timestamp: Date.now(),
-                    action: 'reset'
-                });
-                await this.config.kvClient.xadd(EVENTS_STREAM, STREAM_MAXLEN, {
-                    type: 'reset',
-                    payload,
-                    timestamp: Date.now().toString()
-                });
-                console.log('[Pipeline] Reset event published to stream');
-            } catch (error) {
-                console.error('[Pipeline] Failed to publish reset event:', error);
-            }
-        }
-
         console.log('[Pipeline] Reset complete');
     }