rickub
Sign in Get started
nandi/oripublic⑂ Fork 0
⑂ 34e69b510306161654f26903a269247aefa9c94d
Branches
main View all branches → View all tags →
Commits
⬇ Clone ▾
git clone https://git.rickub.com/nandi/ori.git
git clone ssh://git@rickub.com/nandi/ori.git

Host key fingerprint (ed25519): SHA256:iycHnxEyq0Q7uyVpB7JlznP0G7JrTPXLYRcAU5CSLhc — verify it before your first connect.

forked from bots-garden/ori

bridge.go · 321 lines · 10.1 KBGo Blame HistoryRaw
✨ Introduce new feature(s): ACP web client — Go backend (agent, bridge, httpserver, mockagent) + React SPA (Zed-like agent panel), tests, quality gate PASS 2434cc7 k33g yesterday1// Package bridge relays events between the ACP agent session and the
2// browsers connected over WebSocket.
3//
4// It is the meeting point of two asynchronous worlds: the agent streams
5// updates and blocks on permission requests; browsers connect, disconnect and
6// answer at human speed. The bridge broadcasts agent events to every client,
7// replays the session history to late joiners, and routes each permission
8// decision back to the exact request waiting for it.
9package bridge
10
11import (
12 "context"
13 "encoding/json"
14 "fmt"
15 "log/slog"
✨ Workspace panel, selectors, previews, desktop app, sandbox template, resizable file tree, light/dark theme 76d62ac k33g yesterday16 "path/filepath"
✨ Introduce new feature(s): ACP web client — Go backend (agent, bridge, httpserver, mockagent) + React SPA (Zed-like agent panel), tests, quality gate PASS 2434cc7 k33g yesterday17 "strconv"
18 "sync"
19
20 acp "github.com/coder/acp-go-sdk"
21)
22
23// Prompter is the slice of the agent session the bridge drives. It is
24// satisfied by *agent.Session and by test fakes.
25type Prompter interface {
26 ID() string
✨ Workspace panel, selectors, previews, desktop app, sandbox template, resizable file tree, light/dark theme 76d62ac k33g yesterday27 Prompt(ctx context.Context, blocks []acp.ContentBlock) (acp.StopReason, error)
✨ Introduce new feature(s): ACP web client — Go backend (agent, bridge, httpserver, mockagent) + React SPA (Zed-like agent panel), tests, quality gate PASS 2434cc7 k33g yesterday28 Cancel(ctx context.Context) error
29}
30
31// maxHistory bounds the number of events replayed to late-joining clients;
32// beyond it the oldest events are dropped (the live broadcast is unaffected).
33const maxHistory = 4096
34
35// clientBuffer is each subscriber's outgoing queue; a client too slow to
36// drain it is disconnected rather than allowed to stall the whole bridge.
37const clientBuffer = 256
38
39type permissionDecision struct {
40 optionId string
41 cancelled bool
42}
43
44// Bridge broadcasts agent events to WebSocket clients and routes their
45// responses back to the agent. The zero value is not usable; call New.
46//
47// Example:
48//
49// b := bridge.New(slog.Default())
50// session, _ := agent.Start(ctx, agent.Options{Handler: b, ...})
51// b.SetSession(session)
52// http.Handle("GET /ws", b.WebSocketHandler())
53type Bridge struct {
54 logger *slog.Logger
55
56 mu sync.Mutex
57 session Prompter
✨ Workspace panel, selectors, previews, desktop app, sandbox template, resizable file tree, light/dark theme 76d62ac k33g yesterday58 root string
✨ Introduce new feature(s): ACP web client — Go backend (agent, bridge, httpserver, mockagent) + React SPA (Zed-like agent panel), tests, quality gate PASS 2434cc7 k33g yesterday59 subscribers map[chan []byte]struct{}
60 history [][]byte
61 turnActive bool
62 nextRequest int
63 pendingPerms map[string]chan permissionDecision
64 // pendingRequests keeps the encoded permission_request events so they
65 // can be replayed to clients that connect while a request is open.
66 pendingRequests map[string][]byte
67}
68
69// New creates an empty bridge; attach the agent session with SetSession.
70func New(logger *slog.Logger) *Bridge {
71 if logger == nil {
72 logger = slog.Default()
73 }
74 return &Bridge{
75 logger: logger,
76 subscribers: make(map[chan []byte]struct{}),
77 pendingPerms: make(map[string]chan permissionDecision),
78 pendingRequests: make(map[string][]byte),
79 }
80}
81
82// SetSession attaches the agent session the bridge drives. It must be called
83// before the first client message arrives.
84func (b *Bridge) SetSession(session Prompter) {
85 b.mu.Lock()
86 defer b.mu.Unlock()
87 b.session = session
88}
89
✨ Workspace panel, selectors, previews, desktop app, sandbox template, resizable file tree, light/dark theme 76d62ac k33g yesterday90// SetWorkspaceRoot sets the directory relative attachment paths resolve
91// against (the agent's cwd). Without it, relative paths are sent as given.
92func (b *Bridge) SetWorkspaceRoot(root string) {
93 b.mu.Lock()
94 defer b.mu.Unlock()
95 b.root = root
96}
97
✨ Introduce new feature(s): ACP web client — Go backend (agent, bridge, httpserver, mockagent) + React SPA (Zed-like agent panel), tests, quality gate PASS 2434cc7 k33g yesterday98// Subscribe registers a new client. It returns the client's event channel,
99// the events to replay so the client catches up (history, then any pending
100// permission requests), and an unsubscribe function.
101func (b *Bridge) Subscribe() (events <-chan []byte, replay [][]byte, unsubscribe func()) {
102 ch := make(chan []byte, clientBuffer)
103 b.mu.Lock()
104 defer b.mu.Unlock()
105
106 b.subscribers[ch] = struct{}{}
107
108 sessionId := ""
109 if b.session != nil {
110 sessionId = b.session.ID()
111 }
112 replay = append(replay, encode(Outgoing{Type: OutgoingHello, SessionId: sessionId, TurnActive: b.turnActive}))
113 replay = append(replay, b.history...)
114 for _, raw := range b.pendingRequests {
115 replay = append(replay, raw)
116 }
117
118 return ch, replay, func() {
119 b.mu.Lock()
120 defer b.mu.Unlock()
121 if _, ok := b.subscribers[ch]; ok {
122 delete(b.subscribers, ch)
123 close(ch)
124 }
125 }
126}
127
128// HandleIncoming processes one message from a client. Errors are reported to
129// the clients as OutgoingError events, never returned: a malformed message
130// from one browser must not tear down the connection handling.
131func (b *Bridge) HandleIncoming(ctx context.Context, raw []byte) {
132 var msg Incoming
133 if err := json.Unmarshal(raw, &msg); err != nil {
134 b.broadcast(encode(Outgoing{Type: OutgoingError, Message: "malformed message: " + err.Error()}), false)
135 return
136 }
137
138 switch msg.Type {
139 case IncomingPrompt:
✨ Workspace panel, selectors, previews, desktop app, sandbox template, resizable file tree, light/dark theme 76d62ac k33g yesterday140 b.startTurn(msg.Text, msg.Attachments)
✨ Introduce new feature(s): ACP web client — Go backend (agent, bridge, httpserver, mockagent) + React SPA (Zed-like agent panel), tests, quality gate PASS 2434cc7 k33g yesterday141 case IncomingCancel:
142 b.cancelTurn(ctx)
143 case IncomingPermissionResponse:
144 b.resolvePermission(msg)
145 default:
146 b.broadcast(encode(Outgoing{Type: OutgoingError, Message: fmt.Sprintf("unknown message type %q", msg.Type)}), false)
147 }
148}
149
150// startTurn launches a prompt turn in the background; updates stream through
151// HandleSessionUpdate while it runs.
✨ Workspace panel, selectors, previews, desktop app, sandbox template, resizable file tree, light/dark theme 76d62ac k33g yesterday152func (b *Bridge) startTurn(text string, attachments []Attachment) {
✨ Introduce new feature(s): ACP web client — Go backend (agent, bridge, httpserver, mockagent) + React SPA (Zed-like agent panel), tests, quality gate PASS 2434cc7 k33g yesterday153 b.mu.Lock()
154 session := b.session
✨ Workspace panel, selectors, previews, desktop app, sandbox template, resizable file tree, light/dark theme 76d62ac k33g yesterday155 root := b.root
✨ Introduce new feature(s): ACP web client — Go backend (agent, bridge, httpserver, mockagent) + React SPA (Zed-like agent panel), tests, quality gate PASS 2434cc7 k33g yesterday156 if session == nil {
157 b.mu.Unlock()
158 b.broadcast(encode(Outgoing{Type: OutgoingError, Message: "no agent session"}), false)
159 return
160 }
161 if b.turnActive {
162 b.mu.Unlock()
163 b.broadcast(encode(Outgoing{Type: OutgoingError, Message: "a turn is already running"}), false)
164 return
165 }
166 b.turnActive = true
167 b.mu.Unlock()
168
✨ Workspace panel, selectors, previews, desktop app, sandbox template, resizable file tree, light/dark theme 76d62ac k33g yesterday169 b.broadcast(encode(Outgoing{Type: OutgoingUserMessage, Text: text, Attachments: attachments}), true)
✨ Introduce new feature(s): ACP web client — Go backend (agent, bridge, httpserver, mockagent) + React SPA (Zed-like agent panel), tests, quality gate PASS 2434cc7 k33g yesterday170 b.broadcast(encode(Outgoing{Type: OutgoingTurnStarted}), true)
171
✨ Workspace panel, selectors, previews, desktop app, sandbox template, resizable file tree, light/dark theme 76d62ac k33g yesterday172 blocks := promptBlocks(text, attachments, root)
✨ Introduce new feature(s): ACP web client — Go backend (agent, bridge, httpserver, mockagent) + React SPA (Zed-like agent panel), tests, quality gate PASS 2434cc7 k33g yesterday173 go func() {
174 // The turn outlives the WebSocket message that started it, so it
175 // runs under its own context, ended by session/cancel only.
✨ Workspace panel, selectors, previews, desktop app, sandbox template, resizable file tree, light/dark theme 76d62ac k33g yesterday176 stop, err := session.Prompt(context.Background(), blocks)
✨ Introduce new feature(s): ACP web client — Go backend (agent, bridge, httpserver, mockagent) + React SPA (Zed-like agent panel), tests, quality gate PASS 2434cc7 k33g yesterday177
178 b.mu.Lock()
179 b.turnActive = false
180 b.mu.Unlock()
181
182 if err != nil {
183 b.logger.Error("prompt failed", "error", err)
184 b.broadcast(encode(Outgoing{Type: OutgoingError, Message: "prompt failed: " + err.Error()}), true)
185 b.broadcast(encode(Outgoing{Type: OutgoingTurnEnded, StopReason: string(acp.StopReasonRefusal)}), true)
186 return
187 }
188 b.broadcast(encode(Outgoing{Type: OutgoingTurnEnded, StopReason: string(stop)}), true)
189 }()
190}
191
✨ Workspace panel, selectors, previews, desktop app, sandbox template, resizable file tree, light/dark theme 76d62ac k33g yesterday192// promptBlocks builds the ACP prompt: the text first, then one resource_link
193// per attachment (file:// URI, named after the mention), which every ACP
194// agent must accept.
195//
196// Example:
197//
198// promptBlocks("Explain @src/main.go", []Attachment{{Path: "src/main.go"}}, "/work")
199// // [text "Explain @src/main.go", resource_link name=src/main.go uri=file:///work/src/main.go]
200func promptBlocks(text string, attachments []Attachment, root string) []acp.ContentBlock {
201 blocks := []acp.ContentBlock{acp.TextBlock(text)}
202 for _, attachment := range attachments {
203 if attachment.Path == "" {
204 continue
205 }
206 path := attachment.Path
207 if !filepath.IsAbs(path) && root != "" {
208 path = filepath.Join(root, path)
209 }
210 name := attachment.Name
211 if name == "" {
212 name = attachment.Path
213 }
214 blocks = append(blocks, acp.ResourceLinkBlock(name, "file://"+filepath.ToSlash(path)))
215 }
216 return blocks
217}
218
✨ Introduce new feature(s): ACP web client — Go backend (agent, bridge, httpserver, mockagent) + React SPA (Zed-like agent panel), tests, quality gate PASS 2434cc7 k33g yesterday219func (b *Bridge) cancelTurn(ctx context.Context) {
220 b.mu.Lock()
221 session := b.session
222 b.mu.Unlock()
223 if session == nil {
224 return
225 }
226 if err := session.Cancel(ctx); err != nil {
227 b.logger.Error("cancel failed", "error", err)
228 b.broadcast(encode(Outgoing{Type: OutgoingError, Message: "cancel failed: " + err.Error()}), false)
229 }
230}
231
232func (b *Bridge) resolvePermission(msg Incoming) {
233 b.mu.Lock()
234 waiting, ok := b.pendingPerms[msg.RequestId]
235 if ok {
236 delete(b.pendingPerms, msg.RequestId)
237 delete(b.pendingRequests, msg.RequestId)
238 }
239 b.mu.Unlock()
240 if !ok {
241 // Either already answered by another client or unknown: ignore.
242 return
243 }
244 waiting <- permissionDecision{optionId: msg.OptionId, cancelled: msg.Cancelled}
245 b.broadcast(encode(Outgoing{Type: OutgoingPermissionResolved, RequestId: msg.RequestId}), true)
246}
247
248// HandleSessionUpdate implements agent.Handler: every ACP update is relayed
249// verbatim to the clients and recorded for replay.
250func (b *Bridge) HandleSessionUpdate(_ context.Context, notification acp.SessionNotification) {
251 update, err := json.Marshal(notification.Update)
252 if err != nil {
253 b.logger.Error("marshal session update", "error", err)
254 return
255 }
256 b.broadcast(encode(Outgoing{Type: OutgoingSessionUpdate, Update: update}), true)
257}
258
259// HandlePermissionRequest implements agent.Handler: the request is broadcast
260// to the clients and the call blocks until one of them answers or the agent
261// cancels the turn.
262func (b *Bridge) HandlePermissionRequest(ctx context.Context, request acp.RequestPermissionRequest) (acp.RequestPermissionResponse, error) {
263 rawRequest, err := json.Marshal(request)
264 if err != nil {
265 return acp.RequestPermissionResponse{}, fmt.Errorf("marshal permission request: %w", err)
266 }
267
268 decisionCh := make(chan permissionDecision, 1)
269 b.mu.Lock()
270 b.nextRequest++
271 requestId := "perm-" + strconv.Itoa(b.nextRequest)
272 event := encode(Outgoing{Type: OutgoingPermissionRequest, RequestId: requestId, Request: rawRequest})
273 b.pendingPerms[requestId] = decisionCh
274 b.pendingRequests[requestId] = event
275 b.mu.Unlock()
276
277 b.broadcast(event, false)
278
279 select {
280 case decision := <-decisionCh:
281 if decision.cancelled {
282 return acp.RequestPermissionResponse{Outcome: acp.NewRequestPermissionOutcomeCancelled()}, nil
283 }
284 return acp.RequestPermissionResponse{Outcome: acp.NewRequestPermissionOutcomeSelected(acp.PermissionOptionId(decision.optionId))}, nil
285 case <-ctx.Done():
286 // The agent cancelled the turn (or disconnected) while waiting.
287 b.mu.Lock()
288 delete(b.pendingPerms, requestId)
289 delete(b.pendingRequests, requestId)
290 b.mu.Unlock()
291 b.broadcast(encode(Outgoing{Type: OutgoingPermissionResolved, RequestId: requestId}), true)
292 return acp.RequestPermissionResponse{Outcome: acp.NewRequestPermissionOutcomeCancelled()}, nil
293 }
294}
295
296// broadcast sends an event to every connected client, disconnecting any
297// client whose buffer is full. With record=true the event also joins the
298// replay history.
299func (b *Bridge) broadcast(event []byte, record bool) {
300 b.mu.Lock()
301 defer b.mu.Unlock()
302
303 if record {
304 b.history = append(b.history, event)
305 if len(b.history) > maxHistory {
306 b.history = b.history[len(b.history)-maxHistory:]
307 }
308 }
309
310 for ch := range b.subscribers {
311 select {
312 case ch <- event:
313 default:
314 // The client stopped draining; drop it instead of blocking
315 // the agent's event stream.
316 delete(b.subscribers, ch)
317 close(ch)
318 b.logger.Warn("dropped a slow websocket client")
319 }
320 }
321}