From 8b14a9259e00cb3cfe599a02f4f3f78411bf9b64 Mon Sep 17 00:00:00 2001 From: iJuanPablo Date: Thu, 20 Aug 2026 00:34:28 -0400 Subject: [PATCH 1/4] fix(hub): implement implicit recipient fallback for agent outbound messages MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit handleAgentOutboundMessage's doc comment and an inline comment both state "recipient defaults to the agent's creator when not explicitly specified", but no such fallback existed — the function went straight from a failed explicit-recipient resolution to a 400 "recipient is required" error. This broke every automatic reply-forwarding path that doesn't set Recipient/RecipientID, including the assistant-reply Stop hook (pkg/sciontool/hooks/handlers/hub.go) used by both the web dashboard Messages tab and the Telegram plugin: an agent's reply would be captured successfully but silently fail to deliver with a 400 from the Hub. Add the missing fallback: when no recipient is given at all, resolve the agent's creator (falling back to owner) via the store and use their user record, mirroring the existing explicit-recipient resolution logic. Added TestOutboundMessage_ImplicitRecipientDefaultsToCreator to cover it. --- pkg/hub/handlers_agent_messaging.go | 23 +++++++++ pkg/hub/handlers_test.go | 73 +++++++++++++++++++++++++++++ 2 files changed, 96 insertions(+) diff --git a/pkg/hub/handlers_agent_messaging.go b/pkg/hub/handlers_agent_messaging.go index eec2275dd7..df901f890f 100644 --- a/pkg/hub/handlers_agent_messaging.go +++ b/pkg/hub/handlers_agent_messaging.go @@ -162,6 +162,29 @@ func (s *Server) handleAgentOutboundMessage(w http.ResponseWriter, r *http.Reque } } + // Implicit default: no recipient specified at all — fall back to the + // agent's creator (falling back further to its owner if creator is + // unset, e.g. agents created by automation). This is what lets an agent + // just call "send this reply back" without threading a recipient + // through every message-sending path (assistant-reply hooks, Telegram + // relay, etc.). + if recipientID == "" && recipient == "" { + creatorID := agent.CreatedBy + if creatorID == "" { + creatorID = agent.OwnerID + } + if creatorID != "" { + if u, err := s.store.GetUser(ctx, creatorID); err == nil { + recipientID = u.ID + name := u.DisplayName + if name == "" { + name = u.Email + } + recipient = "user:" + name + } + } + } + if recipientID == "" && recipient == "" { ValidationError(w, "recipient is required — specify a user with 'user:' or 'user:'", nil) return diff --git a/pkg/hub/handlers_test.go b/pkg/hub/handlers_test.go index 2b805bbe57..b3f44a906f 100644 --- a/pkg/hub/handlers_test.go +++ b/pkg/hub/handlers_test.go @@ -3024,3 +3024,76 @@ func TestOutboundMessage_UnknownRecipient(t *testing.T) { t.Errorf("expected 400 for unknown recipient, got %d: %s", rr.Code, rr.Body.String()) } } + +// TestOutboundMessage_ImplicitRecipientDefaultsToCreator verifies that a +// message sent with no recipient at all (the case hit by the assistant-reply +// hook and other auto-forwarding paths, which never set Recipient/RecipientID) +// falls back to the agent's creator instead of rejecting with "recipient is +// required" — matching the doc comment on handleAgentOutboundMessage. +func TestOutboundMessage_ImplicitRecipientDefaultsToCreator(t *testing.T) { + srv, s := testServer(t) + ctx := context.Background() + + creator := &store.User{ + ID: api.NewUUID(), + Email: "creator@example.com", + DisplayName: "Creator", + } + if err := s.CreateUser(ctx, creator); err != nil { + t.Fatal(err) + } + + project := &store.Project{ + ID: api.NewUUID(), + Name: "msg-implicit-project", + Slug: "msg-implicit-project", + Visibility: store.VisibilityPrivate, + } + if err := s.CreateProject(ctx, project); err != nil { + t.Fatal(err) + } + + rb := &store.RuntimeBroker{ + ID: tid("broker-msg-implicit"), + Name: "test-broker-implicit", + Slug: "test-broker-implicit", + Endpoint: "http://localhost:9801", + Status: store.BrokerStatusOnline, + } + if err := s.CreateRuntimeBroker(ctx, rb); err != nil { + t.Fatal(err) + } + + agent := &store.Agent{ + ID: api.NewUUID(), + Name: "sender-implicit", + Slug: "sender-implicit", + ProjectID: project.ID, + Phase: "running", + RuntimeBrokerID: tid("broker-msg-implicit"), + Visibility: store.VisibilityPrivate, + CreatedBy: creator.ID, + } + if err := s.CreateAgent(ctx, agent); err != nil { + t.Fatal(err) + } + + body, _ := json.Marshal(OutboundMessageRequest{ + Msg: "hello with no recipient set", + }) + req := httptest.NewRequest(http.MethodPost, "/api/v1/agents/"+agent.ID+"/outbound-message", bytes.NewReader(body)) + req.Header.Set("Content-Type", "application/json") + + agentIdent := &agentIdentityWrapper{&AgentTokenClaims{ + Claims: jwt.Claims{Subject: agent.ID}, + ProjectID: project.ID, + }} + req = req.WithContext(contextWithIdentity(req.Context(), agentIdent)) + + rr := httptest.NewRecorder() + srv.handleAgentOutboundMessage(rr, req, agent.ID) + + if rr.Code != http.StatusOK && rr.Code != http.StatusCreated { + t.Fatalf("expected success falling back to agent creator, got %d: %s", rr.Code, rr.Body.String()) + } +} From 97aaa929a9a0ef99c6c3f6c1e3a05c2954529d5d Mon Sep 17 00:00:00 2001 From: iJuanPablo Date: Thu, 20 Aug 2026 16:19:21 -0400 Subject: [PATCH 2/4] fix(hub): don't mask GetUser backend errors as 400 in recipient fallback MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Per gemini-code-assist review on this PR: the implicit-recipient fallback silently ignored any error from store.GetUser(), including a real backend failure (DB down, etc.), and fell through to the generic "recipient is required" 400. That mislabels a server-side problem as a client mistake. Distinguish store.ErrNotFound (creator/owner record legitimately missing — correct to fall through) from any other error (write a 500 via writeErrorFromErr, matching how errors are handled elsewhere in this file). --- pkg/hub/handlers_agent_messaging.go | 13 ++++++++++++- 1 file changed, 12 insertions(+), 1 deletion(-) diff --git a/pkg/hub/handlers_agent_messaging.go b/pkg/hub/handlers_agent_messaging.go index df901f890f..0baeeb46dd 100644 --- a/pkg/hub/handlers_agent_messaging.go +++ b/pkg/hub/handlers_agent_messaging.go @@ -174,13 +174,24 @@ func (s *Server) handleAgentOutboundMessage(w http.ResponseWriter, r *http.Reque creatorID = agent.OwnerID } if creatorID != "" { - if u, err := s.store.GetUser(ctx, creatorID); err == nil { + u, err := s.store.GetUser(ctx, creatorID) + switch { + case err == nil: recipientID = u.ID name := u.DisplayName if name == "" { name = u.Email } recipient = "user:" + name + case errors.Is(err, store.ErrNotFound): + // Creator/owner record no longer exists (e.g. deleted user). + // Fall through to the "recipient is required" response below. + default: + // A real backend error (DB down, etc.) shouldn't be reported + // as a 400 validation error — that would mask a transient + // failure as a client mistake. + writeErrorFromErr(w, err, "") + return } } } From 148c77164e808e0f447bcb05d5658f02575a5db2 Mon Sep 17 00:00:00 2001 From: Stephen Ierodiaconou Date: Sun, 23 Aug 2026 06:47:30 +0300 Subject: [PATCH 3/4] Revert the implicit recipient fallback Reverts 8b14a925 and 97aaa929 from this branch. The maintainer's response on #1230 was that the explicit path is preferred - errors early build better habits in agent messaging - and that with native chat becoming prominent the intended direction is for a channel+thread to be a valid recipient, so that an agent posts to a chat space rather than to a person. A creator fallback makes that target more ambiguous. The commits stay in this branch's history because the work that follows was built on top of them and the diagnosis in them was sound. Only the behaviour is removed, so the net change against main is the thread routing alone. --- pkg/hub/handlers_agent_messaging.go | 34 -------------- pkg/hub/handlers_test.go | 73 ----------------------------- 2 files changed, 107 deletions(-) diff --git a/pkg/hub/handlers_agent_messaging.go b/pkg/hub/handlers_agent_messaging.go index 0baeeb46dd..eec2275dd7 100644 --- a/pkg/hub/handlers_agent_messaging.go +++ b/pkg/hub/handlers_agent_messaging.go @@ -162,40 +162,6 @@ func (s *Server) handleAgentOutboundMessage(w http.ResponseWriter, r *http.Reque } } - // Implicit default: no recipient specified at all — fall back to the - // agent's creator (falling back further to its owner if creator is - // unset, e.g. agents created by automation). This is what lets an agent - // just call "send this reply back" without threading a recipient - // through every message-sending path (assistant-reply hooks, Telegram - // relay, etc.). - if recipientID == "" && recipient == "" { - creatorID := agent.CreatedBy - if creatorID == "" { - creatorID = agent.OwnerID - } - if creatorID != "" { - u, err := s.store.GetUser(ctx, creatorID) - switch { - case err == nil: - recipientID = u.ID - name := u.DisplayName - if name == "" { - name = u.Email - } - recipient = "user:" + name - case errors.Is(err, store.ErrNotFound): - // Creator/owner record no longer exists (e.g. deleted user). - // Fall through to the "recipient is required" response below. - default: - // A real backend error (DB down, etc.) shouldn't be reported - // as a 400 validation error — that would mask a transient - // failure as a client mistake. - writeErrorFromErr(w, err, "") - return - } - } - } - if recipientID == "" && recipient == "" { ValidationError(w, "recipient is required — specify a user with 'user:' or 'user:'", nil) return diff --git a/pkg/hub/handlers_test.go b/pkg/hub/handlers_test.go index b3f44a906f..2b805bbe57 100644 --- a/pkg/hub/handlers_test.go +++ b/pkg/hub/handlers_test.go @@ -3024,76 +3024,3 @@ func TestOutboundMessage_UnknownRecipient(t *testing.T) { t.Errorf("expected 400 for unknown recipient, got %d: %s", rr.Code, rr.Body.String()) } } - -// TestOutboundMessage_ImplicitRecipientDefaultsToCreator verifies that a -// message sent with no recipient at all (the case hit by the assistant-reply -// hook and other auto-forwarding paths, which never set Recipient/RecipientID) -// falls back to the agent's creator instead of rejecting with "recipient is -// required" — matching the doc comment on handleAgentOutboundMessage. -func TestOutboundMessage_ImplicitRecipientDefaultsToCreator(t *testing.T) { - srv, s := testServer(t) - ctx := context.Background() - - creator := &store.User{ - ID: api.NewUUID(), - Email: "creator@example.com", - DisplayName: "Creator", - } - if err := s.CreateUser(ctx, creator); err != nil { - t.Fatal(err) - } - - project := &store.Project{ - ID: api.NewUUID(), - Name: "msg-implicit-project", - Slug: "msg-implicit-project", - Visibility: store.VisibilityPrivate, - } - if err := s.CreateProject(ctx, project); err != nil { - t.Fatal(err) - } - - rb := &store.RuntimeBroker{ - ID: tid("broker-msg-implicit"), - Name: "test-broker-implicit", - Slug: "test-broker-implicit", - Endpoint: "http://localhost:9801", - Status: store.BrokerStatusOnline, - } - if err := s.CreateRuntimeBroker(ctx, rb); err != nil { - t.Fatal(err) - } - - agent := &store.Agent{ - ID: api.NewUUID(), - Name: "sender-implicit", - Slug: "sender-implicit", - ProjectID: project.ID, - Phase: "running", - RuntimeBrokerID: tid("broker-msg-implicit"), - Visibility: store.VisibilityPrivate, - CreatedBy: creator.ID, - } - if err := s.CreateAgent(ctx, agent); err != nil { - t.Fatal(err) - } - - body, _ := json.Marshal(OutboundMessageRequest{ - Msg: "hello with no recipient set", - }) - req := httptest.NewRequest(http.MethodPost, "/api/v1/agents/"+agent.ID+"/outbound-message", bytes.NewReader(body)) - req.Header.Set("Content-Type", "application/json") - - agentIdent := &agentIdentityWrapper{&AgentTokenClaims{ - Claims: jwt.Claims{Subject: agent.ID}, - ProjectID: project.ID, - }} - req = req.WithContext(contextWithIdentity(req.Context(), agentIdent)) - - rr := httptest.NewRecorder() - srv.handleAgentOutboundMessage(rr, req, agent.ID) - - if rr.Code != http.StatusOK && rr.Code != http.StatusCreated { - t.Fatalf("expected success falling back to agent creator, got %d: %s", rr.Code, rr.Body.String()) - } -} From e1e0c2ffbed3b0e8ced8b22ccb91a7a45c2a92c9 Mon Sep 17 00:00:00 2001 From: Stephen Ierodiaconou Date: Sun, 23 Aug 2026 06:23:33 +0300 Subject: [PATCH 4/4] feat(chat): make the thread part of reply affinity, not just the channel Reply affinity records which channel a user last spoke from, so an agent's untagged reply can be routed back to chat - and then has nothing to attach to, because a message needs a thread to land in a conversation. The reply arrives beside the conversation it answers rather than in it, which to whoever is watching the thread is indistinguishable from no reply at all. This is independent of who the reply is addressed to. An agent that names its recipient explicitly still cannot get its answer into the conversation the human is reading, because naming a person does not name a place. webchat_conversation_context now stores last_thread_id beside last_channel, and the affinity block in handleAgentOutboundMessage restores both. The thread is filled in only when the caller named none: an agent that addressed a thread explicitly has already said where its reply should go. Both write paths record it - the native web spoke from msg.ThreadID, and broker inbound from req.Message.ThreadID - so a channel that carries threads keeps them, and one that does not records an empty thread rather than inheriting a stale one. Adding the column is idempotent on both backends: postgres takes ADD COLUMN IF NOT EXISTS, and SQLite, which has no such form, treats a duplicate-column error as the success case. Existing rows get an empty thread and behave exactly as they do today until the user next speaks. --- pkg/hub/handlers_agent_messaging.go | 16 ++- pkg/hub/handlers_agent_messaging_test.go | 124 +++++++++++++++++++++++ pkg/hub/handlers_broker_inbound.go | 2 +- pkg/hub/webchannel.go | 2 +- pkg/hub/webchannel_store.go | 60 +++++++---- pkg/hub/webchannel_store_postgres.go | 36 ++++--- pkg/hub/webchannel_test.go | 71 +++++++++++-- 7 files changed, 263 insertions(+), 48 deletions(-) diff --git a/pkg/hub/handlers_agent_messaging.go b/pkg/hub/handlers_agent_messaging.go index eec2275dd7..89efbf3527 100644 --- a/pkg/hub/handlers_agent_messaging.go +++ b/pkg/hub/handlers_agent_messaging.go @@ -169,16 +169,22 @@ func (s *Server) handleAgentOutboundMessage(w http.ResponseWriter, r *http.Reque // Reply affinity (Phase 6, AC22): when the agent sends an untagged reply // (no explicit channel), check webchat_conversation_context for the - // (recipient, project, agent) triple. If a row exists, route to the - // channel the user last spoke from. If no row exists, leave channel - // empty so the message fans out to all spokes (today's default behavior). + // (recipient, project, agent) triple. If a row exists, route to where the + // user last spoke from. If no row exists, leave the route empty so the + // message fans out to all spokes (today's default behavior). + if req.Channel == "" && recipientID != "" && s.webChatStore != nil && s.GetMessageBrokerProxy() != nil { - if lastCh, err := s.webChatStore.GetLastChannel(ctx, recipientID, agent.ProjectID, agent.ID); err != nil { + lastCh, lastThread, err := s.webChatStore.GetLastRoute(ctx, recipientID, agent.ProjectID, agent.ID) + switch { + case err != nil: s.messageLog.Error("Failed to look up reply affinity", "recipient_id", recipientID, "agent_id", agent.ID, "error", err) // Non-fatal: fall through to fan-out-to-all behavior. - } else if lastCh != "" { + case lastCh != "": req.Channel = lastCh + if req.ThreadID == "" { + req.ThreadID = lastThread + } } } diff --git a/pkg/hub/handlers_agent_messaging_test.go b/pkg/hub/handlers_agent_messaging_test.go index 74c095ea48..269dc933fd 100644 --- a/pkg/hub/handlers_agent_messaging_test.go +++ b/pkg/hub/handlers_agent_messaging_test.go @@ -19,11 +19,15 @@ package hub import ( "bytes" "context" + "database/sql" "encoding/json" + "github.com/GoogleCloudPlatform/scion/pkg/eventbus" + "log/slog" "net/http" "net/http/httptest" "strconv" "strings" + "sync" "testing" "time" @@ -264,3 +268,123 @@ func TestOutboundMessage_TranscriptMirrorDoesNotStarveAgentMessages(t *testing.T rr.Code, rr.Body.String()) } } + +// recordingBus is a fan-out spoke that keeps what the handler published, so a +// test can assert the route the handler chose without wiring persistence. +type recordingBus struct { + mu sync.Mutex + sent []*messages.StructuredMessage +} + +func (b *recordingBus) Publish(_ context.Context, _ string, msg *messages.StructuredMessage) error { + b.mu.Lock() + defer b.mu.Unlock() + b.sent = append(b.sent, msg) + return nil +} + +func (b *recordingBus) Subscribe(string, eventbus.EventHandler) (eventbus.Subscription, error) { + return nil, nil +} + +func (b *recordingBus) Close() error { return nil } + +func (b *recordingBus) last() *messages.StructuredMessage { + b.mu.Lock() + defer b.mu.Unlock() + if len(b.sent) == 0 { + return nil + } + return b.sent[len(b.sent)-1] +} + +// TestOutboundMessage_ReplyAffinityRestoresRoute exercises the affinity block +// end to end: with a webchat store and a registered channel present, an agent +// reply that names no route is put back where the user last spoke. +func TestOutboundMessage_ReplyAffinityRestoresRoute(t *testing.T) { + srv, s := testServer(t) + ctx := context.Background() + + db, err := sql.Open("sqlite3", ":memory:") + if err != nil { + t.Fatalf("open sqlite: %v", err) + } + t.Cleanup(func() { _ = db.Close() }) + wcs := NewWebChatStore(db, "sqlite3") + if err := wcs.Init(); err != nil { + t.Fatalf("Init: %v", err) + } + srv.SetWebChatStore(wcs) + + // A fan-out bus with a "web" spoke, so the channel the affinity row names + // passes the registered-channel check below it. + rec := &recordingBus{} + bus := eventbus.NewFanOutEventBus([]eventbus.NamedEventBus{ + {Name: "web", ChannelID: "web", Bus: rec, Observer: true}, + }, slog.Default()) + srv.SetMessageBrokerProxy(NewMessageBrokerProxy(bus, s, srv.events, func() AgentDispatcher { return nil }, slog.Default())) + + project := &store.Project{ + ID: api.NewUUID(), Name: "aff", Slug: "aff", Visibility: store.VisibilityPrivate, + } + if err := s.CreateProject(ctx, project); err != nil { + t.Fatalf("CreateProject: %v", err) + } + human := &store.User{ID: api.NewUUID(), Email: "human@example.com", DisplayName: "Human", Status: store.UserStatusActive} + if err := s.CreateUser(ctx, human); err != nil { + t.Fatalf("CreateUser: %v", err) + } + agent := &store.Agent{ + ID: api.NewUUID(), Name: "a", Slug: "a", ProjectID: project.ID, + Phase: "running", Visibility: store.VisibilityPrivate, + } + if err := s.CreateAgent(ctx, agent); err != nil { + t.Fatalf("CreateAgent: %v", err) + } + + // The user last spoke in a web thread. + if err := wcs.RecordChannel(ctx, human.ID, project.ID, agent.ID, "web", "topic-42", time.Now()); err != nil { + t.Fatalf("RecordChannel: %v", err) + } + + send := func(t *testing.T, threadID string) *messages.StructuredMessage { + t.Helper() + body, _ := json.Marshal(OutboundMessageRequest{ + Recipient: "user:human@example.com", + Msg: "reply", + ThreadID: threadID, + }) + req := httptest.NewRequest(http.MethodPost, "/api/v1/agents/"+agent.ID+"/outbound-message", bytes.NewReader(body)) + req.Header.Set("Content-Type", "application/json") + req = req.WithContext(contextWithIdentity(req.Context(), &agentIdentityWrapper{&AgentTokenClaims{ + Claims: jwt.Claims{Subject: agent.ID}, ProjectID: project.ID, + }})) + rr := httptest.NewRecorder() + srv.handleAgentOutboundMessage(rr, req, agent.ID) + if rr.Code != http.StatusOK { + t.Fatalf("expected 200, got %d: %s", rr.Code, rr.Body.String()) + } + m := rec.last() + if m == nil { + t.Fatal("nothing was published to the spoke") + } + return m + } + + t.Run("an untagged reply is routed back to the thread", func(t *testing.T) { + m := send(t, "") + if m.Channel != "web" { + t.Errorf("channel = %q, want web", m.Channel) + } + if m.ThreadID != "topic-42" { + t.Errorf("thread = %q, want topic-42; an untagged reply lands beside the conversation without it", m.ThreadID) + } + }) + + t.Run("a thread named by the caller is not overwritten", func(t *testing.T) { + m := send(t, "topic-99") + if m.ThreadID != "topic-99" { + t.Errorf("thread = %q, want topic-99", m.ThreadID) + } + }) +} diff --git a/pkg/hub/handlers_broker_inbound.go b/pkg/hub/handlers_broker_inbound.go index 4561c45d46..e99f91c709 100644 --- a/pkg/hub/handlers_broker_inbound.go +++ b/pkg/hub/handlers_broker_inbound.go @@ -265,7 +265,7 @@ func (s *Server) handleBrokerInbound(w http.ResponseWriter, r *http.Request) { // Only record for user-identity senders with a known channel. if s.webChatStore != nil && req.Message.Channel != "" && strings.HasPrefix(req.Message.Sender, "user:") { if senderUserID != "" { - if err := s.webChatStore.RecordChannel(r.Context(), senderUserID, agent.ProjectID, agent.ID, req.Message.Channel, now); err != nil { + if err := s.webChatStore.RecordChannel(r.Context(), senderUserID, agent.ProjectID, agent.ID, req.Message.Channel, req.Message.ThreadID, now); err != nil { log.Error("Failed to record conversation context for broker inbound", "user_id", senderUserID, "agent_id", agent.ID, "channel", req.Message.Channel, "error", err) } diff --git a/pkg/hub/webchannel.go b/pkg/hub/webchannel.go index f3634c7678..2269181e9b 100644 --- a/pkg/hub/webchannel.go +++ b/pkg/hub/webchannel.go @@ -136,7 +136,7 @@ func (b *webChannelBus) Publish(ctx context.Context, topic string, msg *messages } // Reply affinity — still needed for cross-channel reply routing. - if err := b.store.RecordChannel(ctx, userID, projectID, agentID, "web", time.Now().UTC()); err != nil { + if err := b.store.RecordChannel(ctx, userID, projectID, agentID, "web", msg.ThreadID, time.Now().UTC()); err != nil { b.log.Error("Failed to record conversation context", "user_id", userID, "project_id", projectID, "agent_id", agentID, "error", err) return err diff --git a/pkg/hub/webchannel_store.go b/pkg/hub/webchannel_store.go index 5e717ec2ec..014a3cb606 100644 --- a/pkg/hub/webchannel_store.go +++ b/pkg/hub/webchannel_store.go @@ -50,16 +50,16 @@ type WebChatStore interface { // indexed read instead of an aggregate query. TouchThread(ctx context.Context, userID, projectID, agentID, messageID string, activityAt time.Time) error - // RecordChannel upserts reply-affinity context for (user, project, agent). - // Records the last channel a message was seen on, so the hub can route - // untagged replies back to the channel the user last spoke from. - RecordChannel(ctx context.Context, userID, projectID, agentID, channel string, messageAt time.Time) error + // RecordChannel upserts reply-affinity context for (user, project, agent): + // the channel and thread the user's message arrived on, so the hub can + // route an untagged reply back to where they were speaking. + RecordChannel(ctx context.Context, userID, projectID, agentID, channel, threadID string, messageAt time.Time) error - // GetLastChannel returns the last channel recorded for (user, project, agent), - // or "" if no row exists. Used for reply affinity: when an agent sends an - // untagged reply, the hub checks here to route to the channel the user last - // spoke from. - GetLastChannel(ctx context.Context, userID, projectID, agentID string) (string, error) + // GetLastRoute returns the last channel and thread recorded for (user, project, agent), + // or empty strings if no row exists. Used for reply affinity: when an agent + // sends an untagged reply, the hub routes it back to where the user last + // spoke. + GetLastRoute(ctx context.Context, userID, projectID, agentID string) (channel, threadID string, err error) // GetThreadPrefs returns the display preferences for a (user, project, agent) thread. // Returns default prefs (visibility_mode = "conversation") if no row exists. @@ -361,6 +361,7 @@ CREATE TABLE IF NOT EXISTS webchat_conversation_context ( project_id TEXT NOT NULL, agent_id TEXT NOT NULL, last_channel TEXT, + last_thread_id TEXT, last_message_at TEXT, PRIMARY KEY (user_id, project_id, agent_id) ); @@ -492,6 +493,17 @@ CREATE UNIQUE INDEX IF NOT EXISTS idx_webchat_topic_project_name return nil } +// addConversationContextThreadID adds last_thread_id to an existing +// webchat_conversation_context. SQLite has no ADD COLUMN IF NOT EXISTS, so a +// duplicate-column error is the success case on a database that already has it. +func (s *sqliteWebChatStore) addConversationContextThreadID() error { + _, err := s.db.Exec(`ALTER TABLE webchat_conversation_context ADD COLUMN last_thread_id TEXT`) + if err != nil && !strings.Contains(strings.ToLower(err.Error()), "duplicate column") { + return err + } + return nil +} + // TouchThread upserts the thread watermark for the given (user, project, agent) triple. func (s *sqliteWebChatStore) TouchThread(ctx context.Context, userID, projectID, agentID, messageID string, activityAt time.Time) error { const query = ` @@ -510,34 +522,37 @@ DO UPDATE SET } // RecordChannel upserts the reply-affinity context for the given (user, project, agent) triple. -func (s *sqliteWebChatStore) RecordChannel(ctx context.Context, userID, projectID, agentID, channel string, messageAt time.Time) error { +func (s *sqliteWebChatStore) RecordChannel(ctx context.Context, userID, projectID, agentID, channel, threadID string, messageAt time.Time) error { const query = ` -INSERT INTO webchat_conversation_context (user_id, project_id, agent_id, last_channel, last_message_at) -VALUES (?, ?, ?, ?, ?) +INSERT INTO webchat_conversation_context (user_id, project_id, agent_id, last_channel, last_thread_id, last_message_at) +VALUES (?, ?, ?, ?, ?, ?) ON CONFLICT (user_id, project_id, agent_id) DO UPDATE SET last_channel = excluded.last_channel, + last_thread_id = excluded.last_thread_id, last_message_at = excluded.last_message_at ` - _, err := s.db.ExecContext(ctx, query, userID, projectID, agentID, channel, messageAt) + _, err := s.db.ExecContext(ctx, query, userID, projectID, agentID, channel, threadID, messageAt) if err != nil { return fmt.Errorf("webchat store: record channel: %w", err) } return nil } -// GetLastChannel returns the last channel for (user, project, agent), or "" if no row exists. -func (s *sqliteWebChatStore) GetLastChannel(ctx context.Context, userID, projectID, agentID string) (string, error) { - const query = `SELECT last_channel FROM webchat_conversation_context WHERE user_id = ? AND project_id = ? AND agent_id = ?` - var channel sql.NullString - err := s.db.QueryRowContext(ctx, query, userID, projectID, agentID).Scan(&channel) +// GetLastRoute returns where this user last spoke to this agent. The thread is +// empty for channels that do not carry threads, and both are empty when no row +// exists. +func (s *sqliteWebChatStore) GetLastRoute(ctx context.Context, userID, projectID, agentID string) (string, string, error) { + const query = `SELECT last_channel, last_thread_id FROM webchat_conversation_context WHERE user_id = ? AND project_id = ? AND agent_id = ?` + var channel, threadID sql.NullString + err := s.db.QueryRowContext(ctx, query, userID, projectID, agentID).Scan(&channel, &threadID) if err != nil { if err == sql.ErrNoRows { - return "", nil + return "", "", nil } - return "", fmt.Errorf("webchat store: get last channel: %w", err) + return "", "", fmt.Errorf("webchat store: get last route: %w", err) } - return channel.String, nil + return channel.String, threadID.String, nil } // GetThreadPrefs returns the display preferences for the given (user, project, agent) triple. @@ -1215,6 +1230,9 @@ SELECT id, project_id, COALESCE(thread_id, ''), sender, msg, created // runMigrations executes idempotent data migrations. func (s *sqliteWebChatStore) runMigrations() error { + if err := s.addConversationContextThreadID(); err != nil { + return fmt.Errorf("conversation context thread_id: %w", err) + } if err := s.migrateThreadIDs(DefaultMigrationBatchSize); err != nil { return fmt.Errorf("thread_id backfill: %w", err) } diff --git a/pkg/hub/webchannel_store_postgres.go b/pkg/hub/webchannel_store_postgres.go index 833da144af..54ab655e52 100644 --- a/pkg/hub/webchannel_store_postgres.go +++ b/pkg/hub/webchannel_store_postgres.go @@ -52,6 +52,7 @@ CREATE TABLE IF NOT EXISTS webchat_conversation_context ( project_id TEXT NOT NULL, agent_id TEXT NOT NULL, last_channel TEXT, + last_thread_id TEXT, last_message_at TIMESTAMPTZ, PRIMARY KEY (user_id, project_id, agent_id) ); @@ -173,6 +174,14 @@ CREATE UNIQUE INDEX IF NOT EXISTS idx_webchat_topic_project_name return fmt.Errorf("webchat store: create name uniqueness index: %w", err) } + // Adds last_thread_id to databases created before it was in the schema. + const ctxThreadCol = ` +ALTER TABLE webchat_conversation_context ADD COLUMN IF NOT EXISTS last_thread_id TEXT; +` + if _, err := s.db.Exec(ctxThreadCol); err != nil { + return fmt.Errorf("webchat store: add conversation context thread_id: %w", err) + } + // Run idempotent migrations. if err := s.runMigrations(); err != nil { return fmt.Errorf("webchat store: migrations: %w", err) @@ -199,34 +208,37 @@ DO UPDATE SET } // RecordChannel upserts the reply-affinity context for the given (user, project, agent) triple. -func (s *pgWebChatStore) RecordChannel(ctx context.Context, userID, projectID, agentID, channel string, messageAt time.Time) error { +func (s *pgWebChatStore) RecordChannel(ctx context.Context, userID, projectID, agentID, channel, threadID string, messageAt time.Time) error { const query = ` -INSERT INTO webchat_conversation_context (user_id, project_id, agent_id, last_channel, last_message_at) -VALUES ($1, $2, $3, $4, $5) +INSERT INTO webchat_conversation_context (user_id, project_id, agent_id, last_channel, last_thread_id, last_message_at) +VALUES ($1, $2, $3, $4, $5, $6) ON CONFLICT (user_id, project_id, agent_id) DO UPDATE SET last_channel = EXCLUDED.last_channel, + last_thread_id = EXCLUDED.last_thread_id, last_message_at = EXCLUDED.last_message_at ` - _, err := s.db.ExecContext(ctx, query, userID, projectID, agentID, channel, messageAt) + _, err := s.db.ExecContext(ctx, query, userID, projectID, agentID, channel, threadID, messageAt) if err != nil { return fmt.Errorf("webchat store: record channel: %w", err) } return nil } -// GetLastChannel returns the last channel for (user, project, agent), or "" if no row exists. -func (s *pgWebChatStore) GetLastChannel(ctx context.Context, userID, projectID, agentID string) (string, error) { - const query = `SELECT last_channel FROM webchat_conversation_context WHERE user_id = $1 AND project_id = $2 AND agent_id = $3` - var channel sql.NullString - err := s.db.QueryRowContext(ctx, query, userID, projectID, agentID).Scan(&channel) +// GetLastRoute returns where this user last spoke to this agent. The thread is +// empty for channels that do not carry threads, and both are empty when no row +// exists. +func (s *pgWebChatStore) GetLastRoute(ctx context.Context, userID, projectID, agentID string) (string, string, error) { + const query = `SELECT last_channel, last_thread_id FROM webchat_conversation_context WHERE user_id = $1 AND project_id = $2 AND agent_id = $3` + var channel, threadID sql.NullString + err := s.db.QueryRowContext(ctx, query, userID, projectID, agentID).Scan(&channel, &threadID) if err != nil { if err == sql.ErrNoRows { - return "", nil + return "", "", nil } - return "", fmt.Errorf("webchat store: get last channel: %w", err) + return "", "", fmt.Errorf("webchat store: get last route: %w", err) } - return channel.String, nil + return channel.String, threadID.String, nil } // GetThreadPrefs returns the display preferences for the given (user, project, agent) triple. diff --git a/pkg/hub/webchannel_test.go b/pkg/hub/webchannel_test.go index 6647ee6370..95b1f06627 100644 --- a/pkg/hub/webchannel_test.go +++ b/pkg/hub/webchannel_test.go @@ -120,14 +120,15 @@ func TestWebChatStore_RecordChannel_Insert(t *testing.T) { ctx := context.Background() now := time.Now().Truncate(time.Second) - err := store.RecordChannel(ctx, "user1", "proj1", "agent1", "web", now) + err := store.RecordChannel(ctx, "user1", "proj1", "agent1", "web", "topic-1", now) require.NoError(t, err) - var channel string - err = db.QueryRow(`SELECT last_channel FROM webchat_conversation_context - WHERE user_id = 'user1' AND project_id = 'proj1' AND agent_id = 'agent1'`).Scan(&channel) + var channel, threadID string + err = db.QueryRow(`SELECT last_channel, last_thread_id FROM webchat_conversation_context + WHERE user_id = 'user1' AND project_id = 'proj1' AND agent_id = 'agent1'`).Scan(&channel, &threadID) require.NoError(t, err) require.Equal(t, "web", channel) + require.Equal(t, "topic-1", threadID) } func TestWebChatStore_RecordChannel_Upsert(t *testing.T) { @@ -138,10 +139,10 @@ func TestWebChatStore_RecordChannel_Upsert(t *testing.T) { t1 := time.Now().Truncate(time.Second) t2 := t1.Add(5 * time.Minute) - err := store.RecordChannel(ctx, "user1", "proj1", "agent1", "web", t1) + err := store.RecordChannel(ctx, "user1", "proj1", "agent1", "web", "topic-1", t1) require.NoError(t, err) - err = store.RecordChannel(ctx, "user1", "proj1", "agent1", "discord", t2) + err = store.RecordChannel(ctx, "user1", "proj1", "agent1", "discord", "topic-2", t2) require.NoError(t, err) var channel string @@ -151,10 +152,14 @@ func TestWebChatStore_RecordChannel_Upsert(t *testing.T) { require.NoError(t, err) require.Equal(t, 1, count) - err = db.QueryRow(`SELECT last_channel FROM webchat_conversation_context - WHERE user_id = 'user1' AND project_id = 'proj1' AND agent_id = 'agent1'`).Scan(&channel) + var threadID string + err = db.QueryRow(`SELECT last_channel, last_thread_id FROM webchat_conversation_context + WHERE user_id = 'user1' AND project_id = 'proj1' AND agent_id = 'agent1'`).Scan(&channel, &threadID) require.NoError(t, err) require.Equal(t, "discord", channel) + // The thread moves with the channel: a reply must not be put into the + // thread of a conversation the user has since left. + require.Equal(t, "topic-2", threadID) } // --- WebChannelBus tests --- @@ -620,3 +625,53 @@ func TestTouchDMActivity_EmptyMessageID(t *testing.T) { require.Equal(t, "old-msg", lastMsgID, "empty messageID should not overwrite last_message_id") require.True(t, activityAt.Valid, "last_activity_at should be set") } + +func TestWebChatStore_GetLastRoute(t *testing.T) { + store, db := newTestWebChatStore(t) + defer db.Close() //nolint:errcheck + + ctx := context.Background() + now := time.Now().Truncate(time.Second) + + t.Run("no row yields an empty route, not an error", func(t *testing.T) { + ch, th, err := store.GetLastRoute(ctx, "nobody", "proj1", "agent1") + require.NoError(t, err) + require.Empty(t, ch) + require.Empty(t, th) + }) + + t.Run("returns the channel and the thread within it", func(t *testing.T) { + require.NoError(t, store.RecordChannel(ctx, "user1", "proj1", "agent1", "web", "topic-1", now)) + ch, th, err := store.GetLastRoute(ctx, "user1", "proj1", "agent1") + require.NoError(t, err) + require.Equal(t, "web", ch) + require.Equal(t, "topic-1", th) + }) + + t.Run("a channel without threads yields a channel and no thread", func(t *testing.T) { + require.NoError(t, store.RecordChannel(ctx, "user2", "proj1", "agent1", "telegram", "", now)) + ch, th, err := store.GetLastRoute(ctx, "user2", "proj1", "agent1") + require.NoError(t, err) + require.Equal(t, "telegram", ch) + require.Empty(t, th, "a threadless channel must not inherit a stale thread") + }) +} + +func TestAddConversationContextThreadID(t *testing.T) { + store, db := newTestWebChatStore(t) + defer db.Close() //nolint:errcheck + + sq, ok := store.(*sqliteWebChatStore) + require.True(t, ok, "expected the sqlite store") + + // Init already ran it once; running it again must be a no-op rather than a + // duplicate-column error. + require.NoError(t, sq.addConversationContextThreadID()) + require.NoError(t, sq.addConversationContextThreadID()) + + ctx := context.Background() + require.NoError(t, store.RecordChannel(ctx, "u", "p", "a", "web", "topic-9", time.Now())) + _, th, err := store.GetLastRoute(ctx, "u", "p", "a") + require.NoError(t, err) + require.Equal(t, "topic-9", th) +}