From 6b738924d5ccb3943f7af330eea53abbf64059ec Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E3=83=8A=E3=83=A0?= Date: Fri, 10 Jul 2026 23:25:45 +0800 Subject: [PATCH] fix(discord): qualify group titles across surfaces (#1420) Co-authored-by: Collective Developer --- cmd/gateway_announce_queue.go | 3 + cmd/gateway_consumer_handlers.go | 5 +- cmd/gateway_consumer_helpers.go | 23 ++ cmd/gateway_consumer_normal.go | 16 +- cmd/gateway_consumer_normal_test.go | 51 +++ cmd/gateway_cron.go | 2 + cmd/gateway_cron_test.go | 48 +++ cmd/gateway_heartbeat.go | 5 + cmd/gateway_http_handlers.go | 1 + cmd/gateway_subagent_announce_queue.go | 3 + docs/05-channels-messaging.md | 6 + docs/18-http-api.md | 52 ++- docs/22-heartbeat-system.md | 7 +- .../260710-discord-qualified-group-titles.md | 33 ++ internal/channelmemory/service.go | 38 ++ internal/channels/channel.go | 31 ++ internal/channels/discord/contact_refresh.go | 189 ++++++++- .../channels/discord/contact_refresh_test.go | 368 +++++++++++++++++- internal/channels/discord/discord.go | 43 +- internal/channels/discord/handler.go | 28 +- internal/channels/discord/handler_test.go | 30 ++ internal/channels/discord/hierarchy.go | 67 ++++ internal/channels/history.go | 31 ++ internal/channels/manager.go | 17 + internal/heartbeat/ticker.go | 57 ++- internal/http/channel_instance_metadata.go | 11 +- .../http/channel_instance_metadata_test.go | 4 +- .../http/channel_memory_extraction_test.go | 32 ++ internal/http/pending_messages.go | 26 +- internal/store/contact_collector.go | 63 +++ internal/store/contact_store.go | 23 +- internal/store/pending_message_store.go | 1 + internal/store/pg/channel_contact_metadata.go | 40 ++ internal/store/pg/channel_contacts.go | 20 +- internal/store/pg/heartbeat.go | 2 +- internal/store/pg/pending_message_store.go | 2 +- .../sqlitestore/channel_contact_metadata.go | 42 ++ internal/store/sqlitestore/contacts.go | 12 +- internal/store/sqlitestore/heartbeat.go | 3 +- .../heartbeat_delivery_targets_test.go | 42 ++ .../store/sqlitestore/pending_messages.go | 2 +- .../channels/hooks/use-channel-detail.ts | 14 +- .../pending-messages/message-list-dialog.tsx | 4 +- .../pending-messages-page.tsx | 4 +- .../src/pages/pending-messages/types.test.ts | 20 + ui/web/src/pages/pending-messages/types.ts | 9 + 46 files changed, 1442 insertions(+), 88 deletions(-) create mode 100644 docs/journals/260710-discord-qualified-group-titles.md create mode 100644 internal/channels/discord/hierarchy.go create mode 100644 internal/store/pg/channel_contact_metadata.go create mode 100644 internal/store/sqlitestore/channel_contact_metadata.go create mode 100644 internal/store/sqlitestore/heartbeat_delivery_targets_test.go create mode 100644 ui/web/src/pages/pending-messages/types.test.ts diff --git a/cmd/gateway_announce_queue.go b/cmd/gateway_announce_queue.go index 6c21d7b8..e3bc2bbb 100644 --- a/cmd/gateway_announce_queue.go +++ b/cmd/gateway_announce_queue.go @@ -11,6 +11,7 @@ import ( "github.com/nextlevelbuilder/goclaw/internal/agent" "github.com/nextlevelbuilder/goclaw/internal/bus" + "github.com/nextlevelbuilder/goclaw/internal/channels" "github.com/nextlevelbuilder/goclaw/internal/config" orch "github.com/nextlevelbuilder/goclaw/internal/orchestration" "github.com/nextlevelbuilder/goclaw/internal/scheduler" @@ -71,6 +72,7 @@ func processAnnounceLoop( teamStore store.TeamStore, postTurn tools.PostTurnProcessor, cfg *config.Config, + channelMgr *channels.Manager, ) { for { entries := teamAnnounceQueue.Drain(r.LeadSessionKey) @@ -95,6 +97,7 @@ func processAnnounceLoop( SessionKey: r.LeadSessionKey, Message: content, Channel: r.OrigChannel, + ChatTitle: resolveGroupDisplayTitle(ctx, channelMgr, r.OrigChannel, r.OrigChatID, r.OrigPeerKind, ""), ChatID: r.OrigChatID, PeerKind: r.OrigPeerKind, LocalKey: r.OrigLocalKey, diff --git a/cmd/gateway_consumer_handlers.go b/cmd/gateway_consumer_handlers.go index d245ed85..9588eabf 100644 --- a/cmd/gateway_consumer_handlers.go +++ b/cmd/gateway_consumer_handlers.go @@ -144,7 +144,7 @@ func handleSubagentAnnounce( // Fetch live roster for merged announce context. roster := deps.SubagentMgr.RosterForParent(parentAgent) - processSubagentAnnounceLoop(ctx, routing, roster, deps.SubagentMgr, deps.Sched, deps.MsgBus, deps.Cfg) + processSubagentAnnounceLoop(ctx, routing, roster, deps.SubagentMgr, deps.Sched, deps.MsgBus, deps.Cfg, deps.ChannelMgr) }) } @@ -246,6 +246,7 @@ func handleTeammateMessage( Channel: origChannel, ChannelType: origChannelType, ChatID: origChatID, + ChatTitle: resolveGroupDisplayTitle(schedCtx, deps.ChannelMgr, origChannel, origChatID, origPeerKind, ""), PeerKind: origPeerKind, LocalKey: origLocalKey, UserID: announceUserID, @@ -383,7 +384,7 @@ func handleTeammateMessage( ParentRootSpanID: parentRootSpanID, OutMeta: outMeta, } - processAnnounceLoop(ctx, routing, deps.Sched, deps.MsgBus, deps.TeamStore, deps.PostTurn, deps.Cfg) + processAnnounceLoop(ctx, routing, deps.Sched, deps.MsgBus, deps.TeamStore, deps.PostTurn, deps.Cfg, deps.ChannelMgr) }(origChannel, origChatID, msg.SenderID, taskIDStr, outMeta, msg.Metadata) return true diff --git a/cmd/gateway_consumer_helpers.go b/cmd/gateway_consumer_helpers.go index 07d5ae9b..6ae4301c 100644 --- a/cmd/gateway_consumer_helpers.go +++ b/cmd/gateway_consumer_helpers.go @@ -129,6 +129,29 @@ func extractSessionMetadata(msg bus.InboundMessage, peerKind string) map[string] return meta } +// resolveGroupDisplayTitle fills presentation context for internally +// re-ingressed group messages that no longer carry inbound chat metadata. +// It leaves routing identifiers untouched and silently falls back when a +// channel cannot resolve a platform-specific display title. +func resolveGroupDisplayTitle(ctx context.Context, mgr *channels.Manager, channel, chatID, peerKind, title string) string { + if title != "" || peerKind != string(sessions.PeerGroup) || mgr == nil || channel == "" || chatID == "" { + return title + } + resolved, err := mgr.ResolveGroupDisplayTitle(ctx, channel, chatID) + if err != nil { + return title + } + return resolved +} + +func resolveInboundChatTitle(ctx context.Context, mgr *channels.Manager, msg bus.InboundMessage, peerKind string) string { + title := msg.Metadata[tools.MetaChatTitle] + if !bus.IsInternalSender(msg.SenderID) { + return title + } + return resolveGroupDisplayTitle(ctx, mgr, msg.Channel, msg.ChatID, peerKind, title) +} + // buildPancakeSessionLabel returns "Pancake:{senderName}:{pageName}" with non-empty parts only. func buildPancakeSessionLabel(senderName, pageName string) string { label := "Pancake" diff --git a/cmd/gateway_consumer_normal.go b/cmd/gateway_consumer_normal.go index d63e896f..73056f89 100644 --- a/cmd/gateway_consumer_normal.go +++ b/cmd/gateway_consumer_normal.go @@ -67,6 +67,17 @@ func processNormalMessage( if peerKind == "" { peerKind = string(sessions.PeerDirect) // default to DM } + // Channel adapters already attach cache-only titles to external inbound + // messages. Resolve live hierarchy only for synthetic/re-ingressed messages, + // where no adapter metadata exists; this keeps Discord REST off the user + // message hot path. + chatTitle := resolveInboundChatTitle(ctx, deps.ChannelMgr, msg, peerKind) + if chatTitle != "" && chatTitle != msg.Metadata[tools.MetaChatTitle] { + if msg.Metadata == nil { + msg.Metadata = make(map[string]string) + } + msg.Metadata[tools.MetaChatTitle] = chatTitle + } sessionKey := sessions.BuildScopedSessionKey(agentID, msg.Channel, sessions.PeerKind(peerKind), msg.ChatID) // Thread-based isolation override (e.g. Slack DM threads, AI Panel) @@ -136,7 +147,10 @@ func processNormalMessage( // Also collect group chat as a contact (for group permission management / merge). // Group IDs (e.g., Telegram "-100456") differ from user IDs — no UNIQUE conflict. - if peerKind == string(sessions.PeerGroup) && msg.ChatID != "" { + // Discord persists raw entity names plus hierarchy display metadata in its + // adapter. Do not overwrite that raw contact title with ChatTitle, which + // is intentionally parent-qualified for prompt/display context. + if peerKind == string(sessions.PeerGroup) && msg.ChatID != "" && channelType != channels.TypeDiscord { groupTitle := msg.Metadata[tools.MetaChatTitle] // Telegram: message.Chat.Title deps.ContactCollector.EnsureContact(ctx, channelType, msg.Channel, msg.ChatID, "", groupTitle, "", "group", "group", "", "") } diff --git a/cmd/gateway_consumer_normal_test.go b/cmd/gateway_consumer_normal_test.go index c6465fc8..a15e6787 100644 --- a/cmd/gateway_consumer_normal_test.go +++ b/cmd/gateway_consumer_normal_test.go @@ -136,6 +136,46 @@ func TestDeriveGroupUserID(t *testing.T) { } } +func TestResolveGroupDisplayTitleFallsBackThroughChannelManager(t *testing.T) { + manager := channels.NewManager(nil) + manager.RegisterChannel("discord-main", &consumerDisplayTitleChannel{ + consumerTestChannel: consumerTestChannel{name: "discord-main", channelType: channels.TypeDiscord}, + title: "launch-thread / product-planning", + }) + + if got := resolveGroupDisplayTitle(context.Background(), manager, "discord-main", "thread-1", string(sessions.PeerGroup), ""); got != "launch-thread / product-planning" { + t.Fatalf("resolved title = %q, want qualified title", got) + } + if got := resolveGroupDisplayTitle(context.Background(), manager, "discord-main", "thread-1", string(sessions.PeerGroup), "already present"); got != "already present" { + t.Fatalf("metadata title = %q, want original value", got) + } +} + +func TestResolveInboundChatTitleAvoidsLiveLookupForExternalMessage(t *testing.T) { + manager := channels.NewManager(nil) + channel := &consumerDisplayTitleChannel{ + consumerTestChannel: consumerTestChannel{name: "discord-main", channelType: channels.TypeDiscord}, + title: "launch-thread / product-planning", + } + manager.RegisterChannel("discord-main", channel) + + external := bus.InboundMessage{Channel: "discord-main", ChatID: "thread-1", SenderID: "user-1"} + if got := resolveInboundChatTitle(context.Background(), manager, external, string(sessions.PeerGroup)); got != "" { + t.Fatalf("external title = %q, want no live fallback", got) + } + if channel.calls != 0 { + t.Fatalf("external message invoked display resolver %d times", channel.calls) + } + + internal := bus.InboundMessage{Channel: "discord-main", ChatID: "thread-1", SenderID: "system:ticker"} + if got := resolveInboundChatTitle(context.Background(), manager, internal, string(sessions.PeerGroup)); got != channel.title { + t.Fatalf("internal title = %q, want %q", got, channel.title) + } + if channel.calls != 1 { + t.Fatalf("internal message invoked display resolver %d times, want 1", channel.calls) + } +} + func TestResolveSenderNameReadsWhatsAppUserName(t *testing.T) { got := resolveSenderName(bus.InboundMessage{ Metadata: map[string]string{ @@ -237,6 +277,17 @@ type consumerTestChannel struct { running bool } +type consumerDisplayTitleChannel struct { + consumerTestChannel + title string + calls int +} + +func (c *consumerDisplayTitleChannel) ResolveGroupDisplayTitle(context.Context, string) (string, error) { + c.calls++ + return c.title, nil +} + func (c consumerTestChannel) Name() string { return c.name } func (c consumerTestChannel) Type() string { return c.channelType } func (c consumerTestChannel) Start(context.Context) error { diff --git a/cmd/gateway_cron.go b/cmd/gateway_cron.go index 77db869e..07d3aec9 100644 --- a/cmd/gateway_cron.go +++ b/cmd/gateway_cron.go @@ -123,6 +123,7 @@ func makeCronJobHandler(sched *scheduler.Scheduler, msgBus *bus.MessageBus, cfg if job.Payload.CredentialUserID != "" { cronCtx = store.WithCredentialUserID(cronCtx, job.Payload.CredentialUserID) } + chatTitle := resolveGroupDisplayTitle(cronCtx, channelMgr, channel, job.DeliverTo, peerKind, "") // Reset the session before each STATELESS cron run so the run starts fresh: // no carried-over history (the whole point of stateless — saves tokens) and @@ -171,6 +172,7 @@ func makeCronJobHandler(sched *scheduler.Scheduler, msgBus *bus.MessageBus, cfg Channel: channel, ChannelType: channelType, ChatID: job.DeliverTo, + ChatTitle: chatTitle, PeerKind: peerKind, UserID: job.UserID, RunID: fmt.Sprintf("cron:%s", job.ID), diff --git a/cmd/gateway_cron_test.go b/cmd/gateway_cron_test.go index e63991c8..91af25f1 100644 --- a/cmd/gateway_cron_test.go +++ b/cmd/gateway_cron_test.go @@ -10,6 +10,7 @@ import ( "github.com/nextlevelbuilder/goclaw/internal/agent" "github.com/nextlevelbuilder/goclaw/internal/bus" + "github.com/nextlevelbuilder/goclaw/internal/channels" "github.com/nextlevelbuilder/goclaw/internal/config" "github.com/nextlevelbuilder/goclaw/internal/scheduler" "github.com/nextlevelbuilder/goclaw/internal/store" @@ -71,6 +72,53 @@ func TestCronJobHandlerInjectsPayloadCredentialUserID(t *testing.T) { } } +func TestCronJobHandlerResolvesGroupDisplayTitle(t *testing.T) { + var got agent.RunRequest + sched := scheduler.NewScheduler( + scheduler.DefaultLanes(), + scheduler.QueueConfig{Mode: scheduler.QueueModeQueue, Cap: 1, MaxConcurrent: 1}, + func(_ context.Context, req agent.RunRequest) (*agent.RunResult, error) { + got = req + return &agent.RunResult{Content: "ok"}, nil + }, + ) + defer sched.Stop() + msgBus := bus.New() + defer msgBus.Close() + manager := channels.NewManager(nil) + manager.RegisterChannel("discord-main", cronDisplayTitleChannel{consumerTestChannel: consumerTestChannel{name: "discord-main", channelType: channels.TypeDiscord}, title: "launch-thread / product-planning"}) + + handler := makeCronJobHandler(sched, msgBus, &config.Config{}, manager, nil, nil, nil, nil, nil) + if _, err := handler(&store.CronJob{ + ID: uuid.NewString(), + TenantID: uuid.New(), + Name: "thread-report", + AgentID: "reporter", + UserID: "guild:guild-1:user:user-1", + Deliver: true, + DeliverChannel: "discord-main", + DeliverTo: "thread-1", + Payload: store.CronPayload{Kind: "agent_turn", Message: "report"}, + }); err != nil { + t.Fatalf("cron handler: %v", err) + } + if got.ChatID != "thread-1" { + t.Fatalf("chat ID = %q, want stable thread ID", got.ChatID) + } + if got.ChatTitle != "launch-thread / product-planning" { + t.Fatalf("chat title = %q, want qualified title", got.ChatTitle) + } +} + +type cronDisplayTitleChannel struct { + consumerTestChannel + title string +} + +func (c cronDisplayTitleChannel) ResolveGroupDisplayTitle(context.Context, string) (string, error) { + return c.title, nil +} + func TestCronOutputContainsNoReplySentinel(t *testing.T) { tests := []struct { name string diff --git a/cmd/gateway_heartbeat.go b/cmd/gateway_heartbeat.go index 8dd6522c..b6c048eb 100644 --- a/cmd/gateway_heartbeat.go +++ b/cmd/gateway_heartbeat.go @@ -15,6 +15,7 @@ import ( "github.com/nextlevelbuilder/goclaw/internal/heartbeat" "github.com/nextlevelbuilder/goclaw/internal/providers" "github.com/nextlevelbuilder/goclaw/internal/scheduler" + "github.com/nextlevelbuilder/goclaw/internal/sessions" "github.com/nextlevelbuilder/goclaw/internal/store" "github.com/nextlevelbuilder/goclaw/internal/tools" "github.com/nextlevelbuilder/goclaw/pkg/protocol" @@ -60,6 +61,10 @@ func startCronAndHeartbeat( MsgBus: msgBus, Sched: sched, RunAgent: makeHeartbeatRunFn(sched), + ResolveGroupContext: func(ctx context.Context, channel, chatID string) (string, string) { + channelType := resolveChannelType(channelMgr, channel) + return channelType, resolveGroupDisplayTitle(ctx, channelMgr, channel, chatID, string(sessions.PeerGroup), "") + }, }) heartbeatTicker.SetOnEvent(func(event store.HeartbeatEvent) { server.BroadcastEvent(*protocol.NewEvent(protocol.EventHeartbeat, event)) diff --git a/cmd/gateway_http_handlers.go b/cmd/gateway_http_handlers.go index 71c8ea8b..30aa7b2e 100644 --- a/cmd/gateway_http_handlers.go +++ b/cmd/gateway_http_handlers.go @@ -138,6 +138,7 @@ func makeChannelMemoryService(stores *store.Stores, domainBus eventbus.DomainEve return &channelmemory.Service{ Channels: stores.ChannelInstances, Pending: stores.PendingMessages, + Contacts: stores.Contacts, Extractions: stores.ChannelMemory, Episodic: stores.Episodic, EventBus: domainBus, diff --git a/cmd/gateway_subagent_announce_queue.go b/cmd/gateway_subagent_announce_queue.go index 510947d2..1e20cf43 100644 --- a/cmd/gateway_subagent_announce_queue.go +++ b/cmd/gateway_subagent_announce_queue.go @@ -11,6 +11,7 @@ import ( "github.com/nextlevelbuilder/goclaw/internal/agent" "github.com/nextlevelbuilder/goclaw/internal/bus" + "github.com/nextlevelbuilder/goclaw/internal/channels" "github.com/nextlevelbuilder/goclaw/internal/config" orch "github.com/nextlevelbuilder/goclaw/internal/orchestration" "github.com/nextlevelbuilder/goclaw/internal/scheduler" @@ -129,6 +130,7 @@ func processSubagentAnnounceLoop( sched *scheduler.Scheduler, msgBus *bus.MessageBus, cfg *config.Config, + channelMgr *channels.Manager, ) { // Ensure tenant scope is always set for the scheduler. if r.TenantID != uuid.Nil { @@ -176,6 +178,7 @@ func processSubagentAnnounceLoop( Channel: r.OrigChannel, ChannelType: r.OrigChannelType, ChatID: r.OrigChatID, + ChatTitle: resolveGroupDisplayTitle(ctx, channelMgr, r.OrigChannel, r.OrigChatID, r.OrigPeerKind, ""), PeerKind: r.OrigPeerKind, LocalKey: r.OrigLocalKey, UserID: r.UserID, diff --git a/docs/05-channels-messaging.md b/docs/05-channels-messaging.md index 83a896f8..4ac92087 100644 --- a/docs/05-channels-messaging.md +++ b/docs/05-channels-messaging.md @@ -574,6 +574,8 @@ The Discord channel uses the `discordgo` library to connect via the Discord Gate - **Typing indicator**: 9-second keepalive while agent processes - **Group history**: Pending message buffer for context when mentioned - **Thread backfill**: When the bot is mentioned inside a Discord thread, the channel fetches up to 25 prior thread messages before the triggering message through Discord REST, prepends their text as context, and downloads up to 15 prior attachments for the same inbound media pipeline. This is thread-only, bounded to 5 MB per backfilled file with a 30-second timeout, and gracefully falls back to the current message when Discord lacks `READ_MESSAGE_HISTORY` or the REST request fails. +- **Thread presentation titles**: Where a Discord thread is presented as the current chat, a session, a contact, or a delivery target, its label may be qualified as `thread / parent channel`. This is display-only; routing and selection continue to use stable IDs. +- **Metadata refresh**: Refreshing the contact cache unions the Discord group IDs stored in contacts with group and parent IDs retained by pending-message history. It directly looks up targets not already resolved from the live guild/channel state, so titles and parent metadata can be backfilled for inactive or archived threads when Discord returns them. The API response includes per-source coverage and individual failures; the gateway emits an INFO summary for every refresh and WARN entries for individual lookup failures. If any target cannot be fetched (for example, it was deleted or the bot lacks permission), the refresh reports failure rather than false success. --- @@ -726,6 +728,10 @@ Default behavior is privacy-first: - Discord extraction input includes best-effort channel context when available: channel/thread ID, channel name, parent channel, category, and history key. Lookup failures fall back to IDs and do not fail extraction. +- Discord passive-memory metadata retains raw `group_title` and + `parent_group_title` as separate fields. A qualified thread label such as + `thread / parent channel` is presentation-only and is not written back into + either raw field. - New Discord pending history rows include display name, handle when available, and stable Discord user ID in sender/reply labels to make extracted facts less ambiguous when display names change. diff --git a/docs/18-http-api.md b/docs/18-http-api.md index 15edd322..cecd82c1 100644 --- a/docs/18-http-api.md +++ b/docs/18-http-api.md @@ -1098,7 +1098,47 @@ Accepts partial updates. Flag keys are validated against recognized v3 flags. | `GET` | `/v1/channels/instances/{id}` | Get instance | | `PUT` | `/v1/channels/instances/{id}` | Update instance | | `DELETE` | `/v1/channels/instances/{id}` | Delete instance (not default) | -| `POST` | `/v1/channels/instances/{id}/metadata/refresh` | Trigger a Discord metadata refresh into the channel contact cache | +| `POST` | `/v1/channels/instances/{id}/metadata/refresh` | Refresh Discord group metadata and return diagnostic coverage/failures | + +### `POST /v1/channels/instances/{id}/metadata/refresh` + +Available only for a running Discord channel instance and requires tenant-admin +access. The refresh unions IDs from the channel contact cache with group and +parent IDs retained by pending-message history. IDs already found in the live +Discord guild/channel state are reused; remaining IDs are fetched directly, +including inactive or archived threads when Discord permits access. + +The response is always shaped as `{ "ok": boolean, "report": object }` for a +completed refresh. `ok` is `false` when any target failed, while the HTTP status +remains `200` so callers can inspect the complete report. Setup or execution +failures return `409` with the same `report` plus a top-level `error`. + +```json +{ + "ok": false, + "report": { + "ok": false, + "groups_refreshed": 498, + "users_refreshed": 0, + "contact_targets": 501, + "pending_message_targets": 37, + "live_targets": 490, + "direct_lookup_attempts": 11, + "direct_lookup_resolved": 10, + "failures": [{ + "channel_id": "1381538956478251038", + "source": "pending_messages", + "reason": "HTTP 403 Forbidden" + }] + } +} +``` + +`failures[].source` identifies the ID source (`contacts`, `pending_messages`, +or both as a comma-separated value). The report may also contain `errors` for +failures while listing contact or pending-message targets. The gateway logs one +INFO summary (`discord contact cache refreshed`) with the report counts and a +WARN log for each direct lookup or thread-parent lookup failure. ### Contacts @@ -1110,6 +1150,10 @@ Accepts partial updates. Flag keys are validated against recognized v3 flags. | `POST` | `/v1/contacts/unmerge` | Unmerge previously merged contacts | | `GET` | `/v1/contacts/merged/{tenantUserId}` | List merged contacts for tenant user | +Discord thread labels used in current-chat, session, contact, and delivery-target +displays may be qualified as `thread / parent channel`. This presentation does +not change the stable IDs used by API consumers. + ### Tenant Users | Method | Path | Description | @@ -1154,7 +1198,7 @@ require tenant admin. | `PUT` | `/v1/channels/instances/{id}/memory-extraction/settings` | Replace normalized `passive_memory` config in the channel instance config | | `POST` | `/v1/channels/instances/{id}/memory-extraction/run` | Trigger a manual extraction run for the latest eligible group | | `POST` | `/v1/channels/instances/{id}/memory-extraction/run-all` | Trigger manual extraction for all eligible groups, skipping groups below `min_messages` | -| `GET` | `/v1/channels/instances/{id}/memory-extraction/groups` | List Discord history groups for tuning, including best-effort `group_title` and `parent_group_title` | +| `GET` | `/v1/channels/instances/{id}/memory-extraction/groups` | List Discord history groups for tuning, including best-effort raw `group_title` and `parent_group_title` | | `GET` | `/v1/channels/instances/{id}/memory-extraction/items` | List review queue items; optional `status` filter | | `POST` | `/v1/channels/instances/{id}/memory-extraction/items/{itemID}/approve` | Write candidate to episodic memory and publish KG event | | `POST` | `/v1/channels/instances/{id}/memory-extraction/items/{itemID}/reject` | Reject candidate | @@ -1178,6 +1222,10 @@ the exact thread prompt when present, then falls back to the parent channel prompt. Tenant-global prompt remains stored in system config key `channel_memory.extraction.custom_prompt`. +For Discord threads, `group_title` and `parent_group_title` remain separate raw +metadata. A `thread / parent channel` title may be used for presentation, but +does not alter those fields. + Review queue items include `topics` and `entities` arrays for prompt tuning and debugging. Extraction run/item tables store candidate summaries and metadata but do not store raw prompt or raw input text. diff --git a/docs/22-heartbeat-system.md b/docs/22-heartbeat-system.md index 03fa7fbc..88967a50 100644 --- a/docs/22-heartbeat-system.md +++ b/docs/22-heartbeat-system.md @@ -207,7 +207,10 @@ t.msgBus.PublishOutbound(bus.OutboundMessage{ }) ``` -**Delivery targets** are discovered via `heartbeat.targets` RPC, which lists distinct `(channel, chatID)` pairs from the agent's session history. The UI presents these as a dropdown for easy selection. +**Delivery targets** are discovered via `heartbeat.targets` RPC from +tenant-scoped channel contacts, not session history. The UI presents these as a +dropdown while retaining the contacts' stable channel and chat IDs for +selection and delivery. --- @@ -224,7 +227,7 @@ All methods are registered under the `heartbeat.*` namespace on the WebSocket me | `heartbeat.logs` | `agentId`, `limit`, `offset` | Paginated run history | | `heartbeat.checklist.get` | `agentId` | Read HEARTBEAT.md content | | `heartbeat.checklist.set` | `agentId`, `content` | Write HEARTBEAT.md content | -| `heartbeat.targets` | `agentId` | List known delivery targets from session history | +| `heartbeat.targets` | `agentId` | List known delivery targets from tenant-scoped channel contacts | ### Validation Rules (heartbeat.set) diff --git a/docs/journals/260710-discord-qualified-group-titles.md b/docs/journals/260710-discord-qualified-group-titles.md new file mode 100644 index 00000000..07796e91 --- /dev/null +++ b/docs/journals/260710-discord-qualified-group-titles.md @@ -0,0 +1,33 @@ +# Discord Qualified Group Titles + +**Date**: 2026-07-10 +**Component**: Discord channel metadata and delivery routing +**Status**: Resolved + +## Context + +Discord group names can be ambiguous when a thread and its parent need to be distinguished. Passive Memory continues to retain the raw `group_title` and `parent_group_title` independently; this change adds only a qualified display title, `thread / parent`, as presentation metadata. + +## What Happened + +The qualified title is persisted and now follows the group through contacts, delivery targets, current context, cron jobs, heartbeats, and synthetic re-ingress. Stable Discord identifiers were deliberately left unchanged, so routing and stored relationships retain their existing identity contract. + +The backfill gap was traced to archived or inactive groups falling outside the prior lookup window. Backfill now uses stable pagination beyond 100 results, so qualifying stored groups does not depend on their activity or sort position. + +## Decision + +The inbound hot path must not perform a REST lookup merely to decorate a title. An attempted REST fallback there was rejected and corrected: only synthetic re-ingress resolves live Discord state, where the extra lookup is appropriate. + +Stored IDs that cannot be read now produce an explicit refresh failure rather than silently retaining stale or incomplete presentation metadata. Together with stable pagination, this makes maintenance behavior observable and avoids a partial, misleading backfill. + +## Verification + +- `go test ./...` +- PostgreSQL and SQLite builds +- `go vet ./...` +- Web lint and production build +- Review P1 finding fixed + +## Reflection + +Keeping raw memory fields separate from display metadata prevents a UI convenience from silently becoming an identity or memory-schema change. Qualifying at the presentation boundary preserves useful context without adding latency to ordinary inbound messages. diff --git a/internal/channelmemory/service.go b/internal/channelmemory/service.go index 2c18f4d3..0330dafd 100644 --- a/internal/channelmemory/service.go +++ b/internal/channelmemory/service.go @@ -11,6 +11,7 @@ import ( "time" "github.com/google/uuid" + "github.com/nextlevelbuilder/goclaw/internal/channels" "github.com/nextlevelbuilder/goclaw/internal/eventbus" "github.com/nextlevelbuilder/goclaw/internal/providerresolve" "github.com/nextlevelbuilder/goclaw/internal/providers" @@ -21,6 +22,7 @@ import ( type Service struct { Channels store.ChannelInstanceStore Pending store.PendingMessageStore + Contacts store.ContactStore Extractions store.ChannelMemoryExtractionStore Episodic store.EpisodicStore EventBus eventbus.DomainEventBus @@ -122,10 +124,12 @@ func (s *Service) GroupOptions(ctx context.Context, inst *store.ChannelInstanceD titles = nil } out := make([]GroupOption, 0, len(groups)) + seen := make(map[string]int, len(groups)) for _, group := range groups { if group.ChannelName != inst.Name || group.HistoryKey == "" { continue } + seen[group.HistoryKey] = len(out) out = append(out, GroupOption{ ChannelName: group.ChannelName, HistoryKey: group.HistoryKey, @@ -137,6 +141,40 @@ func (s *Service) GroupOptions(ctx context.Context, inst *store.ChannelInstanceD Excluded: contains(cfg.ExcludeHistoryKeys, group.HistoryKey) || contains(cfg.ExcludeHistoryKeys, group.ParentHistoryKey), }) } + if s.Contacts == nil { + return out, nil + } + contacts, err := s.Contacts.ListContacts(ctx, store.ContactListOpts{ + ChannelType: channels.TypeDiscord, + ChannelInstance: inst.Name, + ContactType: "group", + Limit: 2000, + }) + if err != nil { + return nil, err + } + for _, contact := range contacts { + if contact.SenderID == "" { + continue + } + title := "" + if contact.DisplayName != nil { + title = *contact.DisplayName + } + if index, ok := seen[contact.SenderID]; ok { + if out[index].GroupTitle == "" { + out[index].GroupTitle = title + } + continue + } + out = append(out, GroupOption{ + ChannelName: inst.Name, + HistoryKey: contact.SenderID, + GroupTitle: title, + LastActivity: contact.LastSeenAt, + Excluded: contains(cfg.ExcludeHistoryKeys, contact.SenderID), + }) + } return out, nil } diff --git a/internal/channels/channel.go b/internal/channels/channel.go index d5519a0b..fcb4f8b9 100644 --- a/internal/channels/channel.go +++ b/internal/channels/channel.go @@ -744,12 +744,43 @@ type GroupTitleProvider interface { ResolveGroupTitle(ctx context.Context, chatID string) (string, error) } +// MetadataRefreshFailure records one group whose presentation metadata could +// not be refreshed. +type MetadataRefreshFailure struct { + ChannelID string `json:"channel_id"` + Source string `json:"source"` + Reason string `json:"reason"` +} + +// MetadataRefreshReport describes a manual channel metadata refresh. It keeps +// stable IDs intact while making refresh coverage and failures observable. +type MetadataRefreshReport struct { + OK bool `json:"ok"` + GroupsRefreshed int `json:"groups_refreshed"` + UsersRefreshed int `json:"users_refreshed"` + ContactTargets int `json:"contact_targets"` + PendingMessageTargets int `json:"pending_message_targets"` + LiveTargets int `json:"live_targets"` + DirectLookupAttempts int `json:"direct_lookup_attempts"` + DirectLookupResolved int `json:"direct_lookup_resolved"` + Errors []string `json:"errors,omitempty"` + Failures []MetadataRefreshFailure `json:"failures,omitempty"` +} + // GroupTitlesProvider is optionally implemented by channels that can resolve // multiple platform group/channel IDs in one best-effort operation. type GroupTitlesProvider interface { ResolveGroupTitles(ctx context.Context, chatIDs []string) (map[string]string, error) } +// GroupDisplayTitleProvider optionally resolves a presentation title for a +// group/channel ID. Unlike GroupTitleProvider, the result may include a +// platform hierarchy (for example, a Discord thread and its parent channel). +// Routing and persistence must continue to use the stable chat ID. +type GroupDisplayTitleProvider interface { + ResolveGroupDisplayTitle(ctx context.Context, chatID string) (string, error) +} + // TelegramManagerRequest describes a whitelisted Telegram Bot API management // action requested by the agent-facing telegram_manager tool. type TelegramManagerRequest struct { diff --git a/internal/channels/discord/contact_refresh.go b/internal/channels/discord/contact_refresh.go index ff99edf5..428131d2 100644 --- a/internal/channels/discord/contact_refresh.go +++ b/internal/channels/discord/contact_refresh.go @@ -4,6 +4,8 @@ import ( "context" "fmt" "log/slog" + "sort" + "strings" "time" "github.com/bwmarrin/discordgo" @@ -15,16 +17,22 @@ import ( const contactRefreshInterval = time.Hour +const contactRefreshPageSize = 100 + +type contactRefreshResult struct { + report channels.MetadataRefreshReport +} + // RefreshContactCache forces a Discord metadata refresh into channel_contacts. -func (c *Channel) RefreshContactCache(ctx context.Context) error { +func (c *Channel) RefreshContactCache(ctx context.Context) (channels.MetadataRefreshReport, error) { if c == nil || c.session == nil { - return fmt.Errorf("discord session unavailable") + return channels.MetadataRefreshReport{}, fmt.Errorf("discord session unavailable") } if c.ContactCollector() == nil { - return fmt.Errorf("contact collector unavailable") + return channels.MetadataRefreshReport{}, fmt.Errorf("contact collector unavailable") } - c.refreshContactCache(ctx) - return nil + result := c.refreshContactCache(ctx) + return result.report, nil } func (c *Channel) runContactRefreshLoop(ctx context.Context) { @@ -45,21 +53,39 @@ func (c *Channel) runContactRefreshLoop(ctx context.Context) { } } -func (c *Channel) refreshContactCache(ctx context.Context) { +func (c *Channel) refreshContactCache(ctx context.Context) contactRefreshResult { + result := contactRefreshResult{} if c == nil || c.session == nil { - return + return result } cc := c.ContactCollector() if cc == nil { - return + return result } cacheCtx := ctx if tenantID := c.TenantID(); tenantID != uuid.Nil { cacheCtx = store.WithTenantID(ctx, tenantID) } + contactGroupIDs, err := c.storedGroupIDs(cacheCtx, cc) + if err != nil { + result.report.Errors = append(result.report.Errors, "list contact targets: "+err.Error()) + } + result.report.ContactTargets = len(contactGroupIDs) + pendingGroupIDs, err := c.pendingHistoryGroupIDs(cacheCtx) + if err != nil { + result.report.Errors = append(result.report.Errors, "list pending-message targets: "+err.Error()) + } + result.report.PendingMessageTargets = len(pendingGroupIDs) + targets := mergeRefreshTargets(contactGroupIDs, pendingGroupIDs) count := 0 seen := make(map[string]struct{}) + knownChannels := make(map[string]*discordgo.Channel) + for _, ch := range c.stateChannels() { + if ch != nil && ch.ID != "" { + knownChannels[ch.ID] = ch + } + } userCount := 0 seenUsers := make(map[string]struct{}) upsert := func(ch *discordgo.Channel) { @@ -74,7 +100,9 @@ func (c *Channel) refreshContactCache(ctx context.Context) { return } seen[ch.ID] = struct{}{} - cc.RefreshContact(cacheCtx, c.Type(), c.Name(), ch.ID, "", title, "", "group", "group", "", "") + cc.RefreshContactWithMetadata(cacheCtx, c.Type(), c.Name(), ch.ID, "", title, "", "group", "group", "", "", discordContactMetadata(ch, func(id string) *discordgo.Channel { + return knownChannels[id] + })) count++ } upsertMember := func(member *discordgo.Member) { @@ -94,16 +122,13 @@ func (c *Channel) refreshContactCache(ctx context.Context) { userCount++ } - for _, ch := range c.stateChannels() { - upsert(ch) - } for _, member := range c.stateMembers() { upsertMember(member) } for _, guildID := range c.guildIDs() { select { case <-ctx.Done(): - return + return result default: } lookupCtx, cancel := context.WithTimeout(ctx, 10*time.Second) @@ -113,6 +138,11 @@ func (c *Channel) refreshContactCache(ctx context.Context) { slog.Debug("discord contact cache channel sync failed", "channel", c.Name(), "guild_id", guildID, "error", err) continue } + for _, ch := range guildChannels { + if ch != nil && ch.ID != "" { + knownChannels[ch.ID] = ch + } + } for _, ch := range guildChannels { upsert(ch) } @@ -125,14 +155,143 @@ func (c *Channel) refreshContactCache(ctx context.Context) { continue } if activeThreads != nil { + for _, ch := range activeThreads.Threads { + if ch != nil && ch.ID != "" { + knownChannels[ch.ID] = ch + } + } for _, ch := range activeThreads.Threads { upsert(ch) } } } - if count > 0 || userCount > 0 { - slog.Debug("discord contact cache refreshed", "channel", c.Name(), "groups", count, "users", userCount) + for _, ch := range knownChannels { + upsert(ch) } + for channelID, source := range targets { + if _, ok := seen[channelID]; ok { + result.report.LiveTargets++ + continue + } + result.report.DirectLookupAttempts++ + lookupCtx, cancel := context.WithTimeout(ctx, 10*time.Second) + ch, err := c.session.Channel(channelID, discordgo.WithContext(lookupCtx)) + cancel() + if err != nil { + slog.Debug("discord stored group title lookup failed", "channel", c.Name(), "channel_id", channelID, "error", err) + result.report.Failures = append(result.report.Failures, channels.MetadataRefreshFailure{ChannelID: channelID, Source: source, Reason: err.Error()}) + slog.Warn("discord metadata refresh group lookup failed", "channel", c.Name(), "channel_id", channelID, "source", source, "error", err) + continue + } + if ch == nil || ch.ID == "" { + result.report.Failures = append(result.report.Failures, channels.MetadataRefreshFailure{ChannelID: channelID, Source: source, Reason: "Discord returned no channel metadata"}) + continue + } + if channels.SanitizeDisplayName(ch.Name) == "" { + result.report.Failures = append(result.report.Failures, channels.MetadataRefreshFailure{ChannelID: channelID, Source: source, Reason: "Discord returned an empty channel name"}) + continue + } + knownChannels[ch.ID] = ch + parentResolved := true + if ch.IsThread() && ch.ParentID != "" && knownChannels[ch.ParentID] == nil { + parentCtx, parentCancel := context.WithTimeout(ctx, 10*time.Second) + parent, parentErr := c.session.Channel(ch.ParentID, discordgo.WithContext(parentCtx)) + parentCancel() + if parentErr != nil { + slog.Debug("discord stored thread parent lookup failed", "channel", c.Name(), "channel_id", ch.ID, "parent_id", ch.ParentID, "error", parentErr) + parentResolved = false + result.report.Failures = append(result.report.Failures, channels.MetadataRefreshFailure{ChannelID: channelID, Source: source, Reason: "parent lookup: " + parentErr.Error()}) + slog.Warn("discord metadata refresh thread parent lookup failed", "channel", c.Name(), "channel_id", ch.ID, "parent_id", ch.ParentID, "source", source, "error", parentErr) + } else if parent != nil && parent.ID != "" { + knownChannels[parent.ID] = parent + upsert(parent) + } else { + parentResolved = false + result.report.Failures = append(result.report.Failures, channels.MetadataRefreshFailure{ChannelID: channelID, Source: source, Reason: "Discord returned no parent metadata"}) + } + } + upsert(ch) + if parentResolved { + result.report.DirectLookupResolved++ + } + } + result.report.GroupsRefreshed = count + result.report.UsersRefreshed = userCount + result.report.OK = len(result.report.Errors) == 0 && len(result.report.Failures) == 0 + slog.Info("discord contact cache refreshed", "channel", c.Name(), "groups", count, "users", userCount, "contact_targets", result.report.ContactTargets, "pending_message_targets", result.report.PendingMessageTargets, "live_targets", result.report.LiveTargets, "direct_lookup_attempts", result.report.DirectLookupAttempts, "direct_lookup_resolved", result.report.DirectLookupResolved, "failures", len(result.report.Failures), "errors", len(result.report.Errors)) + return result +} + +func (c *Channel) pendingHistoryGroupIDs(ctx context.Context) ([]string, error) { + if c == nil || c.GroupHistory() == nil { + return nil, nil + } + return c.GroupHistory().PersistedGroupIDs(ctx) +} + +func mergeRefreshTargets(contactIDs, pendingIDs []string) map[string]string { + sources := make(map[string]map[string]struct{}, len(contactIDs)+len(pendingIDs)) + add := func(ids []string, source string) { + for _, id := range ids { + if id == "" { + continue + } + if sources[id] == nil { + sources[id] = make(map[string]struct{}) + } + sources[id][source] = struct{}{} + } + } + add(contactIDs, "contacts") + add(pendingIDs, "pending_messages") + targets := make(map[string]string, len(sources)) + for id, set := range sources { + parts := make([]string, 0, len(set)) + for source := range set { + parts = append(parts, source) + } + sort.Strings(parts) + targets[id] = strings.Join(parts, ",") + } + return targets +} + +func (c *Channel) storedGroupIDs(ctx context.Context, cc *store.ContactCollector) ([]string, error) { + if c == nil || cc == nil { + return nil, nil + } + seen := make(map[string]struct{}) + var ids []string + snapshotAt := time.Now().UTC() + for offset := 0; ; offset += contactRefreshPageSize { + contacts, err := cc.ListContacts(ctx, store.ContactListOpts{ + ChannelType: c.Type(), + ChannelInstance: c.Name(), + ContactType: "group", + SnapshotAt: &snapshotAt, + OrderByFirstSeen: true, + Limit: contactRefreshPageSize, + Offset: offset, + }) + if err != nil { + slog.Warn("discord stored group list failed", "channel", c.Name(), "error", err) + return ids, err + } + for _, contact := range contacts { + if contact.SenderID == "" { + continue + } + if _, ok := seen[contact.SenderID]; ok { + continue + } + seen[contact.SenderID] = struct{}{} + ids = append(ids, contact.SenderID) + } + if len(contacts) < contactRefreshPageSize { + break + } + } + return ids, nil } func (c *Channel) guildIDs() []string { diff --git a/internal/channels/discord/contact_refresh_test.go b/internal/channels/discord/contact_refresh_test.go index e1fa96b3..569a9923 100644 --- a/internal/channels/discord/contact_refresh_test.go +++ b/internal/channels/discord/contact_refresh_test.go @@ -2,9 +2,12 @@ package discord import ( "context" + "fmt" "net/http" "net/http/httptest" + "strings" "testing" + "time" "github.com/bwmarrin/discordgo" "github.com/google/uuid" @@ -17,6 +20,48 @@ import ( type contactRefreshStore struct { contacts map[string]string + metadata map[string]map[string]string + existing []store.ChannelContact + listErr error +} + +type pendingGroupsStore struct { + groups []store.PendingMessageGroup +} + +func (s *pendingGroupsStore) AppendBatch(context.Context, []store.PendingMessage) error { return nil } +func (s *pendingGroupsStore) ListByKey(context.Context, string, string) ([]store.PendingMessage, error) { + return nil, nil +} +func (s *pendingGroupsStore) DeleteByKey(context.Context, string, string) error { return nil } +func (s *pendingGroupsStore) Compact(context.Context, []uuid.UUID, *store.PendingMessage) error { + return nil +} +func (s *pendingGroupsStore) DeleteStale(context.Context, time.Duration) (int64, error) { + return 0, nil +} +func (s *pendingGroupsStore) ListGroups(context.Context) ([]store.PendingMessageGroup, error) { + return s.groups, nil +} +func (s *pendingGroupsStore) CountAll(context.Context) (int64, error) { return 0, nil } +func (s *pendingGroupsStore) CountByKey(context.Context, string, string) (int, error) { + return 0, nil +} +func (s *pendingGroupsStore) ResolveGroupTitles(context.Context, []store.PendingMessageGroup) (map[string]string, error) { + return nil, nil +} + +func (s *contactRefreshStore) UpsertContactWithMetadata(ctx context.Context, channelType, channelInstance, senderID, userID, displayName, username, peerKind, contactType, threadID, threadType string, metadata map[string]string) error { + if err := s.UpsertContact(ctx, channelType, channelInstance, senderID, userID, displayName, username, peerKind, contactType, threadID, threadType); err != nil { + return err + } + if len(metadata) > 0 { + if s.metadata == nil { + s.metadata = make(map[string]map[string]string) + } + s.metadata[channelInstance+":"+contactType+":"+senderID] = metadata + } + return nil } func (s *contactRefreshStore) UpsertContact(_ context.Context, _ string, channelInstance string, senderID string, _ string, displayName string, _ string, _ string, contactType string, _ string, _ string) error { @@ -27,8 +72,31 @@ func (s *contactRefreshStore) UpsertContact(_ context.Context, _ string, channel return nil } -func (s *contactRefreshStore) ListContacts(context.Context, store.ContactListOpts) ([]store.ChannelContact, error) { - return nil, nil +func (s *contactRefreshStore) ListContacts(_ context.Context, opts store.ContactListOpts) ([]store.ChannelContact, error) { + if s.listErr != nil { + return nil, s.listErr + } + var filtered []store.ChannelContact + for _, contact := range s.existing { + if opts.ChannelType != "" && contact.ChannelType != opts.ChannelType { + continue + } + if opts.ChannelInstance != "" && (contact.ChannelInstance == nil || *contact.ChannelInstance != opts.ChannelInstance) { + continue + } + if opts.ContactType != "" && contact.ContactType != opts.ContactType { + continue + } + filtered = append(filtered, contact) + } + if opts.Offset >= len(filtered) { + return nil, nil + } + contacts := filtered[opts.Offset:] + if opts.Limit > 0 && len(contacts) > opts.Limit { + contacts = contacts[:opts.Limit] + } + return contacts, nil } func (s *contactRefreshStore) CountContacts(context.Context, store.ContactListOpts) (int, error) { return 0, nil @@ -65,7 +133,7 @@ func TestRefreshContactCacheStoresDiscordChannelTitles(t *testing.T) { {"id":"category-1","name":"operations","type":4} ]`)) case "/guilds/guild-1/threads/active": - _, _ = w.Write([]byte(`{"threads":[{"id":"thread-1","name":"launch-thread","type":11}],"members":[],"has_more":false}`)) + _, _ = w.Write([]byte(`{"threads":[{"id":"thread-1","name":"launch-thread","parent_id":"parent-1","type":11}],"members":[],"has_more":false}`)) default: t.Fatalf("unexpected Discord API path: %s", r.URL.Path) } @@ -122,4 +190,298 @@ func TestRefreshContactCacheStoresDiscordChannelTitles(t *testing.T) { t.Fatalf("contact %s = %q, want %q; all=%#v", key, contactStore.contacts[key], title, contactStore.contacts) } } + if got := contactStore.metadata["discord-main:group:thread-1"][contactDisplayTitleMetadataKey]; got != "launch-thread / product-planning" { + t.Fatalf("thread display metadata = %q, want qualified title", got) + } +} + +func TestRefreshContactCacheBackfillsStoredArchivedGroup(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + switch r.URL.Path { + case "/guilds/guild-1/channels": + _, _ = w.Write([]byte(`[]`)) + case "/guilds/guild-1/threads/active": + _, _ = w.Write([]byte(`{"threads":[],"members":[],"has_more":false}`)) + case "/channels/archived-thread": + _, _ = w.Write([]byte(`{"id":"archived-thread","name":"release-notes","parent_id":"parent-1","type":11}`)) + case "/channels/parent-1": + _, _ = w.Write([]byte(`{"id":"parent-1","name":"announcements","type":0}`)) + default: + t.Fatalf("unexpected Discord API path: %s", r.URL.Path) + } + })) + defer server.Close() + + prevGuildChannels := discordgo.EndpointGuildChannels + prevGuildActiveThreads := discordgo.EndpointGuildActiveThreads + prevChannel := discordgo.EndpointChannel + discordgo.EndpointGuildChannels = func(gID string) string { return server.URL + "/guilds/" + gID + "/channels" } + discordgo.EndpointGuildActiveThreads = func(gID string) string { return server.URL + "/guilds/" + gID + "/threads/active" } + discordgo.EndpointChannel = func(cID string) string { return server.URL + "/channels/" + cID } + t.Cleanup(func() { + discordgo.EndpointGuildChannels = prevGuildChannels + discordgo.EndpointGuildActiveThreads = prevGuildActiveThreads + discordgo.EndpointChannel = prevChannel + }) + + session, err := discordgo.New("Bot test-token") + if err != nil { + t.Fatalf("discordgo.New: %v", err) + } + session.Client = server.Client() + session.State = discordgo.NewState() + if err := session.State.GuildAdd(&discordgo.Guild{ID: "guild-1"}); err != nil { + t.Fatalf("GuildAdd: %v", err) + } + + instance := "discord-main" + contactStore := &contactRefreshStore{existing: []store.ChannelContact{{ + ChannelType: channels.TypeDiscord, + ChannelInstance: &instance, + SenderID: "archived-thread", + ContactType: "group", + }}} + ch := &Channel{BaseChannel: channels.NewBaseChannel(channels.TypeDiscord, bus.New(), nil), session: session} + ch.SetName(instance) + ch.SetType(channels.TypeDiscord) + ch.SetContactCollector(store.NewContactCollector(contactStore, cache.NewInMemoryCache[bool]())) + + ch.refreshContactCache(context.Background()) + + if got := contactStore.contacts["discord-main:group:archived-thread"]; got != "release-notes" { + t.Fatalf("archived group title = %q, want release-notes", got) + } + if got := contactStore.metadata["discord-main:group:archived-thread"][contactDisplayTitleMetadataKey]; got != "release-notes / announcements" { + t.Fatalf("archived group display metadata = %q, want qualified title", got) + } +} + +func TestRefreshContactCacheBackfillsPendingOnlyGroup(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path != "/channels/pending-only" { + t.Fatalf("unexpected Discord API path: %s", r.URL.Path) + } + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(`{"id":"pending-only","name":"history-only-thread","type":11}`)) + })) + defer server.Close() + + prevChannel := discordgo.EndpointChannel + discordgo.EndpointChannel = func(cID string) string { return server.URL + "/channels/" + cID } + t.Cleanup(func() { discordgo.EndpointChannel = prevChannel }) + + session, err := discordgo.New("Bot test-token") + if err != nil { + t.Fatalf("discordgo.New: %v", err) + } + session.Client = server.Client() + instance := "discord-main" + contactStore := &contactRefreshStore{} + pendingStore := &pendingGroupsStore{groups: []store.PendingMessageGroup{{ + ChannelName: instance, + HistoryKey: "pending-only", + }}} + base := channels.NewBaseChannel(channels.TypeDiscord, bus.New(), nil) + base.SetGroupHistory(channels.MakeHistory(instance, pendingStore, uuid.Nil)) + ch := &Channel{BaseChannel: base, session: session} + ch.SetName(instance) + ch.SetType(channels.TypeDiscord) + ch.SetContactCollector(store.NewContactCollector(contactStore, cache.NewInMemoryCache[bool]())) + + report, err := ch.RefreshContactCache(context.Background()) + if err != nil { + t.Fatalf("RefreshContactCache() error = %v", err) + } + if !report.OK || report.PendingMessageTargets != 1 || report.DirectLookupResolved != 1 { + t.Fatalf("report = %+v, want resolved pending-only target", report) + } + if got := contactStore.contacts[instance+":group:pending-only"]; got != "history-only-thread" { + t.Fatalf("pending-only group title = %q, want history-only-thread", got) + } +} + +func TestRefreshContactCacheBackfillsEveryStoredGroupPage(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + switch r.URL.Path { + case "/guilds/guild-1/channels": + _, _ = w.Write([]byte(`[]`)) + case "/guilds/guild-1/threads/active": + _, _ = w.Write([]byte(`{"threads":[],"members":[],"has_more":false}`)) + default: + if !strings.HasPrefix(r.URL.Path, "/channels/archived-") { + t.Fatalf("unexpected Discord API path: %s", r.URL.Path) + } + id := strings.TrimPrefix(r.URL.Path, "/channels/") + _, _ = fmt.Fprintf(w, `{"id":%q,"name":%q,"type":0}`, id, "title-"+id) + } + })) + defer server.Close() + + prevGuildChannels := discordgo.EndpointGuildChannels + prevGuildActiveThreads := discordgo.EndpointGuildActiveThreads + prevChannel := discordgo.EndpointChannel + discordgo.EndpointGuildChannels = func(gID string) string { return server.URL + "/guilds/" + gID + "/channels" } + discordgo.EndpointGuildActiveThreads = func(gID string) string { return server.URL + "/guilds/" + gID + "/threads/active" } + discordgo.EndpointChannel = func(cID string) string { return server.URL + "/channels/" + cID } + t.Cleanup(func() { + discordgo.EndpointGuildChannels = prevGuildChannels + discordgo.EndpointGuildActiveThreads = prevGuildActiveThreads + discordgo.EndpointChannel = prevChannel + }) + + session, err := discordgo.New("Bot test-token") + if err != nil { + t.Fatalf("discordgo.New: %v", err) + } + session.Client = server.Client() + session.State = discordgo.NewState() + if err := session.State.GuildAdd(&discordgo.Guild{ID: "guild-1"}); err != nil { + t.Fatalf("GuildAdd: %v", err) + } + + instance := "discord-main" + contactStore := &contactRefreshStore{} + for i := 0; i <= contactRefreshPageSize; i++ { + id := fmt.Sprintf("archived-%03d", i) + contactStore.existing = append(contactStore.existing, store.ChannelContact{ + ChannelType: channels.TypeDiscord, + ChannelInstance: &instance, + SenderID: id, + ContactType: "group", + }) + } + ch := &Channel{BaseChannel: channels.NewBaseChannel(channels.TypeDiscord, bus.New(), nil), session: session} + ch.SetName(instance) + ch.SetType(channels.TypeDiscord) + ch.SetContactCollector(store.NewContactCollector(contactStore, cache.NewInMemoryCache[bool]())) + + if _, err := ch.RefreshContactCache(context.Background()); err != nil { + t.Fatalf("RefreshContactCache() error = %v", err) + } + for i := 0; i <= contactRefreshPageSize; i++ { + id := fmt.Sprintf("archived-%03d", i) + if got := contactStore.contacts[instance+":group:"+id]; got != "title-"+id { + t.Fatalf("stored group %s title = %q, want %q", id, got, "title-"+id) + } + } +} + +func TestRefreshContactCacheReportsUnreadableStoredGroup(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + http.Error(w, "missing access", http.StatusForbidden) + })) + defer server.Close() + + prevChannel := discordgo.EndpointChannel + discordgo.EndpointChannel = func(cID string) string { return server.URL + "/channels/" + cID } + t.Cleanup(func() { discordgo.EndpointChannel = prevChannel }) + + session, err := discordgo.New("Bot test-token") + if err != nil { + t.Fatalf("discordgo.New: %v", err) + } + session.Client = server.Client() + instance := "discord-main" + contactStore := &contactRefreshStore{existing: []store.ChannelContact{{ + ChannelType: channels.TypeDiscord, + ChannelInstance: &instance, + SenderID: "unreadable-group", + ContactType: "group", + }}} + ch := &Channel{BaseChannel: channels.NewBaseChannel(channels.TypeDiscord, bus.New(), nil), session: session} + ch.SetName(instance) + ch.SetType(channels.TypeDiscord) + ch.SetContactCollector(store.NewContactCollector(contactStore, cache.NewInMemoryCache[bool]())) + + report, err := ch.RefreshContactCache(context.Background()) + if err != nil { + t.Fatalf("RefreshContactCache() error = %v", err) + } + if report.OK || len(report.Failures) != 1 || report.Failures[0].ChannelID != "unreadable-group" { + t.Fatalf("report = %+v, want unreadable stored group failure", report) + } +} + +func TestRefreshContactCacheReportsStoredGroupListFailure(t *testing.T) { + session, err := discordgo.New("Bot test-token") + if err != nil { + t.Fatalf("discordgo.New: %v", err) + } + instance := "discord-main" + contactStore := &contactRefreshStore{listErr: fmt.Errorf("database unavailable")} + ch := &Channel{BaseChannel: channels.NewBaseChannel(channels.TypeDiscord, bus.New(), nil), session: session} + ch.SetName(instance) + ch.SetType(channels.TypeDiscord) + ch.SetContactCollector(store.NewContactCollector(contactStore, cache.NewInMemoryCache[bool]())) + + report, err := ch.RefreshContactCache(context.Background()) + if err != nil { + t.Fatalf("RefreshContactCache() error = %v", err) + } + if report.OK || len(report.Errors) != 1 { + t.Fatalf("report = %+v, want stored group list error", report) + } +} + +func TestRefreshContactCacheReportsMissingStoredThreadParent(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + switch r.URL.Path { + case "/channels/thread-1": + _, _ = w.Write([]byte(`{"id":"thread-1","name":"release-notes","parent_id":"parent-1","type":11}`)) + case "/channels/parent-1": + http.Error(w, "missing access", http.StatusForbidden) + default: + t.Fatalf("unexpected Discord API path: %s", r.URL.Path) + } + })) + defer server.Close() + + prevChannel := discordgo.EndpointChannel + discordgo.EndpointChannel = func(cID string) string { return server.URL + "/channels/" + cID } + t.Cleanup(func() { discordgo.EndpointChannel = prevChannel }) + + session, err := discordgo.New("Bot test-token") + if err != nil { + t.Fatalf("discordgo.New: %v", err) + } + session.Client = server.Client() + instance := "discord-main" + contactStore := &contactRefreshStore{existing: []store.ChannelContact{{ + ChannelType: channels.TypeDiscord, + ChannelInstance: &instance, + SenderID: "thread-1", + ContactType: "group", + }}} + ch := &Channel{BaseChannel: channels.NewBaseChannel(channels.TypeDiscord, bus.New(), nil), session: session} + ch.SetName(instance) + ch.SetType(channels.TypeDiscord) + ch.SetContactCollector(store.NewContactCollector(contactStore, cache.NewInMemoryCache[bool]())) + + report, err := ch.RefreshContactCache(context.Background()) + if err != nil { + t.Fatalf("RefreshContactCache() error = %v", err) + } + if report.OK || len(report.Failures) != 1 || report.Failures[0].ChannelID != "thread-1" { + t.Fatalf("report = %+v, want missing thread parent failure", report) + } +} + +func TestSetContactCollectorStartsRefreshAfterChannelStartup(t *testing.T) { + session, err := discordgo.New("Bot test-token") + if err != nil { + t.Fatalf("discordgo.New: %v", err) + } + ch := &Channel{ + BaseChannel: channels.NewBaseChannel(channels.TypeDiscord, bus.New(), nil), + session: session, + } + ch.SetRunning(true) + ch.SetContactCollector(store.NewContactCollector(&contactRefreshStore{}, cache.NewInMemoryCache[bool]())) + if ch.contactRefreshCancel == nil { + t.Fatal("contact refresh loop was not started after collector wiring") + } + ch.contactRefreshCancel() } diff --git a/internal/channels/discord/discord.go b/internal/channels/discord/discord.go index 4f9fc01a..e4de8aa4 100644 --- a/internal/channels/discord/discord.go +++ b/internal/channels/discord/discord.go @@ -32,6 +32,7 @@ type Channel struct { agentStore store.AgentStore // for agent key lookup (nil = writer commands disabled) configPermStore store.ConfigPermissionStore // for group file writer management (nil = writer commands disabled) audioMgr *audio.Manager // unified STT via audio.Manager (nil = no STT) + contactRefreshMu sync.Mutex contactRefreshCancel context.CancelFunc // pairingService, pairingDebounce, approvedGroups, groupHistory, historyLimit, requireMention // are inherited from channels.BaseChannel. @@ -97,11 +98,46 @@ func (c *Channel) Start(ctx context.Context) error { c.SetRunning(true) slog.Info("discord bot connected", "username", user.Username, "id", user.ID) + c.startContactRefreshLoop(ctx) + + return nil +} + +// SetContactCollector starts the metadata refresh loop when the collector is +// wired after the channel was started. Gateway lifecycle wiring happens after +// StartAll, so relying solely on Start would leave the loop permanently idle. +func (c *Channel) SetContactCollector(cc *store.ContactCollector) { + c.BaseChannel.SetContactCollector(cc) + if cc != nil && c.IsRunning() { + c.startContactRefreshLoop(context.Background()) + } +} + +func (c *Channel) startContactRefreshLoop(ctx context.Context) { + if c == nil || c.ContactCollector() == nil { + return + } + c.contactRefreshMu.Lock() + defer c.contactRefreshMu.Unlock() + if c.contactRefreshCancel != nil { + return + } refreshCtx, cancel := context.WithCancel(ctx) c.contactRefreshCancel = cancel go c.runContactRefreshLoop(refreshCtx) +} - return nil +func (c *Channel) stopContactRefreshLoop() { + if c == nil { + return + } + c.contactRefreshMu.Lock() + cancel := c.contactRefreshCancel + c.contactRefreshCancel = nil + c.contactRefreshMu.Unlock() + if cancel != nil { + cancel() + } } // BlockReplyEnabled returns the per-channel block_reply override (nil = inherit gateway default). @@ -126,10 +162,7 @@ func (c *Channel) SetPendingHistoryTenantID(id uuid.UUID) { // Stop closes the Discord gateway connection. func (c *Channel) Stop(_ context.Context) error { - if c.contactRefreshCancel != nil { - c.contactRefreshCancel() - c.contactRefreshCancel = nil - } + c.stopContactRefreshLoop() c.GroupHistory().StopFlusher() slog.Info("stopping discord bot") c.SetRunning(false) diff --git a/internal/channels/discord/handler.go b/internal/channels/discord/handler.go index eef5cad4..6e80220e 100644 --- a/internal/channels/discord/handler.go +++ b/internal/channels/discord/handler.go @@ -120,7 +120,7 @@ func (c *Channel) handleMessage(_ *discordgo.Session, m *discordgo.MessageCreate // Process media: STT, document extraction, build tags var mediaFiles []bus.MediaFile if len(mediaList) > 0 { - var extraContent string + var extraContent strings.Builder for i := range mediaList { mi := &mediaList[i] @@ -152,7 +152,7 @@ func (c *Channel) handleMessage(_ *discordgo.Session, m *discordgo.MessageCreate if err != nil { slog.Warn("discord: document extraction failed", "file", mi.FileName, "error", err) } else if docContent != "" { - extraContent += "\n\n" + docContent + extraContent.WriteString("\n\n" + docContent) } } } @@ -176,8 +176,8 @@ func (c *Channel) handleMessage(_ *discordgo.Session, m *discordgo.MessageCreate } } - if extraContent != "" { - content += extraContent + if extraContent.String() != "" { + content += extraContent.String() } } @@ -211,7 +211,7 @@ func (c *Channel) handleMessage(_ *discordgo.Session, m *discordgo.MessageCreate // Collect contact even when bot is not mentioned (cache prevents DB spam). if cc := c.ContactCollector(); cc != nil { cc.EnsureContact(ctx, c.Type(), c.Name(), senderID, senderID, displayName, m.Author.Username, "group", "user", "", "") - cc.EnsureContact(ctx, c.Type(), c.Name(), channelID, "", c.resolveCachedChannelTitle(channelID), "", "group", "group", "", "") + cc.EnsureContactWithMetadata(ctx, c.Type(), c.Name(), channelID, "", c.resolveCachedChannelTitle(channelID), "", "group", "group", "", "", c.cachedContactMetadata(channelID)) } slog.Debug("discord group message recorded (no mention)", @@ -302,7 +302,7 @@ func (c *Channel) handleMessage(_ *discordgo.Session, m *discordgo.MessageCreate "placeholder_key": m.ID, // keyed by inbound message ID for placeholder lookup } if !isDM { - if title := c.resolveCachedChannelTitle(channelID); title != "" { + if title := c.resolveCachedGroupDisplayTitle(channelID); title != "" { metadata[tools.MetaChatTitle] = title } } @@ -313,7 +313,7 @@ func (c *Channel) handleMessage(_ *discordgo.Session, m *discordgo.MessageCreate if cc := c.ContactCollector(); cc != nil { cc.EnsureContact(ctx, c.Type(), c.Name(), senderID, senderID, displayName, m.Author.Username, peerKind, "user", "", "") if peerKind == "group" { - cc.EnsureContact(ctx, c.Type(), c.Name(), channelID, "", c.resolveCachedChannelTitle(channelID), "", "group", "group", "", "") + cc.EnsureContactWithMetadata(ctx, c.Type(), c.Name(), channelID, "", c.resolveCachedChannelTitle(channelID), "", "group", "group", "", "", c.cachedContactMetadata(channelID)) } } @@ -497,6 +497,20 @@ func (c *Channel) resolveCachedChannelTitle(channelID string) string { return channels.SanitizeDisplayName(ch.Name) } +func (c *Channel) cachedContactMetadata(channelID string) map[string]string { + if c == nil || c.session == nil || c.session.State == nil || channelID == "" { + return nil + } + ch, err := c.session.State.Channel(channelID) + if err != nil || ch == nil { + return nil + } + return discordContactMetadata(ch, func(id string) *discordgo.Channel { + parent, _ := c.session.State.Channel(id) + return parent + }) +} + func (c *Channel) ResolveGroupTitle(_ context.Context, channelID string) (string, error) { if title := c.resolveCachedChannelTitle(channelID); title != "" { return title, nil diff --git a/internal/channels/discord/handler_test.go b/internal/channels/discord/handler_test.go index 2e6fe63c..739f4674 100644 --- a/internal/channels/discord/handler_test.go +++ b/internal/channels/discord/handler_test.go @@ -91,6 +91,36 @@ func TestResolveCachedChannelTitle(t *testing.T) { } } +func TestResolveGroupDisplayTitleQualifiesDiscordThread(t *testing.T) { + session, err := discordgo.New("Bot test-token") + if err != nil { + t.Fatalf("discordgo.New() error = %v", err) + } + session.State = discordgo.NewState() + if err := session.State.GuildAdd(&discordgo.Guild{ID: "guild-1"}); err != nil { + t.Fatalf("GuildAdd() error = %v", err) + } + for _, channel := range []*discordgo.Channel{ + {ID: "parent-1", GuildID: "guild-1", Name: "product-planning", Type: discordgo.ChannelTypeGuildText}, + {ID: "thread-1", GuildID: "guild-1", Name: "launch-thread", ParentID: "parent-1", Type: discordgo.ChannelTypeGuildPublicThread}, + } { + if err := session.State.ChannelAdd(channel); err != nil { + t.Fatalf("ChannelAdd(%s) error = %v", channel.ID, err) + } + } + + ch := &Channel{session: session} + if got := ch.resolveCachedGroupDisplayTitle("thread-1"); got != "launch-thread / product-planning" { + t.Fatalf("cached display title = %q, want qualified thread title", got) + } + if got := ch.resolveCachedGroupDisplayTitle("parent-1"); got != "product-planning" { + t.Fatalf("parent display title = %q, want raw parent title", got) + } + if got, err := ch.ResolveGroupDisplayTitle(context.Background(), "thread-1"); err != nil || got != "launch-thread / product-planning" { + t.Fatalf("ResolveGroupDisplayTitle() = %q, %v", got, err) + } +} + func TestResolveMemoryExtractionContextIncludesThreadParentAndCategory(t *testing.T) { session, err := discordgo.New("Bot test-token") if err != nil { diff --git a/internal/channels/discord/hierarchy.go b/internal/channels/discord/hierarchy.go new file mode 100644 index 00000000..e436b6e5 --- /dev/null +++ b/internal/channels/discord/hierarchy.go @@ -0,0 +1,67 @@ +package discord + +import ( + "context" + "strings" + + "github.com/bwmarrin/discordgo" + + "github.com/nextlevelbuilder/goclaw/internal/channels" +) + +const contactDisplayTitleMetadataKey = "display_title" + +// ResolveGroupDisplayTitle resolves a display-only Discord title. A thread is +// qualified with its parent channel so identical thread names remain distinct. +func (c *Channel) ResolveGroupDisplayTitle(ctx context.Context, channelID string) (string, error) { + ch := c.resolveDiscordChannel(ctx, channelID) + if ch == nil { + return "", nil + } + var parent *discordgo.Channel + if ch.IsThread() && ch.ParentID != "" { + parent = c.resolveDiscordChannel(ctx, ch.ParentID) + } + return discordDisplayTitle(ch, parent), nil +} + +func (c *Channel) resolveCachedGroupDisplayTitle(channelID string) string { + if c == nil || c.session == nil || c.session.State == nil || channelID == "" { + return "" + } + ch, err := c.session.State.Channel(channelID) + if err != nil || ch == nil { + return "" + } + var parent *discordgo.Channel + if ch.IsThread() && ch.ParentID != "" { + parent, _ = c.session.State.Channel(ch.ParentID) + } + return discordDisplayTitle(ch, parent) +} + +func discordDisplayTitle(ch, parent *discordgo.Channel) string { + if ch == nil { + return "" + } + name := channels.SanitizeDisplayName(ch.Name) + if name == "" || !ch.IsThread() || parent == nil { + return name + } + parentName := channels.SanitizeDisplayName(parent.Name) + if parentName == "" || parentName == name { + return name + } + return name + " / " + parentName +} + +func discordContactMetadata(ch *discordgo.Channel, lookup func(string) *discordgo.Channel) map[string]string { + if ch == nil || !ch.IsThread() || ch.ParentID == "" || lookup == nil { + return nil + } + displayTitle := discordDisplayTitle(ch, lookup(ch.ParentID)) + if displayTitle == "" || displayTitle == strings.TrimSpace(ch.Name) { + return nil + } + return map[string]string{contactDisplayTitleMetadataKey: displayTitle} +} diff --git a/internal/channels/history.go b/internal/channels/history.go index 69a0e3f1..db020348 100644 --- a/internal/channels/history.go +++ b/internal/channels/history.go @@ -103,6 +103,37 @@ func NewPersistentHistory(channelName string, s store.PendingMessageStore, tenan // IsPersistent returns true if this history is backed by a DB store. func (ph *PendingHistory) IsPersistent() bool { return ph.store != nil } +// PersistedGroupIDs returns the current channel's stored group and parent +// history keys. It lets platform metadata refreshes backfill titles for groups +// that have pending history but no channel-contact record. +func (ph *PendingHistory) PersistedGroupIDs(ctx context.Context) ([]string, error) { + if ph == nil || ph.store == nil || ph.channelName == "" { + return nil, nil + } + groups, err := ph.store.ListGroups(ctx) + if err != nil { + return nil, err + } + seen := make(map[string]struct{}) + var ids []string + for _, group := range groups { + if group.ChannelName != ph.channelName { + continue + } + for _, id := range []string{group.HistoryKey, group.ParentHistoryKey} { + if id == "" { + continue + } + if _, ok := seen[id]; ok { + continue + } + seen[id] = struct{}{} + ids = append(ids, id) + } + } + return ids, nil +} + // SetChannelName updates the channel identity used for DB persistence. // DB-backed channel instances are constructed with a platform type first and // renamed to the instance name before start. diff --git a/internal/channels/manager.go b/internal/channels/manager.go index bcec6766..8c20e2db 100644 --- a/internal/channels/manager.go +++ b/internal/channels/manager.go @@ -391,6 +391,23 @@ func (m *Manager) ResolveGroupTitles(ctx context.Context, channelName string, ch return gtp.ResolveGroupTitles(ctx, chatIDs) } +// ResolveGroupDisplayTitle resolves a presentation-only title for a group. +// It is deliberately separate from ResolveGroupTitle so callers that need a +// platform's raw title keep their existing contract. +func (m *Manager) ResolveGroupDisplayTitle(ctx context.Context, channelName, chatID string) (string, error) { + m.mu.RLock() + ch, ok := m.channels[channelName] + m.mu.RUnlock() + if !ok { + return "", fmt.Errorf("channel %q not found", channelName) + } + provider, ok := ch.(GroupDisplayTitleProvider) + if !ok { + return "", fmt.Errorf("channel %q does not support resolving group display titles", channelName) + } + return provider.ResolveGroupDisplayTitle(ctx, chatID) +} + // ManageTelegram delegates a whitelisted Telegram management action to a // Telegram-capable channel instance. func (m *Manager) ManageTelegram(ctx context.Context, channelName string, req TelegramManagerRequest) (TelegramManagerResult, error) { diff --git a/internal/heartbeat/ticker.go b/internal/heartbeat/ticker.go index 3ef88e6f..91948486 100644 --- a/internal/heartbeat/ticker.go +++ b/internal/heartbeat/ticker.go @@ -37,19 +37,23 @@ type TickerConfig struct { MsgBus EventPublisher Sched ActiveSessionChecker RunAgent func(ctx context.Context, req agent.RunRequest) <-chan scheduler.RunOutcome + // ResolveGroupContext optionally supplies platform context for a configured + // group delivery target without changing its stable route ID. + ResolveGroupContext func(ctx context.Context, channel, chatID string) (channelType, chatTitle string) } // Ticker polls for due heartbeats and runs them through the agent loop. type Ticker struct { - store store.HeartbeatStore - agents store.AgentStore - sessions store.SessionStore - providerStore store.ProviderStore - providerReg ProviderResolver - msgBus EventPublisher - sched ActiveSessionChecker - runAgent func(ctx context.Context, req agent.RunRequest) <-chan scheduler.RunOutcome - onEvent func(store.HeartbeatEvent) + store store.HeartbeatStore + agents store.AgentStore + sessions store.SessionStore + providerStore store.ProviderStore + providerReg ProviderResolver + msgBus EventPublisher + sched ActiveSessionChecker + runAgent func(ctx context.Context, req agent.RunRequest) <-chan scheduler.RunOutcome + resolveGroupContext func(ctx context.Context, channel, chatID string) (channelType, chatTitle string) + onEvent func(store.HeartbeatEvent) wakeCh chan uuid.UUID stopCh chan struct{} @@ -59,16 +63,17 @@ type Ticker struct { // NewTicker creates a new heartbeat ticker. func NewTicker(cfg TickerConfig) *Ticker { return &Ticker{ - store: cfg.Store, - agents: cfg.Agents, - sessions: cfg.Sessions, - providerStore: cfg.ProviderStore, - providerReg: cfg.ProviderReg, - msgBus: cfg.MsgBus, - sched: cfg.Sched, - runAgent: cfg.RunAgent, - wakeCh: make(chan uuid.UUID, 16), - stopCh: make(chan struct{}), + store: cfg.Store, + agents: cfg.Agents, + sessions: cfg.Sessions, + providerStore: cfg.ProviderStore, + providerReg: cfg.ProviderReg, + msgBus: cfg.MsgBus, + sched: cfg.Sched, + runAgent: cfg.RunAgent, + resolveGroupContext: cfg.ResolveGroupContext, + wakeCh: make(chan uuid.UUID, 16), + stopCh: make(chan struct{}), } } @@ -237,6 +242,10 @@ func (t *Ticker) runOne(ctx context.Context, hb store.AgentHeartbeat) { if hb.ChatID != nil { chatID = *hb.ChatID } + channelType, chatTitle := "", "" + if t.resolveGroupContext != nil && chatID != "" { + channelType, chatTitle = t.resolveGroupContext(ctx, channel, chatID) + } // [5] Run through agent loop via scheduler. var lastErr error @@ -272,7 +281,10 @@ func (t *Ticker) runOne(ctx context.Context, hb store.AgentHeartbeat) { SessionKey: sessionKey, Message: prompt, Channel: channel, + ChannelType: channelType, ChatID: chatID, + ChatTitle: chatTitle, + PeerKind: heartbeatPeerKind(chatTitle), RunID: fmt.Sprintf("heartbeat:%s", agentIDStr), Stream: false, ExtraSystemPrompt: extraSystem, @@ -331,6 +343,13 @@ func (t *Ticker) runOne(ctx context.Context, hb store.AgentHeartbeat) { t.finishRun(ctx, hb, sessionKey, agentKey, "ok", "", truncate(cleaned, maxSummaryLen), durationMS, inputTokens, outputTokens) } +func heartbeatPeerKind(chatTitle string) string { + if chatTitle != "" { + return "group" + } + return "" +} + func (t *Ticker) finishRun(ctx context.Context, hb store.AgentHeartbeat, sessionKey, agentKey, status, errMsg, summary string, durationMS, inputTokens, outputTokens int) { agentIDStr := hb.AgentID.String() now := time.Now() diff --git a/internal/http/channel_instance_metadata.go b/internal/http/channel_instance_metadata.go index f0af90fa..73b3fc3f 100644 --- a/internal/http/channel_instance_metadata.go +++ b/internal/http/channel_instance_metadata.go @@ -9,7 +9,7 @@ import ( ) type channelMetadataRefresher interface { - RefreshContactCache(ctx context.Context) error + RefreshContactCache(ctx context.Context) (channels.MetadataRefreshReport, error) } func (h *ChannelInstancesHandler) handleRefreshChannelMetadata(w http.ResponseWriter, r *http.Request) { @@ -42,10 +42,13 @@ func (h *ChannelInstancesHandler) handleRefreshChannelMetadata(w http.ResponseWr writeError(w, http.StatusConflict, protocol.ErrInvalidRequest, "Discord channel metadata refresh is not available") return } - if err := refresher.RefreshContactCache(r.Context()); err != nil { - writeError(w, http.StatusConflict, protocol.ErrInvalidRequest, err.Error()) + report, err := refresher.RefreshContactCache(r.Context()) + if err != nil { + report.OK = false + report.Errors = append(report.Errors, err.Error()) + writeJSON(w, http.StatusConflict, map[string]any{"ok": false, "report": report, "error": err.Error()}) return } emitAudit(h.msgBus, r, "channel_metadata.refresh_triggered", "channel_instance", inst.ID.String()) - writeJSON(w, http.StatusOK, map[string]any{"ok": true}) + writeJSON(w, http.StatusOK, map[string]any{"ok": report.OK, "report": report}) } diff --git a/internal/http/channel_instance_metadata_test.go b/internal/http/channel_instance_metadata_test.go index f6acdb44..eed2c46a 100644 --- a/internal/http/channel_instance_metadata_test.go +++ b/internal/http/channel_instance_metadata_test.go @@ -27,9 +27,9 @@ func (c *metadataRefreshChannel) IsRunning() bool { return true } func (c *metadataRefreshChannel) IsAllowed(string) bool { return true } -func (c *metadataRefreshChannel) RefreshContactCache(context.Context) error { +func (c *metadataRefreshChannel) RefreshContactCache(context.Context) (channels.MetadataRefreshReport, error) { c.called = true - return nil + return channels.MetadataRefreshReport{OK: true, GroupsRefreshed: 1}, nil } func TestRefreshChannelMetadataCallsDiscordRefresher(t *testing.T) { diff --git a/internal/http/channel_memory_extraction_test.go b/internal/http/channel_memory_extraction_test.go index 3de14c22..ab43c580 100644 --- a/internal/http/channel_memory_extraction_test.go +++ b/internal/http/channel_memory_extraction_test.go @@ -138,3 +138,35 @@ func TestMemoryExtractionGroupsUsesDBTitlesOnly(t *testing.T) { t.Fatalf("parent group title = %q, want product-planning", body.Groups[0].ParentGroupTitle) } } + +func TestPendingMessageGroupsIncludeParentTitle(t *testing.T) { + pending := &memoryExtractionPendingStore{ + groups: []store.PendingMessageGroup{{ + ChannelName: "discord-main", + HistoryKey: "thread-1", + ParentHistoryKey: "parent-1", + }}, + titles: map[string]string{ + "discord-main:thread-1": "launch-thread", + "discord-main:parent-1": "product-planning", + }, + } + h := NewPendingMessagesHandler(pending, nil, nil) + w := httptest.NewRecorder() + h.handleListGroups(w, httptest.NewRequest(http.MethodGet, "/v1/pending-messages", nil)) + if w.Code != http.StatusOK { + t.Fatalf("status = %d, want %d", w.Code, http.StatusOK) + } + var body struct { + Groups []store.PendingMessageGroup `json:"groups"` + } + if err := json.NewDecoder(w.Body).Decode(&body); err != nil { + t.Fatalf("decode response: %v", err) + } + if len(body.Groups) != 1 { + t.Fatalf("groups = %d, want 1", len(body.Groups)) + } + if got := body.Groups[0].ParentGroupTitle; got != "product-planning" { + t.Fatalf("parent group title = %q, want product-planning", got) + } +} diff --git a/internal/http/pending_messages.go b/internal/http/pending_messages.go index 3f144fa7..24bf7e58 100644 --- a/internal/http/pending_messages.go +++ b/internal/http/pending_messages.go @@ -65,12 +65,34 @@ func (h *PendingMessagesHandler) handleListGroups(w http.ResponseWriter, r *http return } - // Resolve group titles from session metadata (best-effort, non-blocking) - if titles, err := h.store.ResolveGroupTitles(r.Context(), groups); err == nil { + // Resolve both the current group and its parent title so a Discord thread + // can be rendered unambiguously without changing stable history keys. + lookupGroups := make([]store.PendingMessageGroup, 0, len(groups)*2) + seen := make(map[string]struct{}, len(groups)*2) + for _, group := range groups { + key := group.ChannelName + ":" + group.HistoryKey + if group.ChannelName != "" && group.HistoryKey != "" { + if _, ok := seen[key]; !ok { + seen[key] = struct{}{} + lookupGroups = append(lookupGroups, group) + } + } + if group.ParentHistoryKey != "" { + parentKey := group.ChannelName + ":" + group.ParentHistoryKey + if _, ok := seen[parentKey]; !ok { + seen[parentKey] = struct{}{} + lookupGroups = append(lookupGroups, store.PendingMessageGroup{ChannelName: group.ChannelName, HistoryKey: group.ParentHistoryKey}) + } + } + } + if titles, err := h.store.ResolveGroupTitles(r.Context(), lookupGroups); err == nil { for i := range groups { if t, ok := titles[groups[i].ChannelName+":"+groups[i].HistoryKey]; ok { groups[i].GroupTitle = t } + if groups[i].ParentHistoryKey != "" { + groups[i].ParentGroupTitle = titles[groups[i].ChannelName+":"+groups[i].ParentHistoryKey] + } } } diff --git a/internal/store/contact_collector.go b/internal/store/contact_collector.go index c7400203..1e21934e 100644 --- a/internal/store/contact_collector.go +++ b/internal/store/contact_collector.go @@ -2,6 +2,7 @@ package store import ( "context" + "errors" "log/slog" "time" @@ -73,7 +74,69 @@ func (c *ContactCollector) RefreshContact(ctx context.Context, channelType, chan } } +// EnsureContactWithMetadata is the metadata-aware counterpart of +// EnsureContact. It preserves the same hot-path dedup behavior and gracefully +// falls back for stores that only implement ContactStore. +func (c *ContactCollector) EnsureContactWithMetadata(ctx context.Context, channelType, channelInstance, senderID, userID, displayName, username, peerKind, contactType, threadID, threadType string, metadata map[string]string) { + tid := TenantIDFromContext(ctx) + key := tid.String() + ":" + channelType + ":" + channelInstance + ":" + senderID + ":" + threadID + if _, ok := c.seen.Get(ctx, key); ok { + return + } + if contactType == "" { + contactType = "user" + } + metadataStore, ok := c.store.(ContactMetadataStore) + if !ok { + c.EnsureContact(ctx, channelType, channelInstance, senderID, userID, displayName, username, peerKind, contactType, threadID, threadType) + return + } + if err := metadataStore.UpsertContactWithMetadata(ctx, channelType, channelInstance, senderID, userID, displayName, username, peerKind, contactType, threadID, threadType, metadata); err != nil { + slog.Warn("contact_collector.upsert_metadata_failed", + "error", err, + "tenant_id", tid, + "channel", channelType, + "instance", channelInstance, + "sender", senderID, + ) + return + } + c.seen.Set(ctx, key, true, contactSeenTTL) +} + +// RefreshContactWithMetadata persists presentation metadata without hot-path +// dedup, matching RefreshContact's refresh semantics. +func (c *ContactCollector) RefreshContactWithMetadata(ctx context.Context, channelType, channelInstance, senderID, userID, displayName, username, peerKind, contactType, threadID, threadType string, metadata map[string]string) { + if contactType == "" { + contactType = "user" + } + metadataStore, ok := c.store.(ContactMetadataStore) + if !ok { + c.RefreshContact(ctx, channelType, channelInstance, senderID, userID, displayName, username, peerKind, contactType, threadID, threadType) + return + } + if err := metadataStore.UpsertContactWithMetadata(ctx, channelType, channelInstance, senderID, userID, displayName, username, peerKind, contactType, threadID, threadType, metadata); err != nil { + slog.Warn("contact_collector.refresh_metadata_failed", + "error", err, + "tenant_id", TenantIDFromContext(ctx), + "channel", channelType, + "instance", channelInstance, + "sender", senderID, + ) + } +} + // ResolveTenantUserID delegates to the underlying ContactStore. func (c *ContactCollector) ResolveTenantUserID(ctx context.Context, channelType, senderID string) (string, error) { return c.store.ResolveTenantUserID(ctx, channelType, senderID) } + +// ListContacts delegates contact discovery queries to the backing store. +// Background channel syncs use this to backfill metadata for previously seen +// contacts that are no longer present in a platform's live listing. +func (c *ContactCollector) ListContacts(ctx context.Context, opts ContactListOpts) ([]ChannelContact, error) { + if c == nil || c.store == nil { + return nil, errors.New("contact collector unavailable") + } + return c.store.ListContacts(ctx, opts) +} diff --git a/internal/store/contact_store.go b/internal/store/contact_store.go index 584d3d13..6bd04d2a 100644 --- a/internal/store/contact_store.go +++ b/internal/store/contact_store.go @@ -29,13 +29,15 @@ type ChannelContact struct { // ContactListOpts holds pagination and filter options for listing contacts. type ContactListOpts struct { - Search string // ILIKE on display_name, username, sender_id - ChannelType string // filter by platform (telegram, discord, etc.) - ChannelInstance string // filter by channel instance name - PeerKind string // "direct" or "group" - ContactType string // "user" or "group" - Limit int - Offset int + Search string // ILIKE on display_name, username, sender_id + ChannelType string // filter by platform (telegram, discord, etc.) + ChannelInstance string // filter by channel instance name + PeerKind string // "direct" or "group" + ContactType string // "user" or "group" + SnapshotAt *time.Time // include only contacts discovered at or before this instant + OrderByFirstSeen bool // use immutable discovery order for stable background scans + Limit int + Offset int } // ContactStore manages channel contacts (auto-collected user info). @@ -78,3 +80,10 @@ type ContactStore interface { // Returns ("", nil) when the contact is not found or not merged. ResolveTenantUserID(ctx context.Context, channelType, senderID string) (string, error) } + +// ContactMetadataStore is an optional extension for platforms that have +// presentation metadata beyond the stable contact fields. Callers must fall +// back to ContactStore when an implementation does not provide it. +type ContactMetadataStore interface { + UpsertContactWithMetadata(ctx context.Context, channelType, channelInstance, senderID, userID, displayName, username, peerKind, contactType, threadID, threadType string, metadata map[string]string) error +} diff --git a/internal/store/pending_message_store.go b/internal/store/pending_message_store.go index cdf15dca..eeecc7d8 100644 --- a/internal/store/pending_message_store.go +++ b/internal/store/pending_message_store.go @@ -29,6 +29,7 @@ type PendingMessageGroup struct { HistoryKey string `json:"history_key" db:"history_key"` ParentHistoryKey string `json:"parent_history_key,omitempty" db:"parent_history_key"` GroupTitle string `json:"group_title,omitempty" db:"group_title"` + ParentGroupTitle string `json:"parent_group_title,omitempty" db:"-"` MessageCount int `json:"message_count" db:"message_count"` HasSummary bool `json:"has_summary" db:"has_summary"` LastActivity time.Time `json:"last_activity" db:"last_activity"` diff --git a/internal/store/pg/channel_contact_metadata.go b/internal/store/pg/channel_contact_metadata.go new file mode 100644 index 00000000..c8d1a39c --- /dev/null +++ b/internal/store/pg/channel_contact_metadata.go @@ -0,0 +1,40 @@ +package pg + +import ( + "context" + "encoding/json" + "fmt" + + "github.com/google/uuid" + + "github.com/nextlevelbuilder/goclaw/internal/store" +) + +// UpsertContactWithMetadata keeps the canonical contact name intact while +// persisting platform-specific presentation metadata. +func (s *PGContactStore) UpsertContactWithMetadata(ctx context.Context, channelType, channelInstance, senderID, userID, displayName, username, peerKind, contactType, threadID, threadType string, metadata map[string]string) error { + if err := s.UpsertContact(ctx, channelType, channelInstance, senderID, userID, displayName, username, peerKind, contactType, threadID, threadType); err != nil { + return err + } + if len(metadata) == 0 { + return nil + } + encoded, err := json.Marshal(metadata) + if err != nil { + return fmt.Errorf("marshal contact metadata: %w", err) + } + tenantID := store.TenantIDFromContext(ctx) + if tenantID == uuid.Nil { + tenantID = store.MasterTenantID + } + _, err = s.db.ExecContext(ctx, ` + UPDATE channel_contacts + SET metadata = $1::jsonb + WHERE tenant_id = $2 + AND channel_type = $3 + AND sender_id = $4 + AND COALESCE(thread_id, '') = $5`, + string(encoded), tenantID, channelType, senderID, threadID, + ) + return err +} diff --git a/internal/store/pg/channel_contacts.go b/internal/store/pg/channel_contacts.go index 92788091..94412a48 100644 --- a/internal/store/pg/channel_contacts.go +++ b/internal/store/pg/channel_contacts.go @@ -78,6 +78,11 @@ func contactWhereClause(ctx context.Context, opts store.ContactListOpts) (string args = append(args, opts.ContactType) argIdx++ } + if opts.SnapshotAt != nil { + conditions = append(conditions, fmt.Sprintf("first_seen_at <= $%d", argIdx)) + args = append(args, *opts.SnapshotAt) + argIdx++ + } if opts.Search != "" { escaped := strings.NewReplacer("%", "\\%", "_", "\\_").Replace(opts.Search) pattern := escaped + "%" @@ -100,9 +105,14 @@ func (s *PGContactStore) ListContacts(ctx context.Context, opts store.ContactLis where, args, argIdx := contactWhereClause(ctx, opts) query := `SELECT id, channel_type, channel_instance, sender_id, user_id, - display_name, username, avatar_url, peer_kind, contact_type, thread_id, thread_type, merged_id, + COALESCE(NULLIF(metadata->>'display_title', ''), display_name), username, avatar_url, peer_kind, contact_type, thread_id, thread_type, merged_id, first_seen_at, last_seen_at - FROM channel_contacts` + where + " ORDER BY last_seen_at DESC" + FROM channel_contacts` + where + if opts.OrderByFirstSeen { + query += " ORDER BY first_seen_at DESC, id DESC" + } else { + query += " ORDER BY last_seen_at DESC, id DESC" + } limit := opts.Limit if limit <= 0 { @@ -165,7 +175,7 @@ func (s *PGContactStore) GetContactsBySenderIDs(ctx context.Context, senderIDs [ query := fmt.Sprintf(`SELECT DISTINCT ON (sender_id) id, channel_type, channel_instance, sender_id, user_id, - display_name, username, avatar_url, peer_kind, contact_type, thread_id, thread_type, merged_id, + COALESCE(NULLIF(metadata->>'display_title', ''), display_name), username, avatar_url, peer_kind, contact_type, thread_id, thread_type, merged_id, first_seen_at, last_seen_at FROM channel_contacts WHERE sender_id IN (%s) AND tenant_id = %s @@ -196,7 +206,7 @@ func (s *PGContactStore) GetContactByID(ctx context.Context, id uuid.UUID) (*sto tid := store.TenantIDFromContext(ctx) row := s.db.QueryRowContext(ctx, `SELECT id, channel_type, channel_instance, sender_id, user_id, - display_name, username, avatar_url, peer_kind, contact_type, + COALESCE(NULLIF(metadata->>'display_title', ''), display_name), username, avatar_url, peer_kind, contact_type, thread_id, thread_type, merged_id, first_seen_at, last_seen_at FROM channel_contacts WHERE id = $1 AND tenant_id = $2`, id, tid) @@ -299,7 +309,7 @@ func (s *PGContactStore) GetContactsByMergedID(ctx context.Context, mergedID uui tid := store.TenantIDFromContext(ctx) q := `SELECT id, channel_type, channel_instance, sender_id, user_id, - display_name, username, avatar_url, peer_kind, contact_type, thread_id, thread_type, merged_id, + COALESCE(NULLIF(metadata->>'display_title', ''), display_name), username, avatar_url, peer_kind, contact_type, thread_id, thread_type, merged_id, first_seen_at, last_seen_at FROM channel_contacts WHERE merged_id = $1 AND tenant_id = $2 ORDER BY last_seen_at DESC` diff --git a/internal/store/pg/heartbeat.go b/internal/store/pg/heartbeat.go index 81104ff5..481e0924 100644 --- a/internal/store/pg/heartbeat.go +++ b/internal/store/pg/heartbeat.go @@ -235,7 +235,7 @@ func (s *PGHeartbeatStore) ListDeliveryTargets(ctx context.Context, tenantID uui cc.thread_id, cc.thread_type, cc.channel_instance AS channel, - COALESCE(cc.display_name, cc.sender_id) AS title, + COALESCE(NULLIF(cc.metadata->>'display_title', ''), cc.display_name, cc.sender_id) AS title, CASE WHEN cc.contact_type = 'topic' THEN 'topic' WHEN cc.peer_kind = 'group' THEN 'group' ELSE 'dm' END AS kind diff --git a/internal/store/pg/pending_message_store.go b/internal/store/pg/pending_message_store.go index e150a498..55c7fc9e 100644 --- a/internal/store/pg/pending_message_store.go +++ b/internal/store/pg/pending_message_store.go @@ -298,7 +298,7 @@ func (s *PGPendingMessageStore) resolveGroupTitlesFromContacts(ctx context.Conte } rows, err := s.db.QueryContext(ctx, - `SELECT channel_instance, sender_id, display_name + `SELECT channel_instance, sender_id, COALESCE(NULLIF(metadata->>'display_title', ''), display_name) FROM channel_contacts WHERE display_name IS NOT NULL AND display_name <> '' diff --git a/internal/store/sqlitestore/channel_contact_metadata.go b/internal/store/sqlitestore/channel_contact_metadata.go new file mode 100644 index 00000000..dca78937 --- /dev/null +++ b/internal/store/sqlitestore/channel_contact_metadata.go @@ -0,0 +1,42 @@ +//go:build sqlite || sqliteonly + +package sqlitestore + +import ( + "context" + "encoding/json" + "fmt" + + "github.com/google/uuid" + + "github.com/nextlevelbuilder/goclaw/internal/store" +) + +// UpsertContactWithMetadata keeps the canonical contact name intact while +// persisting platform-specific presentation metadata. +func (s *SQLiteContactStore) UpsertContactWithMetadata(ctx context.Context, channelType, channelInstance, senderID, userID, displayName, username, peerKind, contactType, threadID, threadType string, metadata map[string]string) error { + if err := s.UpsertContact(ctx, channelType, channelInstance, senderID, userID, displayName, username, peerKind, contactType, threadID, threadType); err != nil { + return err + } + if len(metadata) == 0 { + return nil + } + encoded, err := json.Marshal(metadata) + if err != nil { + return fmt.Errorf("marshal contact metadata: %w", err) + } + tenantID := store.TenantIDFromContext(ctx) + if tenantID == uuid.Nil { + tenantID = store.MasterTenantID + } + _, err = s.db.ExecContext(ctx, ` + UPDATE channel_contacts + SET metadata = ? + WHERE tenant_id = ? + AND channel_type = ? + AND sender_id = ? + AND COALESCE(thread_id, '') = ?`, + string(encoded), tenantID.String(), channelType, senderID, threadID, + ) + return err +} diff --git a/internal/store/sqlitestore/contacts.go b/internal/store/sqlitestore/contacts.go index cec6377f..03f49547 100644 --- a/internal/store/sqlitestore/contacts.go +++ b/internal/store/sqlitestore/contacts.go @@ -84,6 +84,10 @@ func contactWhereSQLite(ctx context.Context, opts store.ContactListOpts) (string conditions = append(conditions, "contact_type = ?") args = append(args, opts.ContactType) } + if opts.SnapshotAt != nil { + conditions = append(conditions, "first_seen_at <= ?") + args = append(args, *opts.SnapshotAt) + } if opts.Search != "" { escaped := strings.NewReplacer("%", "\\%", "_", "\\_").Replace(opts.Search) pattern := escaped + "%" @@ -99,7 +103,7 @@ func contactWhereSQLite(ctx context.Context, opts store.ContactListOpts) (string } const contactSelectCols = `id, channel_type, channel_instance, sender_id, user_id, - display_name, username, avatar_url, peer_kind, contact_type, thread_id, thread_type, merged_id, + COALESCE(NULLIF(json_extract(metadata, '$.display_title'), ''), display_name), username, avatar_url, peer_kind, contact_type, thread_id, thread_type, merged_id, first_seen_at, last_seen_at` func scanContact(rows *sql.Rows) (store.ChannelContact, error) { @@ -119,8 +123,12 @@ func (s *SQLiteContactStore) ListContacts(ctx context.Context, opts store.Contac if limit <= 0 { limit = 50 } + orderBy := "last_seen_at DESC, id DESC" + if opts.OrderByFirstSeen { + orderBy = "first_seen_at DESC, id DESC" + } query := `SELECT ` + contactSelectCols + ` FROM channel_contacts` + where + - fmt.Sprintf(" ORDER BY last_seen_at DESC LIMIT %d", limit) + fmt.Sprintf(" ORDER BY %s LIMIT %d", orderBy, limit) if opts.Offset > 0 { query += fmt.Sprintf(" OFFSET %d", opts.Offset) } diff --git a/internal/store/sqlitestore/heartbeat.go b/internal/store/sqlitestore/heartbeat.go index fb02277e..18e880c5 100644 --- a/internal/store/sqlitestore/heartbeat.go +++ b/internal/store/sqlitestore/heartbeat.go @@ -289,7 +289,7 @@ func (s *SQLiteHeartbeatStore) ListDeliveryTargets(ctx context.Context, tenantID cc.thread_id, cc.thread_type, cc.channel_instance AS channel, - COALESCE(cc.display_name, cc.sender_id) AS title, + COALESCE(NULLIF(json_extract(cc.metadata, '$.display_title'), ''), cc.display_name, cc.sender_id) AS title, CASE WHEN cc.contact_type = 'topic' THEN 'topic' WHEN cc.peer_kind = 'group' THEN 'group' ELSE 'dm' END AS kind @@ -337,4 +337,3 @@ func (s *SQLiteHeartbeatStore) ListDeliveryTargets(ctx context.Context, tenantID } return targets, nil } - diff --git a/internal/store/sqlitestore/heartbeat_delivery_targets_test.go b/internal/store/sqlitestore/heartbeat_delivery_targets_test.go new file mode 100644 index 00000000..c02c5616 --- /dev/null +++ b/internal/store/sqlitestore/heartbeat_delivery_targets_test.go @@ -0,0 +1,42 @@ +//go:build sqlite || sqliteonly + +package sqlitestore + +import ( + "context" + "testing" + + "github.com/google/uuid" + + "github.com/nextlevelbuilder/goclaw/internal/store" +) + +func TestListDeliveryTargetsUsesDiscordDisplayTitleMetadata(t *testing.T) { + db := openTestDB(t) + if err := EnsureSchema(db); err != nil { + t.Fatalf("EnsureSchema: %v", err) + } + tenantID := uuid.New() + if _, err := db.Exec(`INSERT INTO tenants (id, name, slug, status) VALUES (?, 'T', 't-heartbeat-title', 'active')`, tenantID.String()); err != nil { + t.Fatalf("insert tenant: %v", err) + } + ctx := store.WithTenantID(context.Background(), tenantID) + contacts := NewSQLiteContactStore(db) + if err := contacts.UpsertContactWithMetadata(ctx, "discord", "discord-main", "thread-1", "", "launch-thread", "", "group", "group", "", "", map[string]string{"display_title": "launch-thread / product-planning"}); err != nil { + t.Fatalf("upsert contact: %v", err) + } + + targets, err := NewSQLiteHeartbeatStore(db).ListDeliveryTargets(ctx, tenantID) + if err != nil { + t.Fatalf("ListDeliveryTargets: %v", err) + } + if len(targets) != 1 { + t.Fatalf("targets = %d, want 1", len(targets)) + } + if targets[0].ChatID != "thread-1" { + t.Fatalf("chat id = %q, want stable thread id", targets[0].ChatID) + } + if targets[0].Title != "launch-thread / product-planning" { + t.Fatalf("title = %q, want qualified title", targets[0].Title) + } +} diff --git a/internal/store/sqlitestore/pending_messages.go b/internal/store/sqlitestore/pending_messages.go index 48f70b6a..498ad824 100644 --- a/internal/store/sqlitestore/pending_messages.go +++ b/internal/store/sqlitestore/pending_messages.go @@ -314,7 +314,7 @@ func (s *SQLitePendingMessageStore) resolveGroupTitlesFromContacts(ctx context.C } rows, err := s.db.QueryContext(ctx, - `SELECT channel_instance, sender_id, display_name + `SELECT channel_instance, sender_id, COALESCE(NULLIF(json_extract(metadata, '$.display_title'), ''), display_name) FROM channel_contacts WHERE display_name IS NOT NULL AND display_name <> '' diff --git a/ui/web/src/pages/channels/hooks/use-channel-detail.ts b/ui/web/src/pages/channels/hooks/use-channel-detail.ts index 2f7c635d..f397d6d2 100644 --- a/ui/web/src/pages/channels/hooks/use-channel-detail.ts +++ b/ui/web/src/pages/channels/hooks/use-channel-detail.ts @@ -75,9 +75,21 @@ export function useChannelDetail(instanceId: string | undefined) { async () => { if (!instanceId) return; try { - await http.post(`/v1/channels/instances/${instanceId}/metadata/refresh`); + const result = await http.post<{ + ok: boolean; + report?: { + errors?: string[]; + failures?: Array<{ channel_id: string; reason: string }>; + }; + }>(`/v1/channels/instances/${instanceId}/metadata/refresh`); queryClient.invalidateQueries({ queryKey: queryKeys.channels.memoryExtractionGroups(instanceId), exact: true }); await invalidate(); + if (!result?.ok) { + const detail = result.report?.errors?.[0] + ?? result.report?.failures?.[0]?.reason; + toast.error(i18next.t("channels:detail.discordMetadataRefreshFailed"), detail); + return; + } toast.success(i18next.t("channels:detail.discordMetadataRefreshed")); } catch (err) { toast.error(i18next.t("channels:detail.discordMetadataRefreshFailed"), userFriendlyError(err)); diff --git a/ui/web/src/pages/pending-messages/message-list-dialog.tsx b/ui/web/src/pages/pending-messages/message-list-dialog.tsx index f2ed1d06..0283890b 100644 --- a/ui/web/src/pages/pending-messages/message-list-dialog.tsx +++ b/ui/web/src/pages/pending-messages/message-list-dialog.tsx @@ -8,7 +8,7 @@ import { } from "@/components/ui/dialog"; import { Badge } from "@/components/ui/badge"; import { formatDate } from "@/lib/format"; -import type { PendingMessageGroup, PendingMessage } from "./types"; +import { formatPendingGroupLabel, type PendingMessageGroup, type PendingMessage } from "./types"; interface MessageListDialogProps { group: PendingMessageGroup; @@ -35,7 +35,7 @@ export function MessageListDialog({ - {t("dialog.title", { name: group.group_title || group.history_key })} + {t("dialog.title", { name: formatPendingGroupLabel(group) })} {group.channel_name} diff --git a/ui/web/src/pages/pending-messages/pending-messages-page.tsx b/ui/web/src/pages/pending-messages/pending-messages-page.tsx index 95ca027d..e8b9f22e 100644 --- a/ui/web/src/pages/pending-messages/pending-messages-page.tsx +++ b/ui/web/src/pages/pending-messages/pending-messages-page.tsx @@ -11,7 +11,7 @@ import { useMinLoading } from "@/hooks/use-min-loading"; import { useDeferredLoading } from "@/hooks/use-deferred-loading"; import { usePendingMessages } from "./hooks/use-pending-messages"; import { MessageListDialog } from "./message-list-dialog"; -import type { PendingMessageGroup } from "./types"; +import { formatPendingGroupLabel, type PendingMessageGroup } from "./types"; export function PendingMessagesPage() { const { t } = useTranslation("pending-messages"); @@ -147,7 +147,7 @@ export function PendingMessagesPage() { {g.group_title ? ( - {g.group_title} + {formatPendingGroupLabel(g)} {g.history_key.includes(":topic:") && ( > topic:{g.history_key.split(":topic:")[1]} diff --git a/ui/web/src/pages/pending-messages/types.test.ts b/ui/web/src/pages/pending-messages/types.test.ts new file mode 100644 index 00000000..62304a30 --- /dev/null +++ b/ui/web/src/pages/pending-messages/types.test.ts @@ -0,0 +1,20 @@ +import { describe, expect, it } from "vitest"; +import { formatPendingGroupLabel } from "./types"; + +describe("formatPendingGroupLabel", () => { + it("qualifies a child group with its resolved parent title", () => { + expect(formatPendingGroupLabel({ + history_key: "thread-1", + group_title: "launch-thread", + parent_history_key: "parent-1", + parent_group_title: "product-planning", + })).toBe("launch-thread / product-planning"); + }); + + it("falls back to stable IDs when titles are unavailable", () => { + expect(formatPendingGroupLabel({ + history_key: "thread-1", + parent_history_key: "parent-1", + })).toBe("thread-1 / parent-1"); + }); +}); diff --git a/ui/web/src/pages/pending-messages/types.ts b/ui/web/src/pages/pending-messages/types.ts index a62e78a8..4cf4938d 100644 --- a/ui/web/src/pages/pending-messages/types.ts +++ b/ui/web/src/pages/pending-messages/types.ts @@ -1,12 +1,21 @@ export interface PendingMessageGroup { channel_name: string; history_key: string; + parent_history_key?: string; group_title?: string; + parent_group_title?: string; message_count: number; has_summary: boolean; last_activity: string; } +export function formatPendingGroupLabel(group: Pick): string { + const title = group.group_title || group.history_key; + const parent = group.parent_group_title || group.parent_history_key; + if (!group.parent_history_key || !parent || parent === title) return title; + return `${title} / ${parent}`; +} + export interface PendingMessage { id: string; channel_name: string;