fix(engine): return error instead of panic when NATS engine is uninitialized (#18536)

internal/engine/nats/nats.go

@@ -108,6 +108,10 @@ func (n *NatsEngine) Type() string {
 }
 
 func (n *NatsEngine) PublishTask(subject string, payload []byte) error {
+	if n.jetStream == nil {
+		return errors.New("NATS jetstream is nil, engine not properly initialized")
+	}
+
 	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
 	defer cancel()
 
@@ -120,6 +124,10 @@ func (n *NatsEngine) PublishTask(subject string, payload []byte) error {
 }
 
 func (n *NatsEngine) ShowMessageQueue() (map[string]string, error) {
+	if n.jetStream == nil || n.stream == nil {
+		return nil, errors.New("NATS jetstream/stream is nil, engine not properly initialized")
+	}
+
 	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
 	defer cancel()
 	accountInfo, err := n.jetStream.AccountInfo(ctx)
@@ -226,6 +234,10 @@ func (n *NatsEngine) InitConsumer(subject string) error {
 	return nil
 }
 func (n *NatsEngine) GetMessages(messageCount int) ([]common.TaskHandle, error) {
+	if n.consumer == nil {
+		return nil, errors.New("NATS consumer is nil, engine not properly initialized")
+	}
+
 	resultMessages := make([]common.TaskHandle, 0)
 	messages, err := n.consumer.Fetch(messageCount, jetstream.FetchMaxWait(1*time.Second))
 	if err != nil {
@@ -238,6 +250,9 @@ func (n *NatsEngine) GetMessages(messageCount int) ([]common.TaskHandle, error)
 }
 
 func (n *NatsEngine) CheckStatus() string {
+	if n.nc == nil {
+		return "NATS connection is nil, engine not properly initialized"
+	}
 	n.nc.Stats()
 	return n.nc.Status().String()
 }

internal/engine/nats/nats_uninitialized_test.go

@@ -0,0 +1,33 @@
+package nats
+
+import (
+	"strings"
+	"testing"
+)
+
+// An engine whose Init failed (or was never called) must return errors from
+// its public methods instead of panicking on nil fields. Regression test for
+// the nil-deref panic at PublishTask: the server bootstrap only logs message
+// queue init failures, so a half-initialized engine stays registered
+// process-wide and every later task publish crashed the caller.
+func TestUninitializedEngineMethodsReturnErrors(t *testing.T) {
+	// Nothing listens on this port; Init() is deliberately not called so the
+	// engine keeps its zero-value (nil) connection, jetstream and consumer.
+	e := NewNatsEngine("127.0.0.1", 1)
+
+	if err := e.PublishTask("tasks.RAGFLOW", []byte("{}")); err == nil || !strings.Contains(err.Error(), "not properly initialized") {
+		t.Fatalf("PublishTask on uninitialized engine: err = %v, want 'not properly initialized'", err)
+	}
+
+	if _, err := e.GetMessages(1); err == nil || !strings.Contains(err.Error(), "not properly initialized") {
+		t.Fatalf("GetMessages on uninitialized engine: err = %v, want 'not properly initialized'", err)
+	}
+
+	if _, err := e.ShowMessageQueue(); err == nil || !strings.Contains(err.Error(), "not properly initialized") {
+		t.Fatalf("ShowMessageQueue on uninitialized engine: err = %v, want 'not properly initialized'", err)
+	}
+
+	if status := e.CheckStatus(); !strings.Contains(status, "not properly initialized") {
+		t.Fatalf("CheckStatus on uninitialized engine = %q, want 'not properly initialized'", status)
+	}
+}

internal/service/memory_queue_task_test.go

@@ -0,0 +1,34 @@
+package service
+
+import (
+	"strings"
+	"testing"
+
+	"ragflow/internal/engine"
+	natsengine "ragflow/internal/engine/nats"
+)
+
+// queueMemoryTask is the agent-canvas memory-save publish path. When the
+// message queue engine is a NatsEngine whose Init failed at boot, publishing
+// must surface a clean error (caught by the Message component's best-effort
+// memory save) rather than panicking and failing the whole canvas run.
+func TestQueueMemoryTaskUninitializedQueueReturnsError(t *testing.T) {
+	previous := engine.GetMessageQueueEngine()
+	engine.SetMessageQueueEngine(natsengine.NewNatsEngine("127.0.0.1", 1))
+	t.Cleanup(func() { engine.SetMessageQueueEngine(previous) })
+
+	err := queueMemoryTask(
+		t.Context(),
+		"mem-1",
+		"tenant-1",
+		1,
+		map[string]any{"id": "task-1", "task_type": "memory"},
+		MemoryMessage{UserID: "u1", AgentID: "agent-1", SessionID: "s1", UserInput: "hi", AgentResponse: "hello"},
+	)
+	if err == nil {
+		t.Fatal("queueMemoryTask with uninitialized MQ engine: err = nil, want publish error")
+	}
+	if !strings.Contains(err.Error(), "not properly initialized") {
+		t.Fatalf("queueMemoryTask err = %v, want engine-not-initialized error", err)
+	}
+}