fix(channels): keep direct chat replies in Multica (MUL-4988) (#5645)

Classify outbound replies by task origin (chat_input_task_id) before consulting a channel binding, so web/mobile direct-chat completions and failures stay inside Multica even when the session was originally created from Lark or Slack. Channel-created tasks continue to deliver. Closes #5644.

server/internal/integrations/lark/outbound.go

@@ -287,7 +287,6 @@ func (p *Patcher) processEvent(ctx context.Context, e events.Event) error {
 		// Issue / autopilot tasks have no chat_session.
 		return nil
 	}
-
 	binding, err := p.queries.GetLarkChatSessionBindingBySession(ctx, chatSessionID)
 	if err != nil {
 		if errors.Is(err, pgx.ErrNoRows) {
@@ -297,6 +296,18 @@ func (p *Patcher) processEvent(ctx context.Context, e events.Event) error {
 		return fmt.Errorf("lookup chat session binding: %w", err)
 	}
 
+	// Only bound sessions reach here, so classify the task origin before
+	// spending any send work. Web/mobile direct-chat tasks can reuse a session
+	// that originated in Lark, but their replies belong only in Multica.
+	// Channel-created tasks leave chat_input_task_id NULL and continue below.
+	task, err := p.queries.GetAgentTask(ctx, taskID)
+	if err != nil {
+		return fmt.Errorf("load agent task: %w", err)
+	}
+	if task.ChatInputTaskID.Valid {
+		return nil
+	}
+
 	inst, err := p.queries.GetLarkInstallation(ctx, binding.InstallationID)
 	if err != nil {
 		return fmt.Errorf("load installation: %w", err)

server/internal/integrations/lark/outbound_test.go

@@ -18,6 +18,8 @@ import (
 
 type fakePatcherQueries struct {
 	mu              sync.Mutex
+	task            db.AgentTaskQueue
+	taskErr         error
 	binding         ChatSessionBinding
 	bindingErr      error
 	installation    Installation
@@ -32,7 +34,7 @@ type fakePatcherQueries struct {
 }
 
 func (f *fakePatcherQueries) GetAgentTask(ctx context.Context, id pgtype.UUID) (db.AgentTaskQueue, error) {
-	return db.AgentTaskQueue{}, nil
+	return f.task, f.taskErr
 }
 func (f *fakePatcherQueries) GetChatSession(ctx context.Context, id pgtype.UUID) (db.ChatSession, error) {
 	return db.ChatSession{}, nil
@@ -345,6 +347,52 @@ func TestPatcherSkipsWhenNoChatSessionBinding(t *testing.T) {
 	}
 }
 
+// TestPatcherSkipsDirectChatTaskOnBoundSession guards the channel boundary:
+// opening a Lark-bound session in the web/mobile UI must not make that direct
+// conversation's reply or failure leak back into the external chat. Direct
+// tasks own an input batch through chat_input_task_id; channel tasks leave it
+// NULL and continue through the existing outbound paths.
+func TestPatcherSkipsDirectChatTaskOnBoundSession(t *testing.T) {
+	tests := []struct {
+		name      string
+		eventType string
+		payload   any
+	}{
+		{
+			name:      "completed reply",
+			eventType: protocol.EventChatDone,
+			payload:   protocol.ChatDonePayload{Content: "web-only answer"},
+		},
+		{
+			name:      "failed run",
+			eventType: protocol.EventTaskFailed,
+			payload:   map[string]any{"error": "web-only failure"},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			p, q, api := newTestPatcher(t)
+			taskID := uuidFromString(t, "ee999999-ee99-ee99-ee99-eeeeeeeeeeee")
+			q.task = db.AgentTaskQueue{ChatInputTaskID: taskID}
+
+			p.handleEvent(events.Event{
+				Type:          tt.eventType,
+				TaskID:        uuidString(taskID),
+				ChatSessionID: uuidString(q.binding.ChatSessionID),
+				Payload:       tt.payload,
+			})
+
+			api.mu.Lock()
+			defer api.mu.Unlock()
+			if len(api.textSent) != 0 || len(api.mdCardSent) != 0 || len(api.sent) != 0 || len(api.patched) != 0 {
+				t.Fatalf("direct task must produce no channel outbound; got text=%d markdown=%d cards=%d patches=%d",
+					len(api.textSent), len(api.mdCardSent), len(api.sent), len(api.patched))
+			}
+		})
+	}
+}
+
 // TestPatcherFailEventSendsErrorCard verifies the failure path still
 // surfaces a card. The visual distinction between a successful reply
 // (plain text bubble) and a failure (red header card) is genuinely

server/internal/integrations/slack/outbound.go

@@ -9,6 +9,7 @@ import (
 	"time"
 
 	"github.com/jackc/pgx/v5"
+	"github.com/jackc/pgx/v5/pgtype"
 	"github.com/slack-go/slack"
 
 	"github.com/multica-ai/multica/server/internal/events"
@@ -21,6 +22,7 @@ import (
 // outboundQueries is the slice of generated queries the Slack outbound
 // subscriber needs. *db.Queries satisfies it.
 type outboundQueries interface {
+	GetAgentTask(ctx context.Context, id pgtype.UUID) (db.AgentTaskQueue, error)
 	GetChannelChatSessionBindingBySession(ctx context.Context, arg db.GetChannelChatSessionBindingBySessionParams) (db.ChannelChatSessionBinding, error)
 	GetChannelInstallation(ctx context.Context, arg db.GetChannelInstallationParams) (db.ChannelInstallation, error)
 }
@@ -95,6 +97,22 @@ func (o *Outbound) processEvent(ctx context.Context, e events.Event) error {
 	if content == "" {
 		return nil // nothing to say (empty completion)
 	}
+	// Only bound, non-empty completions reach here, so classify the task origin
+	// before loading credentials or sending. Web/mobile direct-chat tasks can
+	// reuse a session that originated in Slack, but their replies belong only in
+	// Multica. Outbound delivery fails closed when the origin cannot be
+	// established; channel-created tasks leave chat_input_task_id NULL and send.
+	taskID, ok := chatDoneTaskID(e)
+	if !ok {
+		return nil
+	}
+	task, err := o.q.GetAgentTask(ctx, taskID)
+	if err != nil {
+		return fmt.Errorf("load agent task: %w", err)
+	}
+	if task.ChatInputTaskID.Valid {
+		return nil
+	}
 	inst, err := o.q.GetChannelInstallation(ctx, db.GetChannelInstallationParams{
 		ID:          binding.InstallationID,
 		ChannelType: string(TypeSlack),
@@ -120,6 +138,23 @@ func (o *Outbound) processEvent(ctx context.Context, e events.Event) error {
 	return nil
 }
 
+// chatDoneTaskID extracts the task id from the event envelope or the typed/map
+// payload emitted by TaskService. Outbound delivery fails closed when the task
+// origin cannot be established.
+func chatDoneTaskID(e events.Event) (pgtype.UUID, bool) {
+	raw := e.TaskID
+	if raw == "" {
+		switch p := e.Payload.(type) {
+		case protocol.ChatDonePayload:
+			raw = p.TaskID
+		case map[string]any:
+			raw, _ = p["task_id"].(string)
+		}
+	}
+	id, err := util.ParseUUID(raw)
+	return id, err == nil && id.Valid
+}
+
 // outboundTarget recovers the real send target from the chat binding. The
 // channel_chat_id may be a composite "channel:threadRoot" isolation key, so the
 // real channel id is read from the binding config (slackBindingConfig); the

server/internal/integrations/slack/outbound_test.go

@@ -27,6 +27,12 @@ type fakeOutboundQueries struct {
 	bindingErr error
 	inst       db.ChannelInstallation
 	instErr    error
+	task       db.AgentTaskQueue
+	taskErr    error
+}
+
+func (f *fakeOutboundQueries) GetAgentTask(context.Context, pgtype.UUID) (db.AgentTaskQueue, error) {
+	return f.task, f.taskErr
 }
 
 func (f *fakeOutboundQueries) GetChannelChatSessionBindingBySession(context.Context, db.GetChannelChatSessionBindingBySessionParams) (db.ChannelChatSessionBinding, error) {
@@ -69,7 +75,30 @@ func chatDoneEvent(sessionID string, content string) events.Event {
 	return events.Event{
 		Type:          protocol.EventChatDone,
 		ChatSessionID: sessionID,
-		Payload:       protocol.ChatDonePayload{Content: content},
+		Payload: protocol.ChatDonePayload{
+			TaskID:        "00000000-0000-0000-0000-000000000002",
+			ChatSessionID: sessionID,
+			Content:       content,
+		},
+	}
+}
+
+func TestOutbound_SkipsDirectChatTaskOnBoundSlackSession(t *testing.T) {
+	q := &fakeOutboundQueries{
+		task: db.AgentTaskQueue{ChatInputTaskID: uid(2)},
+		binding: db.ChannelChatSessionBinding{
+			InstallationID: uid(1),
+			ChannelChatID:  "C123",
+			Config:         []byte(`{"channel_id":"C123"}`),
+		},
+		inst: db.ChannelInstallation{ID: uid(1), Status: "active", Config: slackInstallConfigJSON()},
+	}
+	fs := &fakeSender{}
+
+	newTestOutbound(q, fs).handleEvent(chatDoneEvent("00000000-0000-0000-0000-000000000001", "private web reply"))
+
+	if fs.called != 0 {
+		t.Fatalf("sender called %d times, want 0 for a direct-chat task", fs.called)
 	}
 }