1package acp
2
3import (
4 "context"
5 "log/slog"
6
7 "github.com/charmbracelet/crush/internal/message"
8 "github.com/charmbracelet/crush/internal/permission"
9 "github.com/charmbracelet/crush/internal/pubsub"
10 "github.com/coder/acp-go-sdk"
11)
12
13// Sink receives events from Crush's pubsub system and translates them to ACP
14// session updates.
15type Sink struct {
16 ctx context.Context
17 conn *acp.AgentSideConnection
18 sessionID string
19
20 // Track text deltas per message to avoid re-sending content.
21 textOffsets map[string]int
22 reasoningOffsets map[string]int
23}
24
25// NewSink creates a new event sink for the given session.
26func NewSink(ctx context.Context, conn *acp.AgentSideConnection, sessionID string) *Sink {
27 return &Sink{
28 ctx: ctx,
29 conn: conn,
30 sessionID: sessionID,
31 textOffsets: make(map[string]int),
32 reasoningOffsets: make(map[string]int),
33 }
34}
35
36// HandleMessage translates a Crush message event to ACP session updates.
37func (s *Sink) HandleMessage(event pubsub.Event[message.Message]) {
38 msg := event.Payload
39
40 // Only handle messages for our session.
41 if msg.SessionID != s.sessionID {
42 return
43 }
44
45 for _, part := range msg.Parts {
46 update := s.translatePart(msg.ID, msg.Role, part)
47 if update == nil {
48 continue
49 }
50
51 if err := s.conn.SessionUpdate(s.ctx, acp.SessionNotification{
52 SessionId: acp.SessionId(s.sessionID),
53 Update: *update,
54 }); err != nil {
55 slog.Error("Failed to send session update", "error", err)
56 }
57 }
58}
59
60// HandlePermission translates a permission request to an ACP permission request.
61func (s *Sink) HandlePermission(req permission.PermissionRequest, permissions permission.Service) {
62 // Only handle permissions for our session.
63 if req.SessionID != s.sessionID {
64 return
65 }
66
67 slog.Debug("ACP permission request", "tool", req.ToolName, "action", req.Action)
68
69 resp, err := s.conn.RequestPermission(s.ctx, acp.RequestPermissionRequest{
70 SessionId: acp.SessionId(s.sessionID),
71 ToolCall: acp.RequestPermissionToolCall{
72 ToolCallId: acp.ToolCallId(req.ToolCallID),
73 Title: acp.Ptr(req.Description),
74 Kind: acp.Ptr(acp.ToolKindEdit),
75 Status: acp.Ptr(acp.ToolCallStatusPending),
76 Locations: []acp.ToolCallLocation{{Path: req.Path}},
77 RawInput: req.Params,
78 },
79 Options: []acp.PermissionOption{
80 {Kind: acp.PermissionOptionKindAllowOnce, Name: "Allow", OptionId: "allow"},
81 {Kind: acp.PermissionOptionKindAllowAlways, Name: "Allow always", OptionId: "allow_always"},
82 {Kind: acp.PermissionOptionKindRejectOnce, Name: "Deny", OptionId: "deny"},
83 },
84 })
85 if err != nil {
86 slog.Error("Failed to request permission", "error", err)
87 permissions.Deny(req)
88 return
89 }
90
91 if resp.Outcome.Cancelled != nil {
92 permissions.Deny(req)
93 return
94 }
95
96 if resp.Outcome.Selected != nil {
97 switch string(resp.Outcome.Selected.OptionId) {
98 case "allow":
99 permissions.Grant(req)
100 case "allow_always":
101 permissions.GrantPersistent(req)
102 default:
103 permissions.Deny(req)
104 }
105 }
106}
107
108// translatePart converts a message part to an ACP session update.
109func (s *Sink) translatePart(msgID string, role message.MessageRole, part message.ContentPart) *acp.SessionUpdate {
110 switch p := part.(type) {
111 case message.TextContent:
112 return s.translateText(msgID, role, p)
113
114 case message.ReasoningContent:
115 return s.translateReasoning(msgID, p)
116
117 case message.ToolCall:
118 return s.translateToolCall(p)
119
120 case message.ToolResult:
121 return s.translateToolResult(p)
122
123 case message.Finish:
124 // Reset offsets on message finish.
125 delete(s.textOffsets, msgID)
126 delete(s.reasoningOffsets, msgID)
127 return nil
128
129 default:
130 return nil
131 }
132}
133
134func (s *Sink) translateText(msgID string, role message.MessageRole, text message.TextContent) *acp.SessionUpdate {
135 offset := s.textOffsets[msgID]
136 if len(text.Text) <= offset {
137 return nil
138 }
139
140 delta := text.Text[offset:]
141 s.textOffsets[msgID] = len(text.Text)
142
143 if delta == "" {
144 return nil
145 }
146
147 switch role {
148 case message.Assistant:
149 update := acp.UpdateAgentMessageText(delta)
150 return &update
151 case message.User:
152 update := acp.UpdateUserMessageText(delta)
153 return &update
154 default:
155 return nil
156 }
157}
158
159func (s *Sink) translateReasoning(msgID string, reasoning message.ReasoningContent) *acp.SessionUpdate {
160 offset := s.reasoningOffsets[msgID]
161 if len(reasoning.Thinking) <= offset {
162 return nil
163 }
164
165 delta := reasoning.Thinking[offset:]
166 s.reasoningOffsets[msgID] = len(reasoning.Thinking)
167
168 if delta == "" {
169 return nil
170 }
171
172 update := acp.UpdateAgentThoughtText(delta)
173 return &update
174}
175
176func (s *Sink) translateToolCall(tc message.ToolCall) *acp.SessionUpdate {
177 if !tc.Finished {
178 update := acp.StartToolCall(
179 acp.ToolCallId(tc.ID),
180 tc.Name,
181 acp.WithStartStatus(acp.ToolCallStatusPending),
182 )
183 return &update
184 }
185
186 update := acp.UpdateToolCall(
187 acp.ToolCallId(tc.ID),
188 acp.WithUpdateStatus(acp.ToolCallStatusInProgress),
189 )
190 return &update
191}
192
193func (s *Sink) translateToolResult(tr message.ToolResult) *acp.SessionUpdate {
194 status := acp.ToolCallStatusCompleted
195 if tr.IsError {
196 status = acp.ToolCallStatusFailed
197 }
198
199 update := acp.UpdateToolCall(
200 acp.ToolCallId(tr.ToolCallID),
201 acp.WithUpdateStatus(status),
202 acp.WithUpdateContent([]acp.ToolCallContent{
203 acp.ToolContent(acp.TextBlock(tr.Content)),
204 }),
205 )
206 return &update
207}