fix(discord): qualify group titles across surfaces (#1420)

Co-authored-by: Collective Developer <man@collective.dev>
This commit is contained in:
ナムandCollective Developer authored and GitHub committed 2026-07-10 22:25:45 +07:00
1 parent 244197c2db
commit 6b738924d5
46 files changed
+1442 -88

No files matched your search

+3
View File
@@ -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,
+3 -2
View File
@@ -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
+23
View File
@@ -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"
+15 -1
View File
@@ -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", "", "")
}
+51
View File
@@ -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 {
+2
View File
@@ -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),
+48
View File
@@ -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
+5
View File
@@ -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))
+1
View File
@@ -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,
+3
View File
@@ -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,
+6
View File
@@ -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.
+50 -2
View File
@@ -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.
+5 -2
View File
@@ -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)
@@ -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.
+38
View File
@@ -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
}
+31
View File
@@ -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 {
+174 -15
View File
@@ -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 {
@@ -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()
}
+38 -5
View File
@@ -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)
+21 -7
View File
@@ -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
+30
View File
@@ -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 {
+67
View File
@@ -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}
}
+31
View File
@@ -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.
+17
View File
@@ -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) {
+38 -19
View File
@@ -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()
+7 -4
View File
@@ -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})
}
@@ -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) {
@@ -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)
}
}
+24 -2
View File
@@ -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]
}
}
}
+63
View File
@@ -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)
}
+16 -7
View File
@@ -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
}
+1
View File
@@ -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"`
@@ -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
}
+15 -5
View File
@@ -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`
+1 -1
View File
@@ -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
+1 -1
View File
@@ -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 <> ''
@@ -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
}
+10 -2
View File
@@ -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)
}
+1 -2
View File
@@ -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
}
@@ -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)
}
}
@@ -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 <> ''
@@ -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));
@@ -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({
<DialogContent className="max-h-[85vh] flex flex-col sm:max-w-2xl">
<DialogHeader>
<DialogTitle className="flex items-center gap-2 truncate">
{t("dialog.title", { name: group.group_title || group.history_key })}
{t("dialog.title", { name: formatPendingGroupLabel(group) })}
<Badge variant="outline" className="text-xs">{group.channel_name}</Badge>
</DialogTitle>
</DialogHeader>
@@ -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() {
<td className="max-w-[300px] truncate px-4 py-3">
{g.group_title ? (
<span className="font-medium">
{g.group_title}
{formatPendingGroupLabel(g)}
{g.history_key.includes(":topic:") && (
<span className="ml-1.5 text-xs font-normal text-muted-foreground">
&gt; topic:{g.history_key.split(":topic:")[1]}
@@ -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");
});
});
@@ -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<PendingMessageGroup, "history_key" | "parent_history_key" | "group_title" | "parent_group_title">): 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;