bugfix on service discovery

packages/meta-sort-core/src/api/UnifiedAPIServer.ts

@@ -988,6 +988,7 @@ export class UnifiedAPIServer {
         url: string;
         api: string;
         status: string;
+        role?: string;
       }> = [];
 
       try {
@@ -1001,6 +1002,7 @@ export class UnifiedAPIServer {
               url: svc.baseUrl || '',
               api: svc.baseUrl || '',
               status: svc.status || 'unknown',
+              role: svc.role,
             });
           }
         }

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

@@ -19,9 +19,13 @@ import type {
 } from './types.js';
 import type { TaskQueueType } from '../plugin-engine/types.js';
 import type { IKVClient } from '../kv/IKVClient.js';
-import { hasPublish, type ExtendedContainerTask } from '../types/ExtendedInterfaces.js';
+import { hasStreamSupport, 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
  */
@@ -212,27 +216,30 @@ export class ContainerPluginScheduler extends EventEmitter {
     }
 
     /**
-     * Publish plugin completion event to Redis
+     * Publish plugin completion event to Redis Streams
-     * Channel: meta-sort:plugin:complete
      *
-     * External services (like meta-stremio) can subscribe to this channel
+     * 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 || !hasPublish(this.kvClient)) {
+        if (!this.kvClient || !hasStreamSupport(this.kvClient)) {
             return;
         }
 
         try {
-            const message = JSON.stringify({
+            const payload = JSON.stringify({
                 fileHash,
                 pluginId,
                 filePath,
                 timestamp: Date.now()
             });
-            await this.kvClient.publish('meta-sort:plugin:complete', message);
+            await this.kvClient.xadd(EVENTS_STREAM, STREAM_MAXLEN, {
+                type: 'plugin:complete',
+                payload,
+                timestamp: Date.now().toString()
+            });
         } catch (error) {
-            // Silent fail - pub/sub is optional
+            // Silent fail - stream publish is optional
         }
     }
 

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

@@ -184,4 +184,7 @@ export interface ServiceInfo {
 
     /** Last heartbeat timestamp (ISO string) */
     lastHeartbeat: string;
+
+    /** Role for meta-core instances: 'leader', 'follower', or undefined for other services */
+    role?: string;
 }

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

@@ -130,8 +130,7 @@ export class KVManager {
         // Start watching for leader changes
         this.leaderClient.startWatching();
 
-        // Initialize service discovery (meta-core handles actual registration)
+        // Initialize and start service discovery for registration and heartbeat
-        // We keep this for service discovery (reading other services)
         const apiUrl = this.config.baseUrl || `http://${hostname()}:${this.config.apiPort}`;
         this.serviceDiscovery = new ServiceDiscovery({
             metaCorePath: this.config.metaCorePath,
@@ -139,7 +138,7 @@ export class KVManager {
             version: '1.0.0',
             apiUrl
         });
-        // Note: We don't start ServiceDiscovery - meta-core handles registration
+        await this.serviceDiscovery.start();
 
         this.isStarted = true;
         this.notifyReady();

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

@@ -39,18 +39,87 @@ export class LeaderClient {
     private watcher: FSWatcher | null = null;
     private onChangeCallbacks: (() => void)[] = [];
 
+    // URL caching
+    private cachedUrls: URLsResponse | null = null;
+    private urlsCacheTime: number = 0;
+    private readonly urlsCacheTTL: number = 5000; // 5 seconds
+
     constructor(config: LeaderClientConfig) {
         this.config = config;
         this.infoFilePath = `${config.metaCorePath}/locks/kv-leader.info`;
     }
 
     /**
-     * Read leader info from the info file
+     * Read API URL from file (plain text format)
+     */
+    private async getApiUrlFromFile(): Promise<string | null> {
+        try {
+            const content = await fs.readFile(this.infoFilePath, 'utf-8');
+            return content.trim() || null;
+        } catch (error) {
+            console.error('[LeaderClient] Failed to read API URL from file:', error);
+            return null;
+        }
+    }
+
+    /**
+     * Fetch URLs from meta-core /urls API with caching
+     */
+    private async fetchUrls(apiUrl: string): Promise<URLsResponse | null> {
+        // Check cache
+        const now = Date.now();
+        if (this.cachedUrls && (now - this.urlsCacheTime) < this.urlsCacheTTL) {
+            return this.cachedUrls;
+        }
+
+        try {
+            const response = await fetch(`${apiUrl}/urls`, {
+                method: 'GET',
+                headers: { 'Accept': 'application/json' },
+                signal: AbortSignal.timeout(5000)
+            });
+
+            if (!response.ok) {
+                console.error('[LeaderClient] Failed to fetch URLs:', response.status, response.statusText);
+                return null;
+            }
+
+            this.cachedUrls = await response.json() as URLsResponse;
+            this.urlsCacheTime = now;
+            return this.cachedUrls;
+        } catch (error) {
+            console.error('[LeaderClient] Error calling /urls API:', error);
+            return null;
+        }
+    }
+
+    /**
+     * Read leader info from file and /urls API
      */
     async getLeaderInfo(): Promise<LeaderLockInfo | null> {
         try {
-            const content = await fs.readFile(this.infoFilePath, 'utf-8');
+            // Read API URL from file (plain text)
-            this.leaderInfo = JSON.parse(content) as LeaderLockInfo;
+            const apiUrl = await this.getApiUrlFromFile();
+            if (!apiUrl) {
+                return null;
+            }
+
+            // Fetch full info from /urls API
+            const urls = await this.fetchUrls(apiUrl);
+            if (!urls) {
+                return null;
+            }
+
+            // Convert URLsResponse to LeaderLockInfo
+            this.leaderInfo = {
+                hostname: urls.hostname,
+                baseUrl: urls.baseUrl,
+                apiUrl: urls.apiUrl,
+                redisUrl: urls.redisUrl,
+                webdavUrl: urls.webdavUrl,
+                timestamp: Date.now(),
+                pid: 0 // Unknown for remote leader
+            };
             return this.leaderInfo;
         } catch (error) {
             console.error('[LeaderClient] Failed to read leader info:', error);
@@ -90,10 +159,9 @@ export class LeaderClient {
         // First try using configured metaCoreUrl
         let apiUrl = this.config.metaCoreUrl;
 
-        // Fall back to reading from leader info
+        // Fall back to reading from file
         if (!apiUrl) {
-            const info = await this.getLeaderInfo();
+            apiUrl = await this.getApiUrlFromFile();
-            apiUrl = info?.apiUrl;
         }
 
         if (!apiUrl) {
@@ -101,23 +169,7 @@ export class LeaderClient {
             return null;
         }
 
-        try {
+        return this.fetchUrls(apiUrl);
-            const response = await fetch(`${apiUrl}/urls`, {
-                method: 'GET',
-                headers: { 'Accept': 'application/json' },
-                signal: AbortSignal.timeout(5000)
-            });
-
-            if (!response.ok) {
-                console.error('[LeaderClient] Failed to get URLs:', response.status, response.statusText);
-                return null;
-            }
-
-            return await response.json() as URLsResponse;
-        } catch (error) {
-            console.error('[LeaderClient] Error calling /urls API:', error);
-            return null;
-        }
     }
 
     /**
@@ -154,7 +206,11 @@ export class LeaderClient {
         try {
             this.watcher = watch(lockDir, (eventType, filename) => {
                 if (filename === 'kv-leader.info') {
-                    console.log('[LeaderClient] Leader info changed, reloading...');
+                    console.log('[LeaderClient] Leader info changed, invalidating cache...');
+                    // Invalidate cache to force fresh API call
+                    this.cachedUrls = null;
+                    this.urlsCacheTime = 0;
+
                     this.getLeaderInfo().then(() => {
                         this.notifyChange();
                     }).catch(console.error);

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

@@ -338,6 +338,39 @@ export class RedisKVClient implements IKVClient {
         return await this.redis.publish(channel, message);
     }
 
+    // ========================================================================
+    // Redis Streams Operations
+    // ========================================================================
+
+    /**
+     * Add an entry to a Redis stream
+     * Used for reliable event delivery between services (replaces pub/sub)
+     *
+     * @param stream - Stream name (e.g., 'meta-sort:events')
+     * @param maxlen - Maximum stream length (approximate, uses ~ for efficiency)
+     * @param fields - Object containing field-value pairs to add
+     * @returns The ID of the added entry
+     */
+    async xadd(stream: string, maxlen: number, fields: Record<string, string>): Promise<string> {
+        // Convert fields object to flat array for ioredis
+        const fieldArray: string[] = [];
+        for (const [key, value] of Object.entries(fields)) {
+            fieldArray.push(key, value);
+        }
+
+        // Use XADD with MAXLEN ~ for approximate trimming (more efficient)
+        const result = await this.redis.xadd(
+            stream,
+            'MAXLEN',
+            '~',
+            maxlen.toString(),
+            '*',
+            ...fieldArray
+        );
+
+        return result as string;
+    }
+
     // ========================================================================
     // Lifecycle
     // ========================================================================

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

@@ -7,8 +7,8 @@
  * Services can discover each other by reading these JSON files.
  * Stale detection: if lastHeartbeat > 60 seconds ago, service is considered stale.
  *
- * Note: meta-core (Go sidecar) handles actual service registration.
+ * Each service must call start() to register itself and begin heartbeat updates.
- * This class is primarily used for discovering other services.
+ * meta-core runs a cleanup process that removes services with stale heartbeats.
  */
 
 import { promises as fs } from 'fs';
@@ -79,9 +79,8 @@ export class ServiceDiscovery {
     }
 
     /**
-     * Register this service
+     * Register this service by writing service info to the shared filesystem.
-     * Note: In the new architecture, meta-core handles registration.
+     * meta-core's cleanup process monitors these files and removes stale entries.
-     * This is kept for backwards compatibility and testing.
      */
     async register(): Promise<void> {
         await this.ensureServicesDir();
@@ -314,8 +313,7 @@ export class ServiceDiscovery {
     // ========================================================================
 
     /**
-     * Full startup sequence
+     * Full startup sequence: register service and start heartbeat loop.
-     * Note: In the new architecture, meta-core handles registration.
      */
     async start(): Promise<void> {
         await this.register();

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

@@ -15,7 +15,11 @@ 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 { hasPublish, type ExtendedFileAnalyzer, type ExtendedHashMeta } from "../types/ExtendedInterfaces.js";
+import { hasStreamSupport, type ExtendedFileAnalyzer, type ExtendedHashMeta } from "../types/ExtendedInterfaces.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";
@@ -244,25 +248,28 @@ export class WatchedFileProcessor implements FileProcessorInterface {
     }
 
     /**
-     * Publish plugin completion event to Redis
+     * Publish plugin completion event to Redis Streams
-     * Channel: meta-sort:plugin:complete
      */
     private async publishPluginComplete(fileHash: string, pluginId: string, filePath: string): Promise<void> {
         const kvClient = this.getKVClient();
-        if (!kvClient || !hasPublish(kvClient)) {
+        if (!kvClient || !hasStreamSupport(kvClient)) {
             return;
         }
 
         try {
-            const message = JSON.stringify({
+            const payload = JSON.stringify({
                 fileHash,
                 pluginId,
                 filePath,
                 timestamp: Date.now()
             });
-            await kvClient.publish('meta-sort:plugin:complete', message);
+            await kvClient.xadd(EVENTS_STREAM, STREAM_MAXLEN, {
+                type: 'plugin:complete',
+                payload,
+                timestamp: Date.now().toString()
+            });
         } catch (error) {
-            // Silent fail - pub/sub is optional
+            // Silent fail - stream publish is optional
         }
     }
 

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

@@ -2,12 +2,14 @@ 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';
 
-// Pub/sub channel for batched updates to meta-fuse
+// Redis Streams for reliable event delivery to meta-fuse
-const UPDATE_CHANNEL = 'meta-sort:file:batch';
+// Single stream with type field for all event types (batch, reset, plugin:complete)
+const EVENTS_STREAM = 'meta-sort:events';
 
-// Pub/sub channel for scan reset notification
+// Maximum stream length (approximate, ~10000 entries provides hours of history)
-const RESET_CHANNEL = 'meta-sort:scan:reset';
+const STREAM_MAXLEN = 10000;
 
 // Batch update interval (5 seconds)
 const BATCH_INTERVAL_MS = 5000;
@@ -24,7 +26,7 @@ interface BatchChange {
  *
  * With midhash256 being instant (< 1s), files appear in VFS immediately
  *
- * Publishes batched updates to meta-fuse via Redis pub/sub every 5 seconds.
+ * Publishes batched updates to meta-fuse via Redis Streams every 5 seconds.
  */
 export class StreamingPipeline {
     private validationQueue: PQueue;
@@ -37,7 +39,7 @@ export class StreamingPipeline {
     private fastProcessedCount = 0;
     private backgroundProcessedCount = 0;
 
-    // Batch buffer for pub/sub notifications
+    // Batch buffer for stream notifications
     private batchBuffer: BatchChange[] = [];
     private batchTimer: NodeJS.Timeout | null = null;
 
@@ -100,7 +102,7 @@ export class StreamingPipeline {
     }
 
     /**
-     * Flush the batch buffer and publish to Redis
+     * Flush the batch buffer and publish to Redis Streams
      */
     private async flushBatch(): Promise<void> {
         if (this.batchBuffer.length === 0) {
@@ -112,13 +114,17 @@ export class StreamingPipeline {
 
         try {
             const kvClient = this.config.kvClient;
-            if (kvClient && typeof (kvClient as any).publish === 'function') {
+            if (kvClient && hasStreamSupport(kvClient)) {
-                const message = JSON.stringify({
+                const payload = JSON.stringify({
                     timestamp: Date.now(),
                     changes
                 });
-                await (kvClient as any).publish(UPDATE_CHANNEL, message);
+                await kvClient.xadd(EVENTS_STREAM, STREAM_MAXLEN, {
-                console.log(`[Pipeline] Published batch update: ${changes.length} changes`);
+                    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
@@ -423,15 +429,19 @@ export class StreamingPipeline {
         // Clear state manager (discovered, processing, done states)
         this.config.stateManager.clear();
 
-        // Publish reset event to notify meta-fuse
+        // Publish reset event to Redis Streams to notify meta-fuse
-        if (this.config.kvClient && typeof (this.config.kvClient as any).publish === 'function') {
+        if (this.config.kvClient && hasStreamSupport(this.config.kvClient)) {
             try {
-                const message = JSON.stringify({
+                const payload = JSON.stringify({
                     timestamp: Date.now(),
                     action: 'reset'
                 });
-                await (this.config.kvClient as any).publish(RESET_CHANNEL, message);
+                await this.config.kvClient.xadd(EVENTS_STREAM, STREAM_MAXLEN, {
-                console.log('[Pipeline] Reset event published to meta-fuse');
+                    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);
             }

packages/meta-sort-core/src/types/ExtendedInterfaces.ts

@@ -41,6 +41,37 @@ export function hasPublish(client: IKVClient): client is IKVClientWithPubSub {
     return typeof (client as IKVClientWithPubSub).publish === 'function';
 }
 
+/**
+ * IKVClient with Redis Streams capabilities
+ *
+ * Redis Streams provide reliable event delivery with persistence,
+ * consumer groups, and replay capabilities. Preferred over pub/sub
+ * for service-to-service communication.
+ */
+export interface IKVClientWithStreams extends IKVClient {
+    /**
+     * Add an entry to a Redis stream
+     *
+     * @param stream - Stream name (e.g., 'meta-sort:events')
+     * @param maxlen - Maximum stream length (approximate, uses ~ for efficiency)
+     * @param fields - Object containing field-value pairs to add
+     * @returns The ID of the added entry
+     */
+    xadd(stream: string, maxlen: number, fields: Record<string, string>): Promise<string>;
+
+    /**
+     * Get the underlying Redis client (implementation-specific)
+     */
+    getRedisClient?(): unknown;
+}
+
+/**
+ * Type guard to check if a KVClient supports Redis Streams
+ */
+export function hasStreamSupport(client: IKVClient): client is IKVClientWithStreams {
+    return typeof (client as IKVClientWithStreams).xadd === 'function';
+}
+
 // =============================================================================
 // HashMeta Extensions
 // =============================================================================

packages/meta-sort-ui/src/components/ServiceNav.tsx

@@ -7,6 +7,7 @@ interface ServiceInfo {
   status: string;
   capabilities: string[];
   version: string;
+  role?: string; // "leader", "follower", or undefined (for non-meta-core services)
 }
 
 interface ServicesResponse {
@@ -55,7 +56,11 @@ function ServiceNav() {
 
   if (loading) return null;
 
-  const sortedServices = [...services].sort((a, b) => a.name.localeCompare(b.name));
+  // Filter out follower meta-core instances (only show leader)
+  const filteredServices = services.filter(
+    s => s.name !== 'meta-core' || s.role === 'leader'
+  );
+  const sortedServices = [...filteredServices].sort((a, b) => a.name.localeCompare(b.name));
 
   if (sortedServices.length === 0) return null;