diff --git a/server/internal/integrations/lark/outbound.go b/server/internal/integrations/lark/outbound.go index f84036aa8..5eae71346 100644 --- a/server/internal/integrations/lark/outbound.go +++ b/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) diff --git a/server/internal/integrations/lark/outbound_test.go b/server/internal/integrations/lark/outbound_test.go index c03a8a260..8fd178e30 100644 --- a/server/internal/integrations/lark/outbound_test.go +++ b/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 diff --git a/server/internal/integrations/slack/outbound.go b/server/internal/integrations/slack/outbound.go index 9770372bf..434b256a8 100644 --- a/server/internal/integrations/slack/outbound.go +++ b/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 diff --git a/server/internal/integrations/slack/outbound_test.go b/server/internal/integrations/slack/outbound_test.go index 514007938..724e7e98c 100644 --- a/server/internal/integrations/slack/outbound_test.go +++ b/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) } }