fix(rpc): delegate REM pipelines from generic Start/Stop/DeletePipeline to dedicated handlers

StartPipeline/StopPipeline/DeletePipeline did not handle REM pipeline
type correctly — REM was falling through to the default CustomPipeline
path in the listener's startPipeline(), so the rem process was never
actually started. WebUI always calls the generic StartPipeline RPC,
which caused REM pipelines created via WebUI to have empty Link and
Subscribe fields.

Add type checks in StartPipeline, StopPipeline, and DeletePipeline to
detect REM pipelines and delegate to StartRem, StopRem, DeleteRem
respectively.

Also fix two related issues:
- Remove stale core.Jobs.AddPipeline() call in StartRem that overwrote
  runtime Link/Subscribe/Port data from SyncPipeline with the old DB
  snapshot.
- Enhance handleStartRem idempotency check to detect dead REM pipelines
  (crashed via runtimeErrorHandler) and recreate them instead of just
  syncing stale state.

Author: h3zh1

server/listener/listener.go

@@ -679,8 +679,15 @@ func (lns *listener) handleStartRem(job *clientpb.Job) error {
 
 	// Idempotency: REM already started in this listener process.
 	if existing := lns.pipelines.Get(pipe.Name); existing != nil {
-		_, err := lns.Rpc.SyncPipeline(lns.Context(), existing.ToProtobuf())
-		return err
+		remPipeline, ok := existing.(*REM)
+		if ok && remPipeline.Enable {
+			// Still healthy — just sync its current state.
+			_, err := lns.Rpc.SyncPipeline(lns.Context(), existing.ToProtobuf())
+			return err
+		}
+		// Dead pipeline (crashed via runtimeErrorHandler) — remove the stale
+		// entry so we can create a fresh one below.
+		lns.pipelines.Delete(existing.ID())
 	}
 
 	rem, err := NewRem(lns.Rpc, pipe)

server/rpc/rpc-pipeline.go

@@ -111,6 +111,14 @@ func (rpc *Server) StartPipeline(ctx context.Context, req *clientpb.CtrlPipeline
 	if err != nil {
 		return nil, err
 	}
+
+	// REM pipelines have their own start path; delegate transparently so
+	// callers (e.g. WebUI) that always use StartPipeline still work.
+	if pipelineDB.Type == consts.RemPipeline {
+		req.ListenerId = listenerID
+		return rpc.StartRem(ctx, req)
+	}
+
 	if pipelineDB.PipelineParams == nil {
 		pipelineDB.PipelineParams = &implanttypes.PipelineParams{}
 	}
@@ -169,6 +177,17 @@ func (rpc *Server) StopPipeline(ctx context.Context, req *clientpb.CtrlPipeline)
 		return nil, err
 	}
 
+	pipelineDB, err := db.FindPipelineByListener(req.Name, listenerID)
+	if err != nil {
+		return nil, err
+	}
+
+	// Delegate REM pipelines to their dedicated handler.
+	if pipelineDB.Type == consts.RemPipeline {
+		req.ListenerId = listenerID
+		return rpc.StopRem(ctx, req)
+	}
+
 	lns, err := core.Listeners.Get(listenerID)
 	if err != nil {
 		return nil, err
@@ -214,6 +233,13 @@ func (rpc *Server) DeletePipeline(ctx context.Context, req *clientpb.CtrlPipelin
 	if err != nil {
 		return nil, err
 	}
+
+	// Delegate REM pipelines to their dedicated handler.
+	if pipelineDB.Type == consts.RemPipeline {
+		req.ListenerId = listenerID
+		return rpc.DeleteRem(ctx, req)
+	}
+
 	lns, err := core.Listeners.Get(listenerID)
 	if err != nil {
 		return nil, err

server/rpc/rpc-rem.go

@@ -101,7 +101,10 @@ func (rpc *Server) StartRem(ctx context.Context, req *clientpb.CtrlPipeline) (*c
 	if err := db.EnablePipelineByListener(rem.Name, listenerID); err != nil {
 		return nil, err
 	}
-	core.Jobs.AddPipeline(rem)
+	// Do not call core.Jobs.AddPipeline(rem) here: the listener's
+	// handleStartRem already invoked SyncPipeline with the runtime-
+	// populated pipeline (Link, Subscribe, Port). Calling AddPipeline
+	// again with the stale DB snapshot would overwrite those values.
 
 	return &clientpb.Empty{}, nil
 }