1package acp
2
3import (
4 "context"
5
6 "github.com/charmbracelet/crush/internal/agent/tools/mcp"
7 "github.com/charmbracelet/crush/internal/message"
8 "github.com/charmbracelet/crush/internal/permission"
9 "github.com/charmbracelet/crush/internal/pubsub"
10 "github.com/charmbracelet/crush/internal/session"
11 "github.com/coder/acp-go-sdk"
12)
13
14// Sink receives events from Crush's pubsub system and translates them to ACP
15// session updates.
16type Sink struct {
17 ctx context.Context
18 cancel context.CancelFunc
19 conn *acp.AgentSideConnection
20 sessionID string
21
22 // Track text deltas per message to avoid re-sending content.
23 textOffsets map[string]int
24 reasoningOffsets map[string]int
25}
26
27// NewSink creates a new event sink for the given session.
28func NewSink(ctx context.Context, conn *acp.AgentSideConnection, sessionID string) *Sink {
29 sinkCtx, cancel := context.WithCancel(ctx)
30 return &Sink{
31 ctx: sinkCtx,
32 cancel: cancel,
33 conn: conn,
34 sessionID: sessionID,
35 textOffsets: make(map[string]int),
36 reasoningOffsets: make(map[string]int),
37 }
38}
39
40// Start subscribes to messages, permissions, and sessions, forwarding events to ACP.
41func (s *Sink) Start(messages message.Service, permissions permission.Service, sessions session.Service) {
42 // Subscribe to message events.
43 go func() {
44 msgCh := messages.Subscribe(s.ctx)
45 for {
46 select {
47 case event, ok := <-msgCh:
48 if !ok {
49 return
50 }
51 s.HandleMessage(event)
52 case <-s.ctx.Done():
53 return
54 }
55 }
56 }()
57
58 // Subscribe to permission events.
59 go func() {
60 permCh := permissions.Subscribe(s.ctx)
61 for {
62 select {
63 case event, ok := <-permCh:
64 if !ok {
65 return
66 }
67 s.HandlePermission(event.Payload, permissions)
68 case <-s.ctx.Done():
69 return
70 }
71 }
72 }()
73
74 // Subscribe to session events for todo/plan updates.
75 go func() {
76 sessCh := sessions.Subscribe(s.ctx)
77 for {
78 select {
79 case event, ok := <-sessCh:
80 if !ok {
81 return
82 }
83 s.HandleSession(event)
84 case <-s.ctx.Done():
85 return
86 }
87 }
88 }()
89
90 // Subscribe to MCP events to refresh commands when prompts change.
91 go func() {
92 mcpCh := mcp.SubscribeEvents(s.ctx)
93 for {
94 select {
95 case event, ok := <-mcpCh:
96 if !ok {
97 return
98 }
99 s.HandleMCPEvent(pubsub.Event[mcp.Event](event))
100 case <-s.ctx.Done():
101 return
102 }
103 }
104 }()
105
106 // Publish initial commands.
107 s.PublishCommands()
108}
109
110// Stop cancels the sink's subscriptions.
111func (s *Sink) Stop() {
112 s.cancel()
113}