cmd detector

cmd/plcbundle/commands/detector.go

···
 	"net/http"
 	_ "net/http/pprof"
 	"os"
+	"runtime"
 	"sort"
 	"strings"
-
-	flag "github.com/spf13/pflag"
+	"sync"
+	"sync/atomic"
 
 	"github.com/goccy/go-json"
+	"github.com/spf13/cobra"
+	"tangled.org/atscan.net/plcbundle/cmd/plcbundle/ui"
 	"tangled.org/atscan.net/plcbundle/detector"
 	"tangled.org/atscan.net/plcbundle/internal/plcclient"
 )
 
-// DetectorCommand handles the detector subcommand
-func DetectorCommand(args []string) error {
-	if len(args) < 1 {
-		printDetectorUsage()
-		return fmt.Errorf("subcommand required")
-	}
+func NewDetectorCommand() *cobra.Command {
+	cmd := &cobra.Command{
+		Use:     "detector",
+		Aliases: []string{"detect"},
+		Short:   "Detect and filter spam/invalid operations",
+		Long: `Detect and filter spam/invalid operations
 
-	subcommand := args[0]
+Run detection algorithms on PLC operations to identify spam, 
+invalid handles, service abuse, and other problematic patterns.
 
-	switch subcommand {
-	case "list":
-		return detectorList(args[1:])
-	case "test":
-		return detectorTest(args[1:])
-	case "run":
-		return detectorRun(args[1:])
-	case "filter":
-		return detectorFilter(args[1:])
-	case "info":
-		return detectorInfo(args[1:])
-	default:
-		printDetectorUsage()
-		return fmt.Errorf("unknown detector subcommand: %s", subcommand)
-	}
-}
+Built-in detectors:
+  • invalid_handle    - Invalid handle patterns (underscores, etc)
+  • aka_spam          - Excessive/garbage alsoKnownAs entries
+  • spam_pds          - Known spam PDS endpoints
+  • service_abuse     - Abused service structures
+  • noop              - Benchmark detector (returns no matches)
 
-func printDetectorUsage() {
-	fmt.Printf(`Usage: plcbundle detector <command> [options]
-
-Commands:
-  list          List available detectors
-  test          Test a detector on specific bundles
-  run           Run detector and output CSV results
-  filter        Filter JSONL operations from stdin
-  info          Show detailed detector information
+Custom detectors:
+  Load JavaScript detectors from .js files with a detect() function.`,
 
-Examples:
+		Example: `  # List available detectors
   plcbundle detector list
+
+  # Run detector on bundles
   plcbundle detector run invalid_handle --bundles 1-100
+
+  # Run with parallel processing
+  plcbundle detector run invalid_handle --bundles 1-100 --workers 8
+
+  # Run custom detector script
   plcbundle detector run ./my_detector.js --bundles 1-100
+
+  # Run multiple detectors
+  plcbundle detector run invalid_handle aka_spam --bundles 1-100
+
+  # Run all detectors
   plcbundle detector run all --bundles 1-100
-  plcbundle backfill | plcbundle detector filter ./my_detector.js > clean.jsonl
-`)
-}
 
-func detectorList(_ []string) error {
-	registry := detector.DefaultRegistry()
-	detectors := registry.List()
+  # Filter JSONL from stdin
+  cat ops.jsonl | plcbundle detector filter invalid_handle > clean.jsonl
 
-	sort.Slice(detectors, func(i, j int) bool {
-		return detectors[i].Name() < detectors[j].Name()
-	})
+  # Get detector info
+  plcbundle detector info invalid_handle`,
+	}
 
-	fmt.Printf("Available detectors:\n\n")
-	for _, d := range detectors {
-		fmt.Printf("  %-20s %s (v%s)\n", d.Name(), d.Description(), d.Version())
-	}
-	fmt.Printf("\nUse 'plcbundle detector info <name>' for details\n")
+	// Add subcommands
+	cmd.AddCommand(newDetectorListCommand())
+	cmd.AddCommand(newDetectorTestCommand())
+	cmd.AddCommand(newDetectorRunCommand())
+	cmd.AddCommand(newDetectorFilterCommand())
+	cmd.AddCommand(newDetectorInfoCommand())
 
-	return nil
+	return cmd
 }
 
-func detectorTest(args []string) error {
-	if len(args) < 1 {
-		return fmt.Errorf("usage: plcbundle detector test <detector-name> --bundle N")
-	}
+// ============================================================================
+// DETECTOR LIST
+// ============================================================================
+
+func newDetectorListCommand() *cobra.Command {
+	return &cobra.Command{
+		Use:   "list",
+		Short: "List available detectors",
+		Long:  `List all available built-in and loaded detectors`,
 
-	detectorName := args[0]
+		Example: `  # List all detectors
+  plcbundle detector list`,
 
-	fs := flag.NewFlagSet("detector test", flag.ExitOnError)
-	bundleNum := fs.Int("bundle", 0, "bundle number to test")
-	confidence := fs.Float64("confidence", 0.90, "minimum confidence threshold")
-	verbose := fs.Bool("v", false, "verbose output")
+		RunE: func(cmd *cobra.Command, args []string) error {
+			registry := detector.DefaultRegistry()
+			detectors := registry.List()
+
+			sort.Slice(detectors, func(i, j int) bool {
+				return detectors[i].Name() < detectors[j].Name()
+			})
+
+			fmt.Printf("Available detectors:\n\n")
+			for _, d := range detectors {
+				fmt.Printf("  %-20s %s (v%s)\n", d.Name(), d.Description(), d.Version())
+			}
+			fmt.Printf("\nUse 'plcbundle detector info <name>' for details\n")
 
-	if err := fs.Parse(args[1:]); err != nil {
-		return err
+			return nil
+		},
 	}
+}
 
-	if *bundleNum == 0 {
-		return fmt.Errorf("--bundle required")
+// ============================================================================
+// DETECTOR TEST
+// ============================================================================
+
+func newDetectorTestCommand() *cobra.Command {
+	var (
+		bundleNum  int
+		confidence float64
+		verbose    bool
+	)
+
+	cmd := &cobra.Command{
+		Use:   "test <detector-name>",
+		Short: "Test detector on specific bundle",
+		Long:  `Test a detector on a specific bundle and show results`,
+
+		Example: `  # Test on bundle 42
+  plcbundle detector test invalid_handle --bundle 42
+
+  # Verbose output with samples
+  plcbundle detector test aka_spam --bundle 100 -v
+
+  # Custom confidence threshold
+  plcbundle detector test spam_pds --bundle 50 --confidence 0.85`,
+
+		Args: cobra.ExactArgs(1),
+
+		RunE: func(cmd *cobra.Command, args []string) error {
+			detectorName := args[0]
+
+			mgr, _, err := getManager(&ManagerOptions{Cmd: cmd})
+			if err != nil {
+				return err
+			}
+			defer mgr.Close()
+
+			ctx := context.Background()
+			bundle, err := mgr.LoadBundle(ctx, bundleNum)
+			if err != nil {
+				return fmt.Errorf("failed to load bundle: %w", err)
+			}
+
+			fmt.Printf("Testing detector '%s' on bundle %06d...\n", detectorName, bundleNum)
+			fmt.Printf("Min confidence: %.2f\n\n", confidence)
+
+			registry := detector.DefaultRegistry()
+			config := detector.DefaultConfig()
+			config.MinConfidence = confidence
+
+			runner := detector.NewRunner(registry, config, &commandLogger{})
+			results, err := runner.RunOnBundle(ctx, detectorName, bundle)
+			if err != nil {
+				return fmt.Errorf("detection failed: %w", err)
+			}
+
+			stats := detector.CalculateStats(results, len(bundle.Operations))
+
+			fmt.Printf("Results:\n")
+			fmt.Printf("  Total operations:    %d\n", stats.TotalOperations)
+			fmt.Printf("  Matches found:       %d (%.2f%%)\n\n", stats.MatchedCount, stats.MatchRate*100)
+
+			if len(stats.ByReason) > 0 {
+				fmt.Printf("Breakdown by reason:\n")
+				for reason, count := range stats.ByReason {
+					pct := float64(count) / float64(stats.MatchedCount) * 100
+					fmt.Printf("  %-25s %d (%.1f%%)\n", reason, count, pct)
+				}
+				fmt.Printf("\n")
+			}
+
+			if verbose && len(results) > 0 {
+				fmt.Printf("Sample matches (first 10):\n")
+				displayCount := min(10, len(results))
+
+				for i := 0; i < displayCount; i++ {
+					res := results[i]
+					globalPos := (res.BundleNumber * 10000) + res.Position
+					fmt.Printf("  %d. Position %d: %s\n", i+1, globalPos, res.DID)
+					fmt.Printf("     Reason: %s (confidence: %.2f)\n", res.Match.Reason, res.Match.Confidence)
+					if res.Match.Note != "" {
+						fmt.Printf("     Note: %s\n", res.Match.Note)
+					}
+				}
+			}
+
+			return nil
+		},
 	}
 
-	mgr, _, err := getManager(nil)
-	if err != nil {
-		return err
+	cmd.Flags().IntVar(&bundleNum, "bundle", 0, "Bundle number to test (required)")
+	cmd.Flags().Float64Var(&confidence, "confidence", 0.90, "Minimum confidence threshold")
+	cmd.Flags().BoolVarP(&verbose, "verbose", "v", false, "Show sample matches")
+
+	cmd.MarkFlagRequired("bundle")
+
+	return cmd
+}
+
+// ============================================================================
+// DETECTOR RUN (with parallel processing)
+// ============================================================================
+
+func newDetectorRunCommand() *cobra.Command {
+	var (
+		bundleRange string
+		confidence  float64
+		pprofPort   string
+		workers     int
+		noProgress  bool
+	)
+
+	cmd := &cobra.Command{
+		Use:   "run <detector1|script.js> [detector2...] [flags]",
+		Short: "Run detector(s) and output CSV results",
+		Long: `Run one or more detectors on bundles and output results as CSV
+
+Output format uses global position: (bundleNumber × 10,000) + position
+Example: 88410345 = bundle 8841, position 345
+
+Supports parallel processing across multiple workers for better performance.
+
+Detectors can be:
+  • Built-in detector names (invalid_handle, aka_spam, etc)
+  • Path to JavaScript detector file (./my_detector.js)
+  • Special keyword 'all' to run all built-in detectors`,
+
+		Example: `  # Run single detector
+  plcbundle detector run invalid_handle --bundles 1-100
+
+  # Run with 8 parallel workers (faster)
+  plcbundle detector run invalid_handle --bundles 1-1000 --workers 8
+
+  # Run multiple detectors in parallel
+  plcbundle detector run invalid_handle aka_spam --bundles 1-100 -w 4
+
+  # Run custom script
+  plcbundle detector run ./my_detector.js --bundles 1-100
+
+  # Run all built-in detectors
+  plcbundle detector run all --bundles 1-100 --workers 8
+
+  # Save results to file
+  plcbundle detector run all --bundles 1-100 -w 8 > results.csv
+
+  # Disable progress bar (for scripting)
+  plcbundle detector run spam --bundles 1-100 --no-progress
+
+  # Enable profiling
+  plcbundle detector run all --bundles 1-100 --pprof :6060`,
+
+		Args: cobra.MinimumNArgs(1),
+
+		RunE: func(cmd *cobra.Command, args []string) error {
+			detectorNames := args
+
+			// Start pprof if requested
+			if pprofPort != "" {
+				go func() {
+					fmt.Fprintf(os.Stderr, "pprof server starting on http://localhost%s/debug/pprof/\n", pprofPort)
+					http.ListenAndServe(pprofPort, nil)
+				}()
+			}
+
+			// Auto-detect workers if not specified
+			if workers <= 0 {
+				workers = runtime.NumCPU()
+				if workers < 1 {
+					workers = 1
+				}
+			}
+
+			mgr, _, err := getManager(&ManagerOptions{Cmd: cmd})
+			if err != nil {
+				return err
+			}
+			defer mgr.Close()
+
+			// Determine range
+			var start, end int
+			if bundleRange == "" {
+				index := mgr.GetIndex()
+				bundles := index.GetBundles()
+				if len(bundles) == 0 {
+					return fmt.Errorf("no bundles available")
+				}
+				start = bundles[0].BundleNumber
+				end = bundles[len(bundles)-1].BundleNumber
+				fmt.Fprintf(os.Stderr, "Using all bundles: %d-%d\n", start, end)
+			} else {
+				start, end, err = parseBundleRange(bundleRange)
+				if err != nil {
+					return err
+				}
+			}
+
+			// Load detectors
+			setup, err := parseAndLoadDetectors(detectorNames, confidence)
+			if err != nil {
+				return err
+			}
+			defer setup.cleanup()
+
+			fmt.Fprintf(os.Stderr, "Running %d detector(s) on bundles %d-%d (%d workers)\n",
+				len(setup.detectors), start, end, workers)
+			fmt.Fprintf(os.Stderr, "Min confidence: %.2f\n\n", confidence)
+
+			return runDetectionParallel(cmd.Context(), mgr, setup, start, end, workers, !noProgress)
+		},
 	}
-	defer mgr.Close()
 
-	ctx := context.Background()
-	bundle, err := mgr.LoadBundle(ctx, *bundleNum)
-	if err != nil {
-		return fmt.Errorf("error loading bundle: %w", err)
+	cmd.Flags().StringVar(&bundleRange, "bundles", "", "Bundle range (e.g., '1-100', default: all)")
+	cmd.Flags().Float64Var(&confidence, "confidence", 0.90, "Minimum confidence threshold")
+	cmd.Flags().IntVarP(&workers, "workers", "w", 0, "Number of parallel workers (0 = auto-detect CPU count)")
+	cmd.Flags().BoolVar(&noProgress, "no-progress", false, "Disable progress bar")
+	cmd.Flags().StringVar(&pprofPort, "pprof", "", "Enable pprof on port (e.g., ':6060')")
+
+	return cmd
+}
+
+// ============================================================================
+// DETECTOR FILTER
+// ============================================================================
+
+func newDetectorFilterCommand() *cobra.Command {
+	var confidence float64
+
+	cmd := &cobra.Command{
+		Use:   "filter <detector1|script.js> [detector2...]",
+		Short: "Filter JSONL operations from stdin",
+		Long: `Filter JSONL operations from stdin using detectors
+
+Reads operations from stdin, runs detectors, and outputs only
+operations that DO NOT match (clean operations).
+
+Perfect for cleaning datasets or pre-processing.`,
+
+		Example: `  # Filter with built-in detector
+  cat ops.jsonl | plcbundle detector filter invalid_handle > clean.jsonl
+
+  # Filter with custom script
+  plcbundle export --all | plcbundle detector filter ./spam.js > clean.jsonl
+
+  # Chain multiple detectors
+  cat ops.jsonl | plcbundle detector filter invalid_handle aka_spam > clean.jsonl
+
+  # Custom confidence
+  cat ops.jsonl | plcbundle detector filter spam_pds --confidence 0.95 > clean.jsonl`,
+
+		Args: cobra.MinimumNArgs(1),
+
+		RunE: func(cmd *cobra.Command, args []string) error {
+			detectorNames := args
+
+			setup, err := parseAndLoadDetectors(detectorNames, confidence)
+			if err != nil {
+				return err
+			}
+			defer setup.cleanup()
+
+			fmt.Fprintf(os.Stderr, "Filtering with %d detector(s), min confidence: %.2f\n\n", len(setup.detectors), confidence)
+
+			return filterFromStdin(cmd.Context(), setup)
+		},
 	}
 
-	fmt.Printf("Testing detector '%s' on bundle %06d...\n", detectorName, *bundleNum)
-	fmt.Printf("Min confidence: %.2f\n\n", *confidence)
+	cmd.Flags().Float64Var(&confidence, "confidence", 0.90, "Minimum confidence threshold")
 
-	registry := detector.DefaultRegistry()
-	config := detector.DefaultConfig()
-	config.MinConfidence = *confidence
+	return cmd
+}
 
-	runner := detector.NewRunner(registry, config, &defaultLogger{})
-	results, err := runner.RunOnBundle(ctx, detectorName, bundle)
-	if err != nil {
-		return fmt.Errorf("detection failed: %w", err)
-	}
+// ============================================================================
+// DETECTOR INFO
+// ============================================================================
 
-	stats := detector.CalculateStats(results, len(bundle.Operations))
+func newDetectorInfoCommand() *cobra.Command {
+	return &cobra.Command{
+		Use:   "info <detector-name>",
+		Short: "Show detailed detector information",
+		Long:  `Show detailed information about a specific detector`,
 
-	fmt.Printf("Results:\n")
-	fmt.Printf("  Total operations:    %d\n", stats.TotalOperations)
-	fmt.Printf("  Matches found:       %d (%.2f%%)\n\n", stats.MatchedCount, stats.MatchRate*100)
+		Example: `  # Show detector info
+  plcbundle detector info invalid_handle
+  plcbundle detector info aka_spam`,
 
-	if len(stats.ByReason) > 0 {
-		fmt.Printf("Breakdown by reason:\n")
-		for reason, count := range stats.ByReason {
-			pct := float64(count) / float64(stats.MatchedCount) * 100
-			fmt.Printf("  %-25s %d (%.1f%%)\n", reason, count, pct)
-		}
-		fmt.Printf("\n")
-	}
+		Args: cobra.ExactArgs(1),
 
-	if *verbose && len(results) > 0 {
-		fmt.Printf("Sample matches (first 10):\n")
-		displayCount := 10
-		if len(results) < displayCount {
-			displayCount = len(results)
-		}
+		RunE: func(cmd *cobra.Command, args []string) error {
+			detectorName := args[0]
 
-		for i := 0; i < displayCount; i++ {
-			res := results[i]
-			fmt.Printf("  %d. Position %d: %s\n", i+1, res.Position, res.DID)
-			fmt.Printf("     Reason: %s (confidence: %.2f)\n", res.Match.Reason, res.Match.Confidence)
-			if res.Match.Note != "" {
-				fmt.Printf("     Note: %s\n", res.Match.Note)
+			registry := detector.DefaultRegistry()
+			d, err := registry.Get(detectorName)
+			if err != nil {
+				return err
 			}
-		}
+
+			fmt.Printf("Detector: %s\n", d.Name())
+			fmt.Printf("Version: %s\n", d.Version())
+			fmt.Printf("Description: %s\n\n", d.Description())
+
+			fmt.Printf("Usage examples:\n")
+			fmt.Printf("  # Test on single bundle\n")
+			fmt.Printf("  plcbundle detector test %s --bundle 42\n\n", d.Name())
+			fmt.Printf("  # Run on range and save\n")
+			fmt.Printf("  plcbundle detector run %s --bundles 1-100 > results.csv\n\n", d.Name())
+			fmt.Printf("  # Filter JSONL stream\n")
+			fmt.Printf("  cat ops.jsonl | plcbundle detector filter %s > clean.jsonl\n\n", d.Name())
+
+			return nil
+		},
 	}
+}
+
+// ============================================================================
+// PARALLEL DETECTION IMPLEMENTATION
+// ============================================================================
 
-	return nil
+type detectionMatch struct {
+	globalPos  int
+	cid        string
+	size       int
+	confidence float64
+	labels     []string
 }
 
-func detectorRun(args []string) error {
-	if len(args) < 1 {
-		return fmt.Errorf("usage: plcbundle detector run <detector1|script.js> [detector2...] [--bundles 1-100]")
-	}
+// detectionResult holds results from processing a bundle
+type detectionResult struct {
+	bundleNum    int
+	matches      []detectionMatch
+	totalOps     int
+	totalBytes   int64
+	matchedBytes int64
+	err          error
+}
 
-	var detectorNames []string
-	var flagArgs []string
-	for i := 0; i < len(args); i++ {
-		if strings.HasPrefix(args[i], "-") {
-			flagArgs = args[i:]
-			break
-		}
-		detectorNames = append(detectorNames, args[i])
-	}
+func runDetectionParallel(ctx context.Context, mgr BundleManager, setup *detectorSetup, start, end int, workers int, showProgress bool) error {
+	totalBundles := end - start + 1
 
-	if len(detectorNames) == 0 {
-		return fmt.Errorf("at least one detector name required")
+	// ✨ FIX: Don't create more workers than bundles
+	if workers > totalBundles {
+		workers = totalBundles
 	}
 
-	fs := flag.NewFlagSet("detector run", flag.ExitOnError)
-	bundleRange := fs.String("bundles", "", "bundle range (default: all)")
-	confidence := fs.Float64("confidence", 0.90, "minimum confidence")
-	pprofPort := fs.String("pprof", "", "enable pprof on port (e.g., :6060)")
+	// ✨ FIX: Use unbuffered channels to avoid blocking issues
+	jobs := make(chan int, workers*2) // Small buffer for job numbers only
+	results := make(chan detectionResult, workers*2)
 
-	if err := fs.Parse(flagArgs); err != nil {
-		return err
+	// Shared counters
+	var (
+		totalOps     int64
+		matchCount   int64
+		totalBytes   int64
+		matchedBytes int64
+	)
+
+	// Progress tracking
+	var progress *ui.ProgressBar
+	if showProgress {
+		progress = ui.NewProgressBar(totalBundles)
 	}
 
-	// Start pprof if requested
-	if *pprofPort != "" {
-		go func() {
-			fmt.Fprintf(os.Stderr, "pprof server starting on http://localhost%s/debug/pprof/\n", *pprofPort)
-			http.ListenAndServe(*pprofPort, nil)
-		}()
-	}
+	// CSV header
+	fmt.Println("position,cid,size,confidence,labels")
+
+	// Start workers BEFORE sending jobs
+	var wg sync.WaitGroup
+	for w := 0; w < workers; w++ {
+		wg.Add(1)
+		go func(workerID int) {
+			defer wg.Done()
 
-	mgr, _, err := getManager(nil)
-	if err != nil {
-		return err
+			for bundleNum := range jobs {
+				select {
+				case <-ctx.Done():
+					return
+				default:
+				}
+
+				// Process bundle
+				res := processBundleForDetection(ctx, mgr, setup, bundleNum)
+
+				// Send result (may block if collector is slow, but that's OK)
+				select {
+				case results <- res:
+				case <-ctx.Done():
+					return
+				}
+			}
+		}(w)
 	}
-	defer mgr.Close()
 
-	// Determine range
-	var start, end int
-	if *bundleRange == "" {
-		index := mgr.GetIndex()
-		bundles := index.GetBundles()
-		if len(bundles) == 0 {
-			return fmt.Errorf("no bundles available")
+	// Send jobs in separate goroutine
+	go func() {
+		defer close(jobs)
+		for bundleNum := start; bundleNum <= end; bundleNum++ {
+			select {
+			case jobs <- bundleNum:
+			case <-ctx.Done():
+				return
+			}
 		}
-		start = bundles[0].BundleNumber
-		end = bundles[len(bundles)-1].BundleNumber
-		fmt.Fprintf(os.Stderr, "Using all bundles: %d-%d\n", start, end)
-	} else {
-		start, end, err = parseBundleRange(*bundleRange)
-		if err != nil {
-			return err
-		}
-	}
+	}()
+
+	// Collect results in separate goroutine
+	done := make(chan struct{})
+	go func() {
+		defer close(done)
+		wg.Wait()
+		close(results)
+	}()
 
-	// Load detectors
-	setup, err := parseAndLoadDetectors(detectorNames, *confidence)
-	if err != nil {
-		return err
-	}
-	defer setup.cleanup()
+	// Process results in main goroutine
+	processed := 0
+	allMatches := make([]detectionMatch, 0, 1000)
 
-	fmt.Fprintf(os.Stderr, "Running %d detector(s) on bundles %d-%d\n", len(setup.detectors), start, end)
-	fmt.Fprintf(os.Stderr, "Min confidence: %.2f\n\n", *confidence)
+	for res := range results {
+		processed++
 
-	ctx := context.Background()
-	fmt.Println("bundle,position,cid,size,confidence,labels")
+		if res.err != nil {
+			if res.err != context.Canceled {
+				fmt.Fprintf(os.Stderr, "\nWarning: bundle %06d failed: %v\n", res.bundleNum, res.err)
+			}
+		} else {
+			// Update counters
+			atomic.AddInt64(&totalOps, int64(res.totalOps))
+			atomic.AddInt64(&matchCount, int64(len(res.matches)))
+			atomic.AddInt64(&totalBytes, res.totalBytes)
+			atomic.AddInt64(&matchedBytes, res.matchedBytes)
 
-	totalOps, matchCount := 0, 0
-	totalBytes, matchedBytes := int64(0), int64(0)
+			// Collect matches
+			allMatches = append(allMatches, res.matches...)
 
-	for bundleNum := start; bundleNum <= end; bundleNum++ {
-		bundle, err := mgr.LoadBundle(ctx, bundleNum)
-		if err != nil {
-			continue
+			// ✨ FIX: Output immediately (don't buffer too much)
+			if len(allMatches) >= 500 {
+				for _, match := range allMatches {
+					fmt.Printf("%d,%s,%d,%.2f,%s\n",
+						match.globalPos, match.cid, match.size,
+						match.confidence, strings.Join(match.labels, ";"))
+				}
+				allMatches = allMatches[:0]
+			}
 		}
 
-		totalOps += len(bundle.Operations)
+		if progress != nil {
+			progress.Set(processed)
+		}
+	}
 
-		for position, op := range bundle.Operations {
-			opSize := len(op.RawJSON)
-			if opSize == 0 {
-				data, _ := json.Marshal(op)
-				opSize = len(data)
-			}
-			totalBytes += int64(opSize)
+	// Flush remaining matches
+	for _, match := range allMatches {
+		fmt.Printf("%d,%s,%d,%.2f,%s\n",
+			match.globalPos, match.cid, match.size,
+			match.confidence, strings.Join(match.labels, ";"))
+	}
 
-			labels, conf := detectOperation(ctx, setup.detectors, op, setup.confidence)
+	if progress != nil {
+		progress.Finish()
+	}
 
-			if len(labels) > 0 {
-				matchCount++
-				matchedBytes += int64(opSize)
+	// Wait for cleanup
+	<-done
 
-				cidShort := op.CID
-				if len(cidShort) > 4 {
-					cidShort = cidShort[len(cidShort)-4:]
-				}
+	// Summary
+	finalTotalOps := atomic.LoadInt64(&totalOps)
+	finalMatchCount := atomic.LoadInt64(&matchCount)
+	finalTotalBytes := atomic.LoadInt64(&totalBytes)
+	finalMatchedBytes := atomic.LoadInt64(&matchedBytes)
 
-				fmt.Printf("%d,%d,%s,%d,%.2f,%s\n",
-					bundleNum, position, cidShort, opSize, conf, strings.Join(labels, ";"))
-			}
-		}
+	if finalTotalOps == 0 {
+		fmt.Fprintf(os.Stderr, "\n⚠️  No operations processed\n")
+		return nil
 	}
 
 	fmt.Fprintf(os.Stderr, "\n✓ Detection complete\n")
-	fmt.Fprintf(os.Stderr, "  Total operations:   %d\n", totalOps)
-	fmt.Fprintf(os.Stderr, "  Matches found:      %d (%.2f%%)\n", matchCount, float64(matchCount)/float64(totalOps)*100)
+	fmt.Fprintf(os.Stderr, "  Total operations:   %d\n", finalTotalOps)
+	fmt.Fprintf(os.Stderr, "  Matches found:      %d (%.2f%%)\n",
+		finalMatchCount, float64(finalMatchCount)/float64(finalTotalOps)*100)
+	fmt.Fprintf(os.Stderr, "  Total size:         %s\n", formatBytes(finalTotalBytes))
+	fmt.Fprintf(os.Stderr, "  Matched size:       %s (%.2f%%)\n",
+		formatBytes(finalMatchedBytes),
+		float64(finalMatchedBytes)/float64(finalTotalBytes)*100)
 
 	return nil
 }
 
-func detectorFilter(args []string) error {
-	if len(args) < 1 {
-		return fmt.Errorf("usage: plcbundle detector filter <detector1|script.js> [detector2...] [--confidence 0.9]")
+func processBundleForDetection(ctx context.Context, mgr BundleManager, setup *detectorSetup, bundleNum int) detectionResult {
+	res := detectionResult{bundleNum: bundleNum}
+
+	bundle, err := mgr.LoadBundle(ctx, bundleNum)
+	if err != nil {
+		res.err = err
+		return res
 	}
 
-	var detectorNames []string
-	var flagArgs []string
-	for i := 0; i < len(args); i++ {
-		if strings.HasPrefix(args[i], "-") {
-			flagArgs = args[i:]
-			break
+	res.totalOps = len(bundle.Operations)
+	matches := make([]detectionMatch, 0)
+
+	for position, op := range bundle.Operations {
+		// Calculate global position
+		globalPos := (bundleNum * 10000) + position
+
+		opSize := len(op.RawJSON)
+		if opSize == 0 {
+			data, _ := json.Marshal(op)
+			opSize = len(data)
 		}
-		detectorNames = append(detectorNames, args[i])
-	}
+		res.totalBytes += int64(opSize)
+
+		labels, conf := detectOperation(ctx, setup.detectors, op, setup.confidence)
 
-	if len(detectorNames) == 0 {
-		return fmt.Errorf("at least one detector name required")
-	}
+		if len(labels) > 0 {
+			res.matchedBytes += int64(opSize)
 
-	fs := flag.NewFlagSet("detector filter", flag.ExitOnError)
-	confidence := fs.Float64("confidence", 0.90, "minimum confidence")
+			cidShort := op.CID
+			if len(cidShort) > 4 {
+				cidShort = cidShort[len(cidShort)-4:]
+			}
 
-	if err := fs.Parse(flagArgs); err != nil {
-		return err
+			matches = append(matches, detectionMatch{
+				globalPos:  globalPos,
+				cid:        cidShort,
+				size:       opSize,
+				confidence: conf,
+				labels:     labels,
+			})
+		}
 	}
 
-	setup, err := parseAndLoadDetectors(detectorNames, *confidence)
-	if err != nil {
-		return err
-	}
-	defer setup.cleanup()
+	res.matches = matches
+	return res
+}
 
-	fmt.Fprintf(os.Stderr, "Filtering with %d detector(s), min confidence: %.2f\n\n", len(setup.detectors), *confidence)
+// ============================================================================
+// FILTER FROM STDIN
+// ============================================================================
 
-	ctx := context.Background()
+func filterFromStdin(ctx context.Context, setup *detectorSetup) error {
 	scanner := bufio.NewScanner(os.Stdin)
 	buf := make([]byte, 0, 64*1024)
 	scanner.Buffer(buf, 1024*1024)
···
 		labels, _ := detectOperation(ctx, setup.detectors, op, setup.confidence)
 
 		if len(labels) == 0 {
+			// Clean - output to stdout
 			cleanCount++
 			fmt.Println(string(line))
 		} else {
+			// Matched - filter out
 			filteredCount++
 			filteredBytes += int64(len(line))
 		}
···
 	return nil
 }
 
-func detectorInfo(args []string) error {
-	if len(args) < 1 {
-		return fmt.Errorf("usage: plcbundle detector info <name>")
-	}
-
-	detectorName := args[0]
-
-	registry := detector.DefaultRegistry()
-	d, err := registry.Get(detectorName)
-	if err != nil {
-		return err
-	}
-
-	fmt.Printf("Detector: %s\n", d.Name())
-	fmt.Printf("Version: %s\n", d.Version())
-	fmt.Printf("Description: %s\n\n", d.Description())
-
-	fmt.Printf("Usage examples:\n")
-	fmt.Printf("  # Test on single bundle\n")
-	fmt.Printf("  plcbundle detector test %s --bundle 42\n\n", d.Name())
-	fmt.Printf("  # Run on range and save\n")
-	fmt.Printf("  plcbundle detector run %s --bundles 1-100 > results.csv\n\n", d.Name())
-
-	return nil
-}
-
-// Helper functions
+// ============================================================================
+// SHARED HELPERS
+// ============================================================================
 
 type detectorSetup struct {
 	detectors       []detector.Detector
···
 func parseAndLoadDetectors(detectorNames []string, confidence float64) (*detectorSetup, error) {
 	registry := detector.DefaultRegistry()
 
+	// Handle "all" keyword
 	if len(detectorNames) == 1 && detectorNames[0] == "all" {
 		detectorNames = registry.Names()
 		fmt.Fprintf(os.Stderr, "Using all detectors: %s\n", strings.Join(detectorNames, ", "))
···
 	}
 
 	for _, name := range detectorNames {
+		// JavaScript detector
 		if strings.HasSuffix(name, ".js") {
 			sd, err := detector.NewScriptDetector(name)
 			if err != nil {
 				setup.cleanup()
-				return nil, fmt.Errorf("error loading script %s: %w", name, err)
+				return nil, fmt.Errorf("failed to load script %s: %w", name, err)
 			}
 			setup.scriptDetectors = append(setup.scriptDetectors, sd)
 			registry.Register(sd)
 			setup.detectors = append(setup.detectors, sd)
-			fmt.Fprintf(os.Stderr, "✓ Started detector server: %s\n", sd.Name())
+			fmt.Fprintf(os.Stderr, "✓ Loaded detector: %s\n", sd.Name())
 		} else {
+			// Built-in detector
 			d, err := registry.Get(name)
 			if err != nil {
 				setup.cleanup()
···
 			continue
 		}
 
+		// Extract labels
 		var labels []string
 		if labelList, ok := match.Metadata["labels"].([]string); ok {
 			labels = labelList
···
 
 	return matchedLabels, maxConfidence
 }
-
-type defaultLogger struct{}
-
-func (d *defaultLogger) Printf(format string, v ...interface{}) {
-	fmt.Fprintf(os.Stderr, format+"\n", v...)
-}

cmd/plcbundle/commands/rebuild.go

···
-package commands
-
-import (
-	"context"
-	"fmt"
-	"os"
-	"path/filepath"
-	"runtime"
-	"time"
-
-	flag "github.com/spf13/pflag"
-
-	"tangled.org/atscan.net/plcbundle/bundle"
-	"tangled.org/atscan.net/plcbundle/cmd/plcbundle/ui"
-	"tangled.org/atscan.net/plcbundle/internal/bundleindex"
-)
-
-// RebuildCommand handles the rebuild subcommand
-func RebuildCommand(args []string) error {
-	fs := flag.NewFlagSet("rebuild", flag.ExitOnError)
-	verbose := fs.Bool("v", false, "verbose output")
-	workers := fs.Int("workers", 0, "number of parallel workers (0 = CPU count)")
-	noProgress := fs.Bool("no-progress", false, "disable progress bar")
-
-	if err := fs.Parse(args); err != nil {
-		return err
-	}
-
-	// Auto-detect CPU count
-	if *workers == 0 {
-		*workers = runtime.NumCPU()
-	}
-
-	// Get working directory
-	dir, err := os.Getwd()
-	if err != nil {
-		return err
-	}
-
-	if err := os.MkdirAll(dir, 0755); err != nil {
-		return err
-	}
-
-	// Create manager WITHOUT auto-rebuild
-	config := bundle.DefaultConfig(dir)
-	config.AutoRebuild = false
-	config.RebuildWorkers = *workers
-
-	mgr, err := bundle.NewManager(config, nil)
-	if err != nil {
-		return err
-	}
-	defer mgr.Close()
-
-	fmt.Printf("Rebuilding index from: %s\n", dir)
-	fmt.Printf("Using %d workers\n", *workers)
-
-	// Find all bundle files
-	files, err := filepath.Glob(filepath.Join(dir, "*.jsonl.zst"))
-	if err != nil {
-		return fmt.Errorf("error scanning directory: %w", err)
-	}
-
-	// Filter out hidden/temp files
-	files = filterBundleFiles(files)
-
-	if len(files) == 0 {
-		fmt.Println("No bundle files found")
-		return nil
-	}
-
-	fmt.Printf("Found %d bundle files\n\n", len(files))
-
-	start := time.Now()
-
-	// Create progress bar
-	var progress *ui.ProgressBar
-	var progressCallback func(int, int, int64)
-
-	if !*noProgress {
-		fmt.Println("Processing bundles:")
-		progress = ui.NewProgressBar(len(files))
-
-		progressCallback = func(current, total int, bytesProcessed int64) {
-			progress.SetWithBytes(current, bytesProcessed)
-		}
-	}
-
-	// Use parallel scan
-	result, err := mgr.ScanDirectoryParallel(*workers, progressCallback)
-
-	if err != nil {
-		if progress != nil {
-			progress.Finish()
-		}
-		return fmt.Errorf("rebuild failed: %w", err)
-	}
-
-	if progress != nil {
-		progress.Finish()
-	}
-
-	elapsed := time.Since(start)
-
-	fmt.Printf("\n✓ Index rebuilt in %s\n", elapsed.Round(time.Millisecond))
-	fmt.Printf("  Total bundles:      %d\n", result.BundleCount)
-	fmt.Printf("  Compressed size:    %s\n", formatBytes(result.TotalSize))
-	fmt.Printf("  Uncompressed size:  %s\n", formatBytes(result.TotalUncompressed))
-
-	if result.TotalUncompressed > 0 {
-		ratio := float64(result.TotalUncompressed) / float64(result.TotalSize)
-		fmt.Printf("  Compression ratio:  %.2fx\n", ratio)
-	}
-
-	fmt.Printf("  Average speed:      %.1f bundles/sec\n", float64(result.BundleCount)/elapsed.Seconds())
-
-	if elapsed.Seconds() > 0 {
-		compressedThroughput := float64(result.TotalSize) / elapsed.Seconds() / (1000 * 1000)
-		uncompressedThroughput := float64(result.TotalUncompressed) / elapsed.Seconds() / (1000 * 1000)
-		fmt.Printf("  Throughput (compressed):   %.1f MB/s\n", compressedThroughput)
-		fmt.Printf("  Throughput (uncompressed): %.1f MB/s\n", uncompressedThroughput)
-	}
-
-	fmt.Printf("  Index file:         %s\n", filepath.Join(dir, bundleindex.INDEX_FILE))
-
-	if len(result.MissingGaps) > 0 {
-		fmt.Printf("  ⚠️  Missing gaps:     %d bundles\n", len(result.MissingGaps))
-	}
-
-	// Verify chain if verbose
-	if *verbose {
-		fmt.Printf("\nVerifying chain integrity...\n")
-
-		ctx := context.Background()
-		verifyResult, err := mgr.VerifyChain(ctx)
-		if err != nil {
-			fmt.Printf("  ⚠️  Verification error: %v\n", err)
-		} else if verifyResult.Valid {
-			fmt.Printf("  ✓ Chain is valid (%d bundles verified)\n", len(verifyResult.VerifiedBundles))
-
-			// Show head hash
-			index := mgr.GetIndex()
-			if lastMeta := index.GetLastBundle(); lastMeta != nil {
-				fmt.Printf("  Chain head: %s...\n", lastMeta.Hash[:16])
-			}
-		} else {
-			fmt.Printf("  ✗ Chain verification failed\n")
-			fmt.Printf("  Broken at: bundle %06d\n", verifyResult.BrokenAt)
-			fmt.Printf("  Error: %s\n", verifyResult.Error)
-		}
-	}
-
-	return nil
-}
-
-func filterBundleFiles(files []string) []string {
-	filtered := make([]string, 0, len(files))
-	for _, file := range files {
-		basename := filepath.Base(file)
-		if len(basename) > 0 && (basename[0] == '.' || basename[0] == '_') {
-			continue
-		}
-		filtered = append(filtered, file)
-	}
-	return filtered
-}

cmd/plcbundle/main.go

···
 	cmd.AddCommand(commands.NewDIDCommand())
 	cmd.AddCommand(commands.NewIndexCommand())
 	cmd.AddCommand(commands.NewMempoolCommand())
-	/*cmd.AddCommand(commands.NewDetectorCommand())
+	cmd.AddCommand(commands.NewDetectorCommand())
 
 	// Monitoring & maintenance
-	cmd.AddCommand(commands.NewWatchCommand())
+	/*cmd.AddCommand(commands.NewWatchCommand())
 	cmd.AddCommand(commands.NewHealCommand())
 	cmd.AddCommand(commands.NewCleanCommand())*/
 

detector/script.go

···
 	"os/exec"
 	"path/filepath"
 	"strings"
+	"sync"
 	"time"
 
 	"github.com/goccy/go-json"
···
 	conn       net.Conn
 	writer     *bufio.Writer
 	reader     *bufio.Reader
+	mu         sync.Mutex
 }
 
 func NewScriptDetector(scriptPath string) (*ScriptDetector, error) {
···
 	if d.conn == nil {
 		return nil, fmt.Errorf("not connected to server")
 	}
+
+	// ✨ LOCK for entire socket communication
+	d.mu.Lock()
+	defer d.mu.Unlock()
 
 	// Use RawJSON directly
 	data := op.RawJSON

go.mod

···
 
 require (
 	github.com/spf13/cobra v1.10.1
-	github.com/spf13/pflag v1.0.10
 	golang.org/x/term v0.36.0
 )
 
-require github.com/inconshreveable/mousetrap v1.1.0 // indirect
+require (
+	github.com/inconshreveable/mousetrap v1.1.0 // indirect
+	github.com/spf13/pflag v1.0.9 // indirect
+)

go.sum

···
 github.com/russross/blackfriday/v2 v2.1.0/go.mod h1:+Rmxgy9KzJVeS9/2gXHxylqXiyQDYRxCVz55jmeOWTM=
 github.com/spf13/cobra v1.10.1 h1:lJeBwCfmrnXthfAupyUTzJ/J4Nc1RsHC/mSRU2dll/s=
 github.com/spf13/cobra v1.10.1/go.mod h1:7SmJGaTHFVBY0jW4NXGluQoLvhqFQM+6XSKD+P4XaB0=
+github.com/spf13/pflag v1.0.9 h1:9exaQaMOCwffKiiiYk6/BndUBv+iRViNW+4lEMi0PvY=
 github.com/spf13/pflag v1.0.9/go.mod h1:McXfInJRrz4CZXVZOBLb0bTZqETkiAhM9Iw0y3An2Bg=
-github.com/spf13/pflag v1.0.10 h1:4EBh2KAYBwaONj6b2Ye1GiHfwjqyROoF4RwYO+vPwFk=
-github.com/spf13/pflag v1.0.10/go.mod h1:McXfInJRrz4CZXVZOBLb0bTZqETkiAhM9Iw0y3An2Bg=
 golang.org/x/sys v0.37.0 h1:fdNQudmxPjkdUTPnLn5mdQv7Zwvbvpaxqs831goi9kQ=
 golang.org/x/sys v0.37.0/go.mod h1:OgkHotnGiDImocRcuBABYBEXf8A9a87e/uXjp9XT3ks=
 golang.org/x/term v0.36.0 h1:zMPR+aF8gfksFprF/Nc/rd1wRS1EI6nDBGyWAvDzx2Q=