nandi/ori · bridge_test.go

nandi/oripublic⑂ Fork 0

⑂ 2434cc7fb724f02b34c0189dfd5fb30c1a8a68e3

bridge_test.go · 342 lines · 9.8 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   1package bridge_test
4
"encoding/json"
8
acp "github.com/coder/acp-go-sdk"
"github.com/bots-garden/ori/internal/bridge"
16// fakeSession implements bridge.Prompter without any agent behind it.
17type fakeSession struct {
mu        sync.Mutex
prompts   []string
cancelled bool
// respond controls what PromptText returns; nil means end_turn.
respond func(text string) (acp.StopReason, error)
// block, when non-nil, is closed by the test to let PromptText return.
block chan struct{}
27func (f *fakeSession) ID() string { return "sess-fake" }
29func (f *fakeSession) PromptText(_ context.Context, text string) (acp.StopReason, error) {
f.mu.Lock()
f.prompts = append(f.prompts, text)
respond := f.respond
block := f.block
f.mu.Unlock()
if block != nil {
36
if respond != nil {
return respond(text)
return acp.StopReasonEndTurn, nil
44func (f *fakeSession) Cancel(context.Context) error {
f.mu.Lock()
defer f.mu.Unlock()
f.cancelled = true
return nil
51// collect drains a subscriber channel until the wanted message type shows up
52// or the timeout hits; it returns every message seen, decoded.
53func collect(t *testing.T, ch <-chan []byte, wantType string) []bridge.Outgoing {
t.Helper()
var seen []bridge.Outgoing
deadline := time.After(5 * time.Second)
58
case raw, ok := <-ch:
	if !ok {
		t.Fatalf("channel closed while waiting for %q; saw %+v", wantType, seen)
	var msg bridge.Outgoing
	if err := json.Unmarshal(raw, &msg); err != nil {
		t.Fatalf("undecodable outgoing message %q: %v", raw, err)
	seen = append(seen, msg)
	if msg.Type == wantType {
		return seen
case <-deadline:
	t.Fatalf("timed out waiting for %q; saw %+v", wantType, seen)
77func decode(t *testing.T, raw []byte) bridge.Outgoing {
t.Helper()
var msg bridge.Outgoing
if err := json.Unmarshal(raw, &msg); err != nil {
t.Fatalf("undecodable message %q: %v", raw, err)
return msg
86func TestSubscribeSendsHello(t *testing.T) {
b := bridge.New(nil)
b.SetSession(&fakeSession{})
_, replay, unsubscribe := b.Subscribe()
defer unsubscribe()
if len(replay) == 0 {
t.Fatal("Subscribe returned no replay events, want at least the hello")
hello := decode(t, replay[0])
if hello.Type != bridge.OutgoingHello || hello.SessionId != "sess-fake" {
t.Errorf("first replay event = %+v, want hello for sess-fake", hello)
102func TestPromptRunsTurnAndBroadcastsMarkers(t *testing.T) {
b := bridge.New(nil)
session := &fakeSession{}
b.SetSession(session)
events, _, unsubscribe := b.Subscribe()
defer unsubscribe()
b.HandleIncoming(context.Background(), []byte(`{"type":"prompt","text":"hello agent"}`))
seen := collect(t, events, bridge.OutgoingTurnEnded)
if seen[0].Type != bridge.OutgoingUserMessage || seen[0].Text != "hello agent" {
t.Errorf("first event = %+v, want the echoed user message", seen[0])
if seen[1].Type != bridge.OutgoingTurnStarted {
t.Errorf("second event = %q, want turn_started", seen[1].Type)
last := seen[len(seen)-1]
if last.StopReason != string(acp.StopReasonEndTurn) {
t.Errorf("turn_ended stopReason = %q, want end_turn", last.StopReason)
session.mu.Lock()
defer session.mu.Unlock()
if len(session.prompts) != 1 || session.prompts[0] != "hello agent" {
t.Errorf("session received prompts %v, want [hello agent]", session.prompts)
130func TestSecondPromptDuringTurnIsRejected(t *testing.T) {
b := bridge.New(nil)
session := &fakeSession{block: make(chan struct{})}
b.SetSession(session)
events, _, unsubscribe := b.Subscribe()
defer unsubscribe()
b.HandleIncoming(context.Background(), []byte(`{"type":"prompt","text":"first"}`))
collect(t, events, bridge.OutgoingTurnStarted)
b.HandleIncoming(context.Background(), []byte(`{"type":"prompt","text":"second"}`))
seen := collect(t, events, bridge.OutgoingError)
if last := seen[len(seen)-1]; last.Message != "a turn is already running" {
t.Errorf("error message = %q", last.Message)
close(session.block)
collect(t, events, bridge.OutgoingTurnEnded)
151func TestPromptWithoutSessionReportsError(t *testing.T) {
b := bridge.New(nil)
events, _, unsubscribe := b.Subscribe()
defer unsubscribe()
b.HandleIncoming(context.Background(), []byte(`{"type":"prompt","text":"x"}`))
seen := collect(t, events, bridge.OutgoingError)
if last := seen[len(seen)-1]; last.Message != "no agent session" {
t.Errorf("error message = %q, want \"no agent session\"", last.Message)
163func TestPromptFailureEndsTurnWithError(t *testing.T) {
b := bridge.New(nil)
b.SetSession(&fakeSession{respond: func(string) (acp.StopReason, error) {
return "", errors.New("agent exploded")
events, _, unsubscribe := b.Subscribe()
defer unsubscribe()
b.HandleIncoming(context.Background(), []byte(`{"type":"prompt","text":"x"}`))
seen := collect(t, events, bridge.OutgoingTurnEnded)
foundError := false
for _, msg := range seen {
if msg.Type == bridge.OutgoingError {
	foundError = true
if !foundError {
t.Error("no error event before turn_ended, want one mentioning the failure")
185func TestCancelReachesSession(t *testing.T) {
b := bridge.New(nil)
session := &fakeSession{}
b.SetSession(session)
b.HandleIncoming(context.Background(), []byte(`{"type":"cancel"}`))
session.mu.Lock()
defer session.mu.Unlock()
if !session.cancelled {
t.Error("cancel never reached the session")
199func TestMalformedAndUnknownMessagesReportErrors(t *testing.T) {
b := bridge.New(nil)
events, _, unsubscribe := b.Subscribe()
defer unsubscribe()
b.HandleIncoming(context.Background(), []byte(`{not json`))
collect(t, events, bridge.OutgoingError)
b.HandleIncoming(context.Background(), []byte(`{"type":"nope"}`))
seen := collect(t, events, bridge.OutgoingError)
if last := seen[len(seen)-1]; last.Message != `unknown message type "nope"` {
t.Errorf("error message = %q", last.Message)
214func TestSessionUpdateIsRelayedAndReplayed(t *testing.T) {
b := bridge.New(nil)
b.SetSession(&fakeSession{})
events, _, unsubscribe := b.Subscribe()
defer unsubscribe()
b.HandleSessionUpdate(context.Background(), acp.SessionNotification{
SessionId: "sess-fake",
Update:    acp.UpdateAgentMessageText("chunk!"),
seen := collect(t, events, bridge.OutgoingSessionUpdate)
update := seen[len(seen)-1].Update
if update == nil {
t.Fatal("session_update event has no update payload")
var acpUpdate struct {
SessionUpdate string `json:"sessionUpdate"`
Content       struct {
	Text string `json:"text"`
} `json:"content"`
if err := json.Unmarshal(update, &acpUpdate); err != nil {
t.Fatalf("update payload is not the raw ACP shape: %v", err)
if acpUpdate.SessionUpdate != "agent_message_chunk" || acpUpdate.Content.Text != "chunk!" {
t.Errorf("relayed update = %+v, want agent_message_chunk with text \"chunk!\"", acpUpdate)
// A client connecting after the fact receives the update in its replay.
_, replay, unsubscribeLate := b.Subscribe()
defer unsubscribeLate()
found := false
for _, raw := range replay {
if decode(t, raw).Type == bridge.OutgoingSessionUpdate {
	found = true
if !found {
t.Error("late subscriber's replay lacks the session update")
257func TestPermissionRoundTrip(t *testing.T) {
b := bridge.New(nil)
b.SetSession(&fakeSession{})
events, _, unsubscribe := b.Subscribe()
defer unsubscribe()
type result struct {
resp acp.RequestPermissionResponse
err  error
got := make(chan result, 1)
go func() {
resp, err := b.HandlePermissionRequest(context.Background(), acp.RequestPermissionRequest{
	SessionId: "sess-fake",
	Options:   []acp.PermissionOption{{OptionId: "allow", Name: "Allow", Kind: acp.PermissionOptionKindAllowOnce}},
got <- result{resp, err}
seen := collect(t, events, bridge.OutgoingPermissionRequest)
request := seen[len(seen)-1]
if request.RequestId == "" || request.Request == nil {
t.Fatalf("permission_request event incomplete: %+v", request)
b.HandleIncoming(context.Background(), []byte(`{"type":"permission_response","requestId":"`+request.RequestId+`","optionId":"allow"}`))
r := <-got
if r.err != nil {
t.Fatalf("HandlePermissionRequest returned an error: %v", r.err)
if r.resp.Outcome.Selected == nil || r.resp.Outcome.Selected.OptionId != "allow" {
t.Errorf("outcome = %+v, want selected \"allow\"", r.resp.Outcome)
collect(t, events, bridge.OutgoingPermissionResolved)
295func TestPermissionCancelledByAgentContext(t *testing.T) {
b := bridge.New(nil)
b.SetSession(&fakeSession{})
ctx, cancel := context.WithCancel(context.Background())
go cancel()
resp, err := b.HandlePermissionRequest(ctx, acp.RequestPermissionRequest{SessionId: "sess-fake"})
if err != nil {
t.Fatalf("HandlePermissionRequest returned an error: %v", err)
if resp.Outcome.Cancelled == nil {
t.Errorf("outcome = %+v, want cancelled", resp.Outcome)
311func TestPendingPermissionIsReplayedToLateJoiner(t *testing.T) {
b := bridge.New(nil)
b.SetSession(&fakeSession{})
answered := make(chan struct{})
go func() {
_, _ = b.HandlePermissionRequest(context.Background(), acp.RequestPermissionRequest{SessionId: "sess-fake"})
close(answered)
// Wait for the request to be registered by watching a live subscriber.
events, _, unsubscribe := b.Subscribe()
seen := collect(t, events, bridge.OutgoingPermissionRequest)
requestId := seen[len(seen)-1].RequestId
unsubscribe()
_, replay, unsubscribeLate := b.Subscribe()
defer unsubscribeLate()
found := false
for _, raw := range replay {
msg := decode(t, raw)
if msg.Type == bridge.OutgoingPermissionRequest && msg.RequestId == requestId {
	found = true
if !found {
t.Error("late subscriber's replay lacks the pending permission request")
b.HandleIncoming(context.Background(), []byte(`{"type":"permission_response","requestId":"`+requestId+`","cancelled":true}`))
<-answered

✨ 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 yesterday	1	package bridge_test
	2
	3	import (
	4	"context"
	5	"encoding/json"
	6	"errors"
	7	"sync"
	8	"testing"
	9	"time"
	10
	11	acp "github.com/coder/acp-go-sdk"
	12
	13	"github.com/bots-garden/ori/internal/bridge"
	14	)
	15
	16	// fakeSession implements bridge.Prompter without any agent behind it.
	17	type fakeSession struct {
	18	mu sync.Mutex
	19	prompts []string
	20	cancelled bool
	21	// respond controls what PromptText returns; nil means end_turn.
	22	respond func(text string) (acp.StopReason, error)
	23	// block, when non-nil, is closed by the test to let PromptText return.
	24	block chan struct{}
	25	}
	26
	27	func (f *fakeSession) ID() string { return "sess-fake" }
	28
	29	func (f *fakeSession) PromptText(_ context.Context, text string) (acp.StopReason, error) {
	30	f.mu.Lock()
	31	f.prompts = append(f.prompts, text)
	32	respond := f.respond
	33	block := f.block
	34	f.mu.Unlock()
	35	if block != nil {
	36	<-block
	37	}
	38	if respond != nil {
	39	return respond(text)
	40	}
	41	return acp.StopReasonEndTurn, nil
	42	}
	43
	44	func (f *fakeSession) Cancel(context.Context) error {
	45	f.mu.Lock()
	46	defer f.mu.Unlock()
	47	f.cancelled = true
	48	return nil
	49	}
	50
	51	// collect drains a subscriber channel until the wanted message type shows up
	52	// or the timeout hits; it returns every message seen, decoded.
	53	func collect(t *testing.T, ch <-chan []byte, wantType string) []bridge.Outgoing {
	54	t.Helper()
	55	var seen []bridge.Outgoing
	56	deadline := time.After(5 * time.Second)
	57	for {
	58	select {
	59	case raw, ok := <-ch:
	60	if !ok {
	61	t.Fatalf("channel closed while waiting for %q; saw %+v", wantType, seen)
	62	}
	63	var msg bridge.Outgoing
	64	if err := json.Unmarshal(raw, &msg); err != nil {
65	t.Fatalf("undecodable outgoing message %q: %v", raw, err)
66	}
67	seen = append(seen, msg)
68	if msg.Type == wantType {
69	return seen
70	}
71	case <-deadline:
72	t.Fatalf("timed out waiting for %q; saw %+v", wantType, seen)
73	}
74	}
75	}
76
77	func decode(t *testing.T, raw []byte) bridge.Outgoing {
78	t.Helper()
79	var msg bridge.Outgoing
80	if err := json.Unmarshal(raw, &msg); err != nil {
81	t.Fatalf("undecodable message %q: %v", raw, err)
82	}
83	return msg
84	}
85
86	func TestSubscribeSendsHello(t *testing.T) {
87	b := bridge.New(nil)
88	b.SetSession(&fakeSession{})
89
90	_, replay, unsubscribe := b.Subscribe()
91	defer unsubscribe()
92
93	if len(replay) == 0 {
94	t.Fatal("Subscribe returned no replay events, want at least the hello")
95	}
96	hello := decode(t, replay[0])
97	if hello.Type != bridge.OutgoingHello \|\| hello.SessionId != "sess-fake" {
98	t.Errorf("first replay event = %+v, want hello for sess-fake", hello)
99	}
100	}
101
102	func TestPromptRunsTurnAndBroadcastsMarkers(t *testing.T) {
103	b := bridge.New(nil)
104	session := &fakeSession{}
105	b.SetSession(session)
106
107	events, _, unsubscribe := b.Subscribe()
108	defer unsubscribe()
109
110	b.HandleIncoming(context.Background(), []byte(`{"type":"prompt","text":"hello agent"}`))
111
112	seen := collect(t, events, bridge.OutgoingTurnEnded)
113	if seen[0].Type != bridge.OutgoingUserMessage \|\| seen[0].Text != "hello agent" {
114	t.Errorf("first event = %+v, want the echoed user message", seen[0])
115	}
116	if seen[1].Type != bridge.OutgoingTurnStarted {
117	t.Errorf("second event = %q, want turn_started", seen[1].Type)
118	}
119	last := seen[len(seen)-1]
120	if last.StopReason != string(acp.StopReasonEndTurn) {
121	t.Errorf("turn_ended stopReason = %q, want end_turn", last.StopReason)
122	}
123	session.mu.Lock()
124	defer session.mu.Unlock()
125	if len(session.prompts) != 1 \|\| session.prompts[0] != "hello agent" {
126	t.Errorf("session received prompts %v, want [hello agent]", session.prompts)
127	}
128	}
129
130	func TestSecondPromptDuringTurnIsRejected(t *testing.T) {
131	b := bridge.New(nil)
132	session := &fakeSession{block: make(chan struct{})}
133	b.SetSession(session)
134
135	events, _, unsubscribe := b.Subscribe()
136	defer unsubscribe()
137
138	b.HandleIncoming(context.Background(), []byte(`{"type":"prompt","text":"first"}`))
139	collect(t, events, bridge.OutgoingTurnStarted)
140
141	b.HandleIncoming(context.Background(), []byte(`{"type":"prompt","text":"second"}`))
142	seen := collect(t, events, bridge.OutgoingError)
143	if last := seen[len(seen)-1]; last.Message != "a turn is already running" {
144	t.Errorf("error message = %q", last.Message)
145	}
146
147	close(session.block)
148	collect(t, events, bridge.OutgoingTurnEnded)
149	}
150
151	func TestPromptWithoutSessionReportsError(t *testing.T) {
152	b := bridge.New(nil)
153	events, _, unsubscribe := b.Subscribe()
154	defer unsubscribe()
155
156	b.HandleIncoming(context.Background(), []byte(`{"type":"prompt","text":"x"}`))
157	seen := collect(t, events, bridge.OutgoingError)
158	if last := seen[len(seen)-1]; last.Message != "no agent session" {
159	t.Errorf("error message = %q, want \"no agent session\"", last.Message)
160	}
161	}
162
163	func TestPromptFailureEndsTurnWithError(t *testing.T) {
164	b := bridge.New(nil)
165	b.SetSession(&fakeSession{respond: func(string) (acp.StopReason, error) {
166	return "", errors.New("agent exploded")
167	}})
168	events, _, unsubscribe := b.Subscribe()
169	defer unsubscribe()
170
171	b.HandleIncoming(context.Background(), []byte(`{"type":"prompt","text":"x"}`))
172	seen := collect(t, events, bridge.OutgoingTurnEnded)
173
174	foundError := false
175	for _, msg := range seen {
176	if msg.Type == bridge.OutgoingError {
177	foundError = true
178	}
179	}
180	if !foundError {
181	t.Error("no error event before turn_ended, want one mentioning the failure")
182	}
183	}
184
185	func TestCancelReachesSession(t *testing.T) {
186	b := bridge.New(nil)
187	session := &fakeSession{}
188	b.SetSession(session)
189
190	b.HandleIncoming(context.Background(), []byte(`{"type":"cancel"}`))
191
192	session.mu.Lock()
193	defer session.mu.Unlock()
194	if !session.cancelled {
195	t.Error("cancel never reached the session")
196	}
197	}
198
199	func TestMalformedAndUnknownMessagesReportErrors(t *testing.T) {
200	b := bridge.New(nil)
201	events, _, unsubscribe := b.Subscribe()
202	defer unsubscribe()
203
204	b.HandleIncoming(context.Background(), []byte(`{not json`))
205	collect(t, events, bridge.OutgoingError)
206
207	b.HandleIncoming(context.Background(), []byte(`{"type":"nope"}`))
208	seen := collect(t, events, bridge.OutgoingError)
209	if last := seen[len(seen)-1]; last.Message != `unknown message type "nope"` {
210	t.Errorf("error message = %q", last.Message)
211	}
212	}
213
214	func TestSessionUpdateIsRelayedAndReplayed(t *testing.T) {
215	b := bridge.New(nil)
216	b.SetSession(&fakeSession{})
217	events, _, unsubscribe := b.Subscribe()
218	defer unsubscribe()
219
220	b.HandleSessionUpdate(context.Background(), acp.SessionNotification{
221	SessionId: "sess-fake",
222	Update: acp.UpdateAgentMessageText("chunk!"),
223	})
224
225	seen := collect(t, events, bridge.OutgoingSessionUpdate)
226	update := seen[len(seen)-1].Update
227	if update == nil {
228	t.Fatal("session_update event has no update payload")
229	}
230	var acpUpdate struct {
231	SessionUpdate string `json:"sessionUpdate"`
232	Content struct {
233	Text string `json:"text"`
234	} `json:"content"`
235	}
236	if err := json.Unmarshal(update, &acpUpdate); err != nil {
237	t.Fatalf("update payload is not the raw ACP shape: %v", err)
238	}
239	if acpUpdate.SessionUpdate != "agent_message_chunk" \|\| acpUpdate.Content.Text != "chunk!" {
240	t.Errorf("relayed update = %+v, want agent_message_chunk with text \"chunk!\"", acpUpdate)
241	}
242
243	// A client connecting after the fact receives the update in its replay.
244	_, replay, unsubscribeLate := b.Subscribe()
245	defer unsubscribeLate()
246	found := false
247	for _, raw := range replay {
248	if decode(t, raw).Type == bridge.OutgoingSessionUpdate {
249	found = true
250	}
251	}
252	if !found {
253	t.Error("late subscriber's replay lacks the session update")
254	}
255	}
256
257	func TestPermissionRoundTrip(t *testing.T) {
258	b := bridge.New(nil)
259	b.SetSession(&fakeSession{})
260	events, _, unsubscribe := b.Subscribe()
261	defer unsubscribe()
262
263	type result struct {
264	resp acp.RequestPermissionResponse
265	err error
266	}
267	got := make(chan result, 1)
268	go func() {
269	resp, err := b.HandlePermissionRequest(context.Background(), acp.RequestPermissionRequest{
270	SessionId: "sess-fake",
271	Options: []acp.PermissionOption{{OptionId: "allow", Name: "Allow", Kind: acp.PermissionOptionKindAllowOnce}},
272	})
273	got <- result{resp, err}
274	}()
275
276	seen := collect(t, events, bridge.OutgoingPermissionRequest)
277	request := seen[len(seen)-1]
278	if request.RequestId == "" \|\| request.Request == nil {
279	t.Fatalf("permission_request event incomplete: %+v", request)
280	}
281
282	b.HandleIncoming(context.Background(), []byte(`{"type":"permission_response","requestId":"`+request.RequestId+`","optionId":"allow"}`))
283
284	r := <-got
285	if r.err != nil {
286	t.Fatalf("HandlePermissionRequest returned an error: %v", r.err)
287	}
288	if r.resp.Outcome.Selected == nil \|\| r.resp.Outcome.Selected.OptionId != "allow" {
289	t.Errorf("outcome = %+v, want selected \"allow\"", r.resp.Outcome)
290	}
291
292	collect(t, events, bridge.OutgoingPermissionResolved)
293	}
294
295	func TestPermissionCancelledByAgentContext(t *testing.T) {
296	b := bridge.New(nil)
297	b.SetSession(&fakeSession{})
298
299	ctx, cancel := context.WithCancel(context.Background())
300	go cancel()
301
302	resp, err := b.HandlePermissionRequest(ctx, acp.RequestPermissionRequest{SessionId: "sess-fake"})
303	if err != nil {
304	t.Fatalf("HandlePermissionRequest returned an error: %v", err)
305	}
306	if resp.Outcome.Cancelled == nil {
307	t.Errorf("outcome = %+v, want cancelled", resp.Outcome)
308	}
309	}
310
311	func TestPendingPermissionIsReplayedToLateJoiner(t *testing.T) {
312	b := bridge.New(nil)
313	b.SetSession(&fakeSession{})
314
315	answered := make(chan struct{})
316	go func() {
317	_, _ = b.HandlePermissionRequest(context.Background(), acp.RequestPermissionRequest{SessionId: "sess-fake"})
318	close(answered)
319	}()
320
321	// Wait for the request to be registered by watching a live subscriber.
322	events, _, unsubscribe := b.Subscribe()
323	seen := collect(t, events, bridge.OutgoingPermissionRequest)
324	requestId := seen[len(seen)-1].RequestId
325	unsubscribe()
326
327	_, replay, unsubscribeLate := b.Subscribe()
328	defer unsubscribeLate()
329	found := false
330	for _, raw := range replay {
331	msg := decode(t, raw)
332	if msg.Type == bridge.OutgoingPermissionRequest && msg.RequestId == requestId {
333	found = true
334	}
335	}
336	if !found {
337	t.Error("late subscriber's replay lacks the pending permission request")
338	}
339
340	b.HandleIncoming(context.Background(), []byte(`{"type":"permission_response","requestId":"`+requestId+`","cancelled":true}`))
341	<-answered
342	}