OPC/synapbus
2
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity

feat: add reply_to parameter to send_channel_message for threading support

Browse Source 
The BroadcastMessage function now accepts a replyTo parameter, allowing
channel messages to reference a parent message ID and create threads.
Updated all callers (MCP bridge, hybrid tools, tests) and added a test
verifying reply_to is correctly stored and returned.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
Algis Dumbris
2026-03-16 20:21:52 +02:00
co-authored by Claude Opus 4.6
parent bd166843d6
commit d0b548f75f
6 changed files with 37 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/actions/registry.go
+1
View File
@@ -306,6 +306,7 @@ func allActions() []Action {
{Name: "body", Type: "string", Description: "Message body text", Required: true},
{Name: "priority", Type: "number", Description: "Message priority (1-10, default 5)", Default: "5"},
{Name: "metadata", Type: "string", Description: "JSON metadata object (optional)"},
{Name: "reply_to", Type: "number", Description: "Message ID to reply to (creates a thread)", Required: false},
},
Returns: "JSON with channel_id, message_id, and status 'sent'",
Examples: []Example{
internal/api/notifications_handler_test.go
+1 -1
View File
@@ -333,7 +333,7 @@ func TestChannelMessages_IncludesLastRead(t *testing.T) {
}
// Broadcast messages
msgs, err := channelService.BroadcastMessage(ctx, ch.ID, "human-agent", "Hello channel", 5, "")
msgs, err := channelService.BroadcastMessage(ctx, ch.ID, "human-agent", "Hello channel", 5, "", nil)
if err != nil {
t.Fatalf("broadcast: %v", err)
}
internal/channels/service.go
+2 -1
View File
@@ -465,7 +465,7 @@ func (s *Service) UpdateChannel(ctx context.Context, channelID int64, req Update
// If the message body contains @mentions, mentioned members receive a
// "mention":true flag in their inbox notification metadata, and the channel
// message metadata includes "mentioned_agents".
func (s *Service) BroadcastMessage(ctx context.Context, channelID int64, fromAgent, body string, priority int, metadata string) ([]*messaging.Message, error) {
func (s *Service) BroadcastMessage(ctx context.Context, channelID int64, fromAgent, body string, priority int, metadata string, replyTo *int64) ([]*messaging.Message, error) {
ch, err := s.store.GetChannel(ctx, channelID)
if err != nil {
return nil, err
@@ -526,6 +526,7 @@ func (s *Service) BroadcastMessage(ctx context.Context, channelID int64, fromAge
Priority: priority,
Metadata: string(channelMetaBytes),
ChannelID: &channelID,
ReplyTo: replyTo,
})
if err != nil {
return nil, fmt.Errorf("create channel message: %w", err)
internal/channels/service_test.go
+23 -10
View File
@@ -533,7 +533,7 @@ func TestService_BroadcastMessage(t *testing.T) {
ch, _ := svc.CreateChannel(ctx, CreateChannelRequest{Name: "alerts", Type: TypeStandard, CreatedBy: "agent-a"})
t.Run("broadcast creates channel message", func(t *testing.T) {
msgs, err := svc.BroadcastMessage(ctx, ch.ID, "agent-a", "hello", 5, "")
msgs, err := svc.BroadcastMessage(ctx, ch.ID, "agent-a", "hello", 5, "", nil)
if err != nil {
t.Fatalf("BroadcastMessage: %v", err)
}
@@ -568,7 +568,7 @@ func TestService_BroadcastMessage(t *testing.T) {
t.Run("broadcast without mentions sends no DMs", func(t *testing.T) {
svc.JoinChannel(ctx, ch.ID, "agent-b")
svc.JoinChannel(ctx, ch.ID, "agent-c")
_, err := svc.BroadcastMessage(ctx, ch.ID, "agent-a", "no-dm-test", 5, "")
_, err := svc.BroadcastMessage(ctx, ch.ID, "agent-a", "no-dm-test", 5, "", nil)
if err != nil {
t.Fatalf("BroadcastMessage: %v", err)
}
@@ -583,7 +583,7 @@ func TestService_BroadcastMessage(t *testing.T) {
})
t.Run("sender does not receive own message", func(t *testing.T) {
svc.BroadcastMessage(ctx, ch.ID, "agent-a", "no self-message", 5, "")
svc.BroadcastMessage(ctx, ch.ID, "agent-a", "no self-message", 5, "", nil)
inboxResult, _ := svc.msgService.ReadInbox(ctx, "agent-a", messaging.ReadOptions{IncludeRead: true})
for _, m := range inboxResult.Messages {
if m.Body == "no self-message" {
@@ -594,7 +594,7 @@ func TestService_BroadcastMessage(t *testing.T) {
t.Run("non-member auto-joins public channel on broadcast", func(t *testing.T) {
seedAgent(t, svc.store.(*SQLiteChannelStore).db, "outsider")
_, err := svc.BroadcastMessage(ctx, ch.ID, "outsider", "auto-joined", 5, "")
_, err := svc.BroadcastMessage(ctx, ch.ID, "outsider", "auto-joined", 5, "", nil)
if err != nil {
t.Fatalf("expected auto-join for public channel, got %v", err)
}
@@ -604,6 +604,19 @@ func TestService_BroadcastMessage(t *testing.T) {
}
})
t.Run("broadcast with reply_to", func(t *testing.T) {
msgs, _ := svc.BroadcastMessage(ctx, ch.ID, "agent-a", "original", 5, "", nil)
original := msgs[0]
replies, err := svc.BroadcastMessage(ctx, ch.ID, "agent-b", "reply to original", 5, "", &original.ID)
if err != nil {
t.Fatalf("BroadcastMessage with reply_to: %v", err)
}
if replies[0].ReplyTo == nil || *replies[0].ReplyTo != original.ID {
t.Errorf("reply_to = %v, want %d", replies[0].ReplyTo, original.ID)
}
})
t.Run("non-member cannot broadcast to private channel", func(t *testing.T) {
privCh, err := svc.CreateChannel(ctx, CreateChannelRequest{
Name: "private-test", Type: TypeStandard, IsPrivate: true, CreatedBy: "agent-a",
@@ -612,7 +625,7 @@ func TestService_BroadcastMessage(t *testing.T) {
t.Fatalf("create private channel: %v", err)
}
seedAgent(t, svc.store.(*SQLiteChannelStore).db, "outsider2")
_, err = svc.BroadcastMessage(ctx, privCh.ID, "outsider2", "unauthorized", 5, "")
_, err = svc.BroadcastMessage(ctx, privCh.ID, "outsider2", "unauthorized", 5, "", nil)
if !errors.Is(err, ErrNotChannelMember) {
t.Errorf("expected ErrNotChannelMember for private channel, got %v", err)
}
@@ -631,7 +644,7 @@ func TestService_BroadcastMessage_Mentions(t *testing.T) {
svc.JoinChannel(ctx, ch.ID, "agent-c")
t.Run("mentioned member gets mention flag in inbox", func(t *testing.T) {
_, err := svc.BroadcastMessage(ctx, ch.ID, "agent-a", "hey @agent-b check this", 5, "")
_, err := svc.BroadcastMessage(ctx, ch.ID, "agent-a", "hey @agent-b check this", 5, "", nil)
if err != nil {
t.Fatalf("BroadcastMessage: %v", err)
}
@@ -664,7 +677,7 @@ func TestService_BroadcastMessage_Mentions(t *testing.T) {
})
t.Run("channel message metadata includes mentioned_agents", func(t *testing.T) {
_, err := svc.BroadcastMessage(ctx, ch.ID, "agent-a", "cc @agent-b and @agent-c", 5, "")
_, err := svc.BroadcastMessage(ctx, ch.ID, "agent-a", "cc @agent-b and @agent-c", 5, "", nil)
if err != nil {
t.Fatalf("BroadcastMessage: %v", err)
}
@@ -694,7 +707,7 @@ func TestService_BroadcastMessage_Mentions(t *testing.T) {
})
t.Run("self-mention is excluded", func(t *testing.T) {
_, err := svc.BroadcastMessage(ctx, ch.ID, "agent-a", "I am @agent-a and cc @agent-b", 5, "")
_, err := svc.BroadcastMessage(ctx, ch.ID, "agent-a", "I am @agent-a and cc @agent-b", 5, "", nil)
if err != nil {
t.Fatalf("BroadcastMessage: %v", err)
}
@@ -719,7 +732,7 @@ func TestService_BroadcastMessage_Mentions(t *testing.T) {
})
t.Run("no mentions produces no mention metadata", func(t *testing.T) {
_, err := svc.BroadcastMessage(ctx, ch.ID, "agent-a", "just a normal message", 5, "")
_, err := svc.BroadcastMessage(ctx, ch.ID, "agent-a", "just a normal message", 5, "", nil)
if err != nil {
t.Fatalf("BroadcastMessage: %v", err)
}
@@ -739,7 +752,7 @@ func TestService_BroadcastMessage_Mentions(t *testing.T) {
t.Run("non-member mention is ignored", func(t *testing.T) {
seedAgent(t, svc.store.(*SQLiteChannelStore).db, "outsider")
_, err := svc.BroadcastMessage(ctx, ch.ID, "agent-a", "hey @outsider and @agent-b", 5, "")
_, err := svc.BroadcastMessage(ctx, ch.ID, "agent-a", "hey @outsider and @agent-b", 5, "", nil)
if err != nil {
t.Fatalf("BroadcastMessage: %v", err)
}
internal/mcp/bridge.go
+9 -1
View File
@@ -546,7 +546,15 @@ func (b *ServiceBridge) callSendChannelMessage(ctx context.Context, args map[str
priority := getInt(args, "priority", 5)
metadata := getString(args, "metadata", "")
messages, err := b.channelService.BroadcastMessage(ctx, channelID, b.agentName, body, priority, metadata)
var replyTo *int64
if v, ok := args["reply_to"]; ok {
if f, ok := v.(float64); ok {
r := int64(f)
replyTo = &r
}
}
messages, err := b.channelService.BroadcastMessage(ctx, channelID, b.agentName, body, priority, metadata, replyTo)
if err != nil {
return nil, err
}
internal/mcp/tools_hybrid.go
+1 -1
View File
@@ -335,7 +335,7 @@ func (h *HybridToolRegistrar) handleSendMessage(ctx context.Context, req mcplib.
return mcplib.NewToolResultError(fmt.Sprintf("send_message to channel failed: %s", err)), nil
}
messages, err := h.channelService.BroadcastMessage(ctx, channelID, agentName, body, priority, metadataStr)
messages, err := h.channelService.BroadcastMessage(ctx, channelID, agentName, body, priority, metadataStr, replyTo)
if err != nil {
return mcplib.NewToolResultError(fmt.Sprintf("send_message to channel failed: %s", err)), nil
}