返回 DeepSeek-Reasonix
server_test.go
根目录 / internal / acp / server_test.go
1 package acp
2
3 import (
4 "bytes"
5 "context"
6 "encoding/json"
7 "errors"
8 "io"
9 "os"
10 "path/filepath"
11 "strings"
12 "sync"
13 "testing"
14 "time"
15
16 "reasonix/internal/agent"
17 "reasonix/internal/agent/testutil"
18 "reasonix/internal/command"
19 "reasonix/internal/control"
20 "reasonix/internal/event"
21 "reasonix/internal/hook"
22 "reasonix/internal/jobs"
23 "reasonix/internal/provider"
24 "reasonix/internal/skill"
25 "reasonix/internal/tool"
26 )
27
28 // --- fakes: a Factory wrapping a behavior-driven runner in a real Controller ---
29
30 // fakeRunner stands in for an agent.Runner; it emits to the session's sink and
31 // honors ctx cancellation, but runs no model.
32 type fakeRunner struct {
33 sink event.Sink
34 behavior func(ctx context.Context, sink event.Sink, input string) error
35 }
36
37 func (r *fakeRunner) Run(ctx context.Context, input string) error {
38 return r.behavior(ctx, r.sink, input)
39 }
40
41 // fakeFactory builds a real control.Controller around the fake runner, so the
42 // service exercises the actual controller surface (Run/Cancel/Close) it uses.
43 type fakeFactory struct {
44 behavior func(ctx context.Context, sink event.Sink, input string) error
45 }
46
47 func (f *fakeFactory) NewSession(_ context.Context, p SessionParams) (*control.Controller, error) {
48 runner := &fakeRunner{sink: p.Sink, behavior: f.behavior}
49 return control.New(control.Options{Runner: runner, Sink: p.Sink}), nil
50 }
51
52 type steerBarrierTool struct {
53 started chan struct{}
54 release chan struct{}
55 }
56
57 func (t *steerBarrierTool) Name() string { return "steer_barrier" }
58 func (t *steerBarrierTool) Description() string { return "waits for a steer" }
59 func (t *steerBarrierTool) Schema() json.RawMessage {
60 return json.RawMessage(`{"type":"object","properties":{}}`)
61 }
62 func (t *steerBarrierTool) ReadOnly() bool { return true }
63 func (t *steerBarrierTool) Execute(ctx context.Context, _ json.RawMessage) (string, error) {
64 close(t.started)
65 select {
66 case <-t.release:
67 return "released", nil
68 case <-ctx.Done():
69 return "", ctx.Err()
70 }
71 }
72
73 type steerFactory struct {
74 provider *testutil.MockProvider
75 barrier *steerBarrierTool
76 }
77
78 func (f *steerFactory) NewSession(_ context.Context, p SessionParams) (*control.Controller, error) {
79 tools := tool.NewRegistry()
80 tools.Add(f.barrier)
81 executor := agent.New(f.provider, tools, agent.NewSession(""), agent.Options{MaxSteps: 2}, p.Sink)
82 return control.New(control.Options{Runner: executor, Executor: executor, Sink: p.Sink}), nil
83 }
84
85 type commandFactory struct {
86 commands []command.Command
87 skills []skill.Skill
88 seen chan string
89 dir string
90 }
91
92 func (f *commandFactory) SessionDir() string { return f.dir }
93
94 func (f *commandFactory) NewSession(_ context.Context, p SessionParams) (*control.Controller, error) {
95 runner := &fakeRunner{
96 sink: p.Sink,
97 behavior: func(_ context.Context, sink event.Sink, input string) error {
98 f.seen <- input
99 sink.Emit(event.Event{Kind: event.Text, Text: input})
100 return nil
101 },
102 }
103 return control.New(control.Options{Runner: runner, Sink: p.Sink, Commands: f.commands, Skills: f.skills, SessionDir: f.dir}), nil
104 }
105
106 type configurableFactory struct {
107 mu sync.Mutex
108 builds []SessionParams
109 dir string
110 withHooks bool
111 hookEvents []hook.Event
112 behavior func(ctx context.Context, sink event.Sink, input string, p SessionParams) error
113 managers []*jobs.Manager
114 withCtrl func(ctx context.Context, sink event.Sink, input string, p SessionParams, ctrl *control.Controller) error
115 onBuild func(index int, p SessionParams)
116 }
117
118 func (f *configurableFactory) NewSession(_ context.Context, p SessionParams) (*control.Controller, error) {
119 f.mu.Lock()
120 buildIndex := len(f.builds)
121 f.builds = append(f.builds, SessionParams{
122 Cwd: p.Cwd,
123 Model: p.Model,
124 EffortOverride: cloneStringPtr(p.EffortOverride),
125 RuntimeProfile: p.RuntimeProfile,
126 FileOverlay: p.FileOverlay,
127 Terminal: p.Terminal,
128 })
129 onBuild := f.onBuild
130 f.mu.Unlock()
131 if onBuild != nil {
132 onBuild(buildIndex, p)
133 }
134 behavior := f.behavior
135 if behavior == nil {
136 behavior = func(_ context.Context, sink event.Sink, input string, p SessionParams) error {
137 sink.Emit(event.Event{Kind: event.Text, Text: p.Model + ":" + input})
138 return nil
139 }
140 }
141 var ctrl *control.Controller
142 runner := &fakeRunner{
143 sink: p.Sink,
144 behavior: func(ctx context.Context, sink event.Sink, input string) error {
145 if f.withCtrl != nil {
146 return f.withCtrl(ctx, sink, input, p, ctrl)
147 }
148 return behavior(ctx, sink, input, p)
149 },
150 }
151 opts := control.Options{Runner: runner, Sink: p.Sink, SessionDir: f.dir, OnSessionRecovered: p.OnSessionRecovered}
152 if f.withHooks {
153 opts.Hooks = f.hookRunner()
154 }
155 if f.managers != nil {
156 jm := jobs.NewManager(event.Discard)
157 f.mu.Lock()
158 f.managers = append(f.managers, jm)
159 f.mu.Unlock()
160 opts.Jobs = jm
161 }
162 ctrl = control.New(opts)
163 return ctrl, nil
164 }
165
166 func (f *configurableFactory) SessionDir() string { return f.dir }
167
168 type teardownFactory struct {
169 dir string
170 grace time.Duration
171 mu sync.Mutex
172 manager *jobs.Manager
173 }
174
175 func (f *teardownFactory) SessionDir() string { return f.dir }
176
177 func (f *teardownFactory) NewSession(_ context.Context, p SessionParams) (*control.Controller, error) {
178 jm := jobs.NewManager(event.Discard, jobs.WithTeardownGrace(f.grace))
179 f.mu.Lock()
180 f.manager = jm
181 f.mu.Unlock()
182 runner := &fakeRunner{
183 sink: p.Sink,
184 behavior: func(context.Context, event.Sink, string) error { return nil },
185 }
186 return control.New(control.Options{
187 Runner: runner,
188 Sink: p.Sink,
189 SessionDir: f.dir,
190 Jobs: jm,
191 }), nil
192 }
193
194 func (f *teardownFactory) lastManager(t *testing.T) *jobs.Manager {
195 t.Helper()
196 f.mu.Lock()
197 defer f.mu.Unlock()
198 if f.manager == nil {
199 t.Fatal("session manager was not created")
200 }
201 return f.manager
202 }
203
204 func (f *configurableFactory) SessionConfigState(_ context.Context, p SessionConfigStateParams) (SessionConfigState, error) {
205 model := strings.TrimSpace(p.Model)
206 if model == "" {
207 model = "fast"
208 }
209 if model != "fast" && model != "pro" {
210 return SessionConfigState{}, os.ErrInvalid
211 }
212 effort := "auto"
213 effortOverride := cloneStringPtr(p.EffortOverride)
214 if effortOverride != nil && *effortOverride != "" {
215 effort = *effortOverride
216 }
217 modelOptions := []SessionConfigSelectOption{
218 {Value: "fast", Name: "Fast"},
219 {Value: "pro", Name: "Pro"},
220 }
221 effortOptions := []SessionConfigSelectOption{
222 {Value: "auto", Name: "Auto"},
223 {Value: "high", Name: "High"},
224 }
225 runtimeProfile := strings.TrimSpace(p.RuntimeProfile)
226 if runtimeProfile == "" || runtimeProfile == "full" {
227 runtimeProfile = "balanced"
228 }
229 if runtimeProfile != "economy" && runtimeProfile != "balanced" && runtimeProfile != "delivery" {
230 return SessionConfigState{}, os.ErrInvalid
231 }
232 return SessionConfigState{
233 Model: model,
234 EffortOverride: effortOverride,
235 RuntimeProfile: runtimeProfile,
236 Models: &SessionModelState{
237 AvailableModels: []ModelInfo{{ModelID: "fast", Name: "Fast"}, {ModelID: "pro", Name: "Pro"}},
238 CurrentModelID: model,
239 },
240 ConfigOptions: []SessionConfigOption{
241 {ID: "model", Name: "Model", Category: "model", Type: "select", CurrentValue: model, Options: modelOptions},
242 {ID: "effort", Name: "Effort", Category: "thought_level", Type: "select", CurrentValue: effort, Options: effortOptions},
243 {ID: "work_mode", Name: "Work Mode", Category: "work_mode", Type: "select", CurrentValue: runtimeProfile, Options: []SessionConfigSelectOption{
244 {Value: "economy", Name: "Economy"}, {Value: "balanced", Name: "Balanced"}, {Value: "delivery", Name: "Delivery"},
245 }},
246 },
247 }, nil
248 }
249
250 func (f *configurableFactory) buildAt(t *testing.T, idx int) SessionParams {
251 t.Helper()
252 f.mu.Lock()
253 defer f.mu.Unlock()
254 if len(f.builds) <= idx {
255 t.Fatalf("builds = %d, want index %d", len(f.builds), idx)
256 }
257 return f.builds[idx]
258 }
259
260 func (f *configurableFactory) buildCount() int {
261 f.mu.Lock()
262 defer f.mu.Unlock()
263 return len(f.builds)
264 }
265
266 func (f *configurableFactory) managerAt(t *testing.T, idx int) *jobs.Manager {
267 t.Helper()
268 f.mu.Lock()
269 defer f.mu.Unlock()
270 if f.managers == nil {
271 t.Fatal("factory does not create job managers")
272 }
273 if len(f.managers) <= idx {
274 t.Fatalf("builds = %d, want manager index %d", len(f.builds), idx)
275 }
276 return f.managers[idx]
277 }
278
279 func (f *configurableFactory) hookRunner() *hook.Runner {
280 hooks := []hook.ResolvedHook{
281 {HookConfig: hook.HookConfig{Command: "session-start"}, Event: hook.SessionStart},
282 {HookConfig: hook.HookConfig{Command: "session-end"}, Event: hook.SessionEnd},
283 }
284 return hook.NewRunner(hooks, "", func(_ context.Context, in hook.SpawnInput) hook.SpawnResult {
285 var payload hook.Payload
286 _ = json.Unmarshal([]byte(in.Stdin), &payload)
287 f.mu.Lock()
288 f.hookEvents = append(f.hookEvents, payload.Event)
289 f.mu.Unlock()
290 return hook.SpawnResult{ExitCode: 0}
291 }, nil)
292 }
293
294 func (f *configurableFactory) hookEventsSnapshot() []hook.Event {
295 f.mu.Lock()
296 defer f.mu.Unlock()
297 return append([]hook.Event(nil), f.hookEvents...)
298 }
299
300 // --- a minimal JSON-RPC client over the wire, for integration tests ---
301
302 type frame struct {
303 ID *json.RawMessage `json:"id"`
304 Method string `json:"method"`
305 Params json.RawMessage `json:"params"`
306 Result json.RawMessage `json:"result"`
307 Error *rpcError `json:"error"`
308 }
309
310 type rpcClient struct {
311 enc *json.Encoder
312 wmu sync.Mutex
313
314 mu sync.Mutex
315 nextID int64
316 waits map[int64]chan frame
317
318 notifs chan frame
319 reqs chan frame
320 }
321
322 func newRPCClient(in io.Writer, out io.Reader) *rpcClient {
323 c := &rpcClient{
324 enc: json.NewEncoder(in),
325 waits: make(map[int64]chan frame),
326 notifs: make(chan frame, 64),
327 reqs: make(chan frame, 16),
328 }
329 dec := json.NewDecoder(out)
330 go func() {
331 for {
332 var f frame
333 if err := dec.Decode(&f); err != nil {
334 return
335 }
336 switch {
337 case f.Method != "" && f.ID != nil:
338 c.reqs <- f
339 case f.Method != "" && f.ID == nil:
340 c.notifs <- f
341 case f.Method == "" && f.ID != nil:
342 var id int64
343 if json.Unmarshal(*f.ID, &id) != nil {
344 continue
345 }
346 c.mu.Lock()
347 ch := c.waits[id]
348 delete(c.waits, id)
349 c.mu.Unlock()
350 if ch != nil {
351 ch <- f
352 }
353 }
354 }
355 }()
356 return c
357 }
358
359 func (c *rpcClient) send(v any) {
360 c.wmu.Lock()
361 _ = c.enc.Encode(v)
362 c.wmu.Unlock()
363 }
364
365 func (c *rpcClient) callAsync(method string, params any) chan frame {
366 c.mu.Lock()
367 c.nextID++
368 id := c.nextID
369 ch := make(chan frame, 1)
370 c.waits[id] = ch
371 c.mu.Unlock()
372 c.send(map[string]any{"jsonrpc": "2.0", "id": id, "method": method, "params": params})
373 return ch
374 }
375
376 func (c *rpcClient) call(t *testing.T, method string, params any) frame {
377 t.Helper()
378 select {
379 case f := <-c.callAsync(method, params):
380 return f
381 case <-time.After(2 * time.Second):
382 t.Fatalf("%s: timed out", method)
383 return frame{}
384 }
385 }
386
387 func (c *rpcClient) notify(method string, params any) {
388 c.send(map[string]any{"jsonrpc": "2.0", "method": method, "params": params})
389 }
390
391 func (c *rpcClient) reply(id *json.RawMessage, result any) {
392 c.send(map[string]any{"jsonrpc": "2.0", "id": id, "result": result})
393 }
394
395 func (c *rpcClient) replyError(id *json.RawMessage, code int, message string) {
396 c.send(map[string]any{"jsonrpc": "2.0", "id": id, "error": rpcError{Code: code, Message: message}})
397 }
398
399 func startServer(t *testing.T, factory Factory) (*rpcClient, func()) {
400 t.Helper()
401 inR, inW := io.Pipe()
402 outR, outW := io.Pipe()
403 done := make(chan struct{})
404 go func() {
405 _ = Serve(context.Background(), inR, outW, factory, AgentInfo{Name: "reasonix-test", Version: "0"})
406 close(done)
407 }()
408 client := newRPCClient(inW, outR)
409 return client, func() {
410 _ = inW.Close()
411 <-done
412 _ = outW.Close()
413 }
414 }
415
416 type orderedRPCClient struct {
417 enc *json.Encoder
418 frames chan frame
419 }
420
421 func newOrderedRPCClient(in io.Writer, out io.Reader) *orderedRPCClient {
422 c := &orderedRPCClient{enc: json.NewEncoder(in), frames: make(chan frame, 16)}
423 dec := json.NewDecoder(out)
424 go func() {
425 defer close(c.frames)
426 for {
427 var f frame
428 if err := dec.Decode(&f); err != nil {
429 return
430 }
431 c.frames <- f
432 }
433 }()
434 return c
435 }
436
437 func (c *orderedRPCClient) send(t *testing.T, id int, method string, params any) {
438 t.Helper()
439 if err := c.enc.Encode(map[string]any{"jsonrpc": "2.0", "id": id, "method": method, "params": params}); err != nil {
440 t.Fatalf("%s: send: %v", method, err)
441 }
442 }
443
444 func (c *orderedRPCClient) next(t *testing.T) frame {
445 t.Helper()
446 select {
447 case f, ok := <-c.frames:
448 if !ok {
449 t.Fatal("ACP output closed")
450 }
451 return f
452 case <-time.After(2 * time.Second):
453 t.Fatal("timed out waiting for ACP frame")
454 return frame{}
455 }
456 }
457
458 func startOrderedServer(t *testing.T, factory Factory) (*orderedRPCClient, func()) {
459 t.Helper()
460 inR, inW := io.Pipe()
461 outR, outW := io.Pipe()
462 done := make(chan struct{})
463 go func() {
464 _ = Serve(context.Background(), inR, outW, factory, AgentInfo{Name: "reasonix-test", Version: "0"})
465 close(done)
466 }()
467 client := newOrderedRPCClient(inW, outR)
468 return client, func() {
469 _ = inW.Close()
470 <-done
471 _ = outW.Close()
472 }
473 }
474
475 func requireResponseFrame(t *testing.T, f frame, id int) {
476 t.Helper()
477 if f.Method != "" || f.ID == nil {
478 t.Fatalf("first frame = %+v, want response %d", f, id)
479 }
480 if f.Error != nil {
481 t.Fatalf("response %d error = %+v", id, f.Error)
482 }
483 var gotID int
484 if err := json.Unmarshal(*f.ID, &gotID); err != nil || gotID != id {
485 t.Fatalf("response id = %d (%v), want %d", gotID, err, id)
486 }
487 }
488
489 func requireAvailableCommandsFrame(t *testing.T, f frame) {
490 t.Helper()
491 if f.Method != "session/update" || f.ID != nil {
492 t.Fatalf("second frame = %+v, want session/update notification", f)
493 }
494 var params map[string]any
495 if err := json.Unmarshal(f.Params, &params); err != nil {
496 t.Fatalf("available commands update: %v", err)
497 }
498 update, ok := params["update"].(map[string]any)
499 if !ok {
500 t.Fatalf("session update payload = %#v, want object", params["update"])
501 }
502 if got := update["sessionUpdate"]; got != "available_commands_update" {
503 t.Fatalf("session update = %v, want available_commands_update", got)
504 }
505 }
506
507 // drainPrompt collects session/update notifications until the prompt's response
508 // arrives, then sweeps any notifications still buffered.
509 func drainPrompt(t *testing.T, c *rpcClient, promptCh chan frame) ([]frame, frame) {
510 t.Helper()
511 var notifs []frame
512 var resp frame
513 for {
514 select {
515 case f := <-c.notifs:
516 notifs = append(notifs, f)
517 case resp = <-promptCh:
518 for {
519 select {
520 case f := <-c.notifs:
521 notifs = append(notifs, f)
522 default:
523 return notifs, resp
524 }
525 }
526 // A full prompt crosses the ACP server, controller, agent, and transcript
527 // persistence path. Loaded Windows release runners can leave that
528 // asynchronous pipeline idle for more than two seconds, so keep a
529 // generous but bounded responsiveness limit for the end-to-end helper.
530 case <-time.After(5 * time.Second):
531 t.Fatal("session/prompt: timed out")
532 }
533 }
534 }
535
536 func updateKind(t *testing.T, f frame) string {
537 t.Helper()
538 var p struct {
539 Update struct {
540 SessionUpdate string `json:"sessionUpdate"`
541 } `json:"update"`
542 }
543 if err := json.Unmarshal(f.Params, &p); err != nil {
544 t.Fatalf("decode update: %v", err)
545 }
546 return p.Update.SessionUpdate
547 }
548
549 func configOptionValueFromUpdate(t *testing.T, f frame, id string) (string, bool) {
550 t.Helper()
551 var p struct {
552 Update struct {
553 SessionUpdate string `json:"sessionUpdate"`
554 ConfigOptions []SessionConfigOption `json:"configOptions"`
555 } `json:"update"`
556 }
557 if err := json.Unmarshal(f.Params, &p); err != nil {
558 t.Fatalf("decode config update: %v", err)
559 }
560 if p.Update.SessionUpdate != "config_option_update" {
561 return "", false
562 }
563 opt, ok := findConfigOption(p.Update.ConfigOptions, id)
564 if !ok {
565 return "", false
566 }
567 return opt.CurrentValue, true
568 }
569
570 func messageChunkText(t *testing.T, f frame) (string, bool) {
571 t.Helper()
572 var p struct {
573 Update struct {
574 SessionUpdate string `json:"sessionUpdate"`
575 Content ContentBlock `json:"content"`
576 } `json:"update"`
577 }
578 if err := json.Unmarshal(f.Params, &p); err != nil {
579 t.Fatalf("decode message update: %v", err)
580 }
581 if p.Update.SessionUpdate != "agent_message_chunk" || p.Update.Content.Type != "text" {
582 return "", false
583 }
584 return p.Update.Content.Text, true
585 }
586
587 // --- tests ---
588
589 func TestServeLifecycle(t *testing.T) {
590 factory := &fakeFactory{behavior: func(_ context.Context, sink event.Sink, input string) error {
591 sink.Emit(event.Event{Kind: event.Text, Text: "hi " + input})
592 sink.Emit(event.Event{Kind: event.ToolDispatch, Tool: event.Tool{ID: "c1", Name: "ls", Args: `{}`}})
593 sink.Emit(event.Event{Kind: event.ToolResult, Tool: event.Tool{ID: "c1", Name: "ls", Output: "file.go"}})
594 return nil
595 }}
596 client, stop := startServer(t, factory)
597 defer stop()
598
599 initResp := client.call(t, "initialize", InitializeParams{ProtocolVersion: 1})
600 var ir InitializeResult
601 if err := json.Unmarshal(initResp.Result, &ir); err != nil {
602 t.Fatalf("initialize result: %v", err)
603 }
604 if ir.ProtocolVersion != ProtocolVersion {
605 t.Errorf("protocolVersion = %d, want %d", ir.ProtocolVersion, ProtocolVersion)
606 }
607 if !ir.AgentCapabilities.PromptCapabilities.EmbeddedContext {
608 t.Errorf("embeddedContext should be advertised")
609 }
610 if ir.AgentCapabilities.SessionCapabilities.List == nil ||
611 ir.AgentCapabilities.SessionCapabilities.Resume == nil ||
612 ir.AgentCapabilities.SessionCapabilities.Close == nil ||
613 ir.AgentCapabilities.SessionCapabilities.Delete == nil {
614 t.Errorf("sessionCapabilities = %+v, want list/resume/close/delete", ir.AgentCapabilities.SessionCapabilities)
615 }
616 if ir.AgentCapabilities.PromptCapabilities.Image {
617 t.Errorf("image must not be advertised")
618 }
619 var extensions struct {
620 AgentCapabilities struct {
621 Meta map[string]ReasonixExtensionCapabilities `json:"_meta"`
622 } `json:"agentCapabilities"`
623 }
624 if err := json.Unmarshal(initResp.Result, &extensions); err != nil {
625 t.Fatalf("initialize extensions: %v", err)
626 }
627 steer := extensions.AgentCapabilities.Meta["reasonix.io"].SessionSteer
628 if steer == nil || steer.Method != sessionSteerMethod {
629 t.Errorf("sessionSteer capability = %+v, want method %q", steer, sessionSteerMethod)
630 }
631 for _, method := range []string{sessionStatusMethod, sessionStatusUpdateMethod} {
632 capability, ok := ir.AgentCapabilities.Meta[method].(map[string]any)
633 if !ok || capability["schemaVersion"] != float64(reasonixStatusSchemaVersion) {
634 t.Errorf("%s capability = %#v, want schemaVersion %d", method, ir.AgentCapabilities.Meta[method], reasonixStatusSchemaVersion)
635 }
636 }
637 if len(ir.AuthMethods) != 1 || ir.AuthMethods[0].ID != "reasonix-setup" || ir.AuthMethods[0].Type != "terminal" {
638 t.Fatalf("authMethods = %+v, want terminal reasonix setup", ir.AuthMethods)
639 }
640 if len(ir.AuthMethods[0].Args) != 1 || ir.AuthMethods[0].Args[0] != "setup" {
641 t.Fatalf("auth args = %+v, want [setup]", ir.AuthMethods[0].Args)
642 }
643
644 authResp := client.call(t, "authenticate", AuthenticateParams{MethodID: "reasonix-setup"})
645 if authResp.Error != nil {
646 t.Fatalf("authenticate errored: %+v", authResp.Error)
647 }
648 badAuthResp := client.call(t, "authenticate", AuthenticateParams{MethodID: "missing"})
649 if badAuthResp.Error == nil || badAuthResp.Error.Code != ErrInvalidParams {
650 t.Fatalf("bad authenticate = %+v, want invalid params", badAuthResp.Error)
651 }
652
653 newResp := client.call(t, "session/new", SessionNewParams{Cwd: t.TempDir()})
654 var nr SessionNewResult
655 if err := json.Unmarshal(newResp.Result, &nr); err != nil || nr.SessionID == "" {
656 t.Fatalf("session/new result: %v (%q)", err, nr.SessionID)
657 }
658
659 promptCh := client.callAsync("session/prompt", SessionPromptParams{
660 SessionID: nr.SessionID,
661 Prompt: []ContentBlock{{Type: "text", Text: "there"}},
662 })
663 notifs, resp := drainPrompt(t, client, promptCh)
664
665 kinds := map[string]bool{}
666 for _, n := range notifs {
667 kinds[updateKind(t, n)] = true
668 }
669 for _, want := range []string{"agent_message_chunk", "tool_call", "tool_call_update"} {
670 if !kinds[want] {
671 t.Errorf("missing %s update; saw %v", want, kinds)
672 }
673 }
674 var pr SessionPromptResult
675 if err := json.Unmarshal(resp.Result, &pr); err != nil {
676 t.Fatalf("prompt result: %v", err)
677 }
678 if pr.StopReason != StopEndTurn {
679 t.Errorf("stopReason = %q, want %q", pr.StopReason, StopEndTurn)
680 }
681 }
682
683 func TestServeAdvertisesAndExpandsCustomCommands(t *testing.T) {
684 factory := &commandFactory{
685 seen: make(chan string, 1),
686 skills: []skill.Skill{{
687 Name: "writing-plans", Plugin: "superpowers", Description: "Write a plan", Body: "Plan $ARGUMENTS",
688 }},
689 commands: []command.Command{
690 {
691 Name: "review",
692 Description: "Review the target",
693 ArgHint: "path",
694 Body: "Review $1",
695 },
696 {
697 Name: "plan",
698 Body: "Plan $ARGUMENTS",
699 Hidden: true,
700 },
701 },
702 }
703 client, stop := startServer(t, factory)
704 defer stop()
705
706 client.call(t, "initialize", InitializeParams{ProtocolVersion: 1})
707 newResp := client.call(t, "session/new", SessionNewParams{Cwd: t.TempDir()})
708 var nr SessionNewResult
709 if err := json.Unmarshal(newResp.Result, &nr); err != nil || nr.SessionID == "" {
710 t.Fatalf("session/new result: %v (%q)", err, nr.SessionID)
711 }
712
713 var advertised bool
714 var hiddenAdvertised bool
715 var pluginSkillAdvertised bool
716 select {
717 case n := <-client.notifs:
718 var p struct {
719 Update struct {
720 SessionUpdate string `json:"sessionUpdate"`
721 AvailableCommands []AvailableCommand `json:"availableCommands"`
722 } `json:"update"`
723 }
724 if err := json.Unmarshal(n.Params, &p); err != nil {
725 t.Fatalf("available commands update: %v", err)
726 }
727 for _, cmd := range p.Update.AvailableCommands {
728 if cmd.Name == "superpowers:writing-plans" {
729 pluginSkillAdvertised = true
730 }
731 if cmd.Name == "writing-plans" {
732 hiddenAdvertised = true
733 }
734 if cmd.Name == "plan" {
735 hiddenAdvertised = true
736 }
737 if p.Update.SessionUpdate == "available_commands_update" &&
738 cmd.Name == "review" &&
739 cmd.Description == "Review the target" &&
740 cmd.Input != nil &&
741 cmd.Input.Hint == "path" {
742 advertised = true
743 }
744 }
745 case <-time.After(2 * time.Second):
746 t.Fatal("timed out waiting for available_commands_update")
747 }
748 if !advertised {
749 t.Fatal("review command was not advertised")
750 }
751 if hiddenAdvertised {
752 t.Fatal("hidden compatibility command was advertised")
753 }
754 if !pluginSkillAdvertised {
755 t.Fatal("qualified plugin skill was not advertised")
756 }
757
758 promptCh := client.callAsync("session/prompt", SessionPromptParams{
759 SessionID: nr.SessionID,
760 Prompt: []ContentBlock{{Type: "text", Text: "/review src/main.go"}},
761 })
762 _, resp := drainPrompt(t, client, promptCh)
763 if resp.Error != nil {
764 t.Fatalf("prompt errored: %+v", resp.Error)
765 }
766 select {
767 case got := <-factory.seen:
768 if got != "Review src/main.go" {
769 t.Fatalf("runner input = %q, want expanded command", got)
770 }
771 case <-time.After(2 * time.Second):
772 t.Fatal("runner did not receive prompt")
773 }
774 }
775
776 func TestServeRequestRunsAfterResponseHookAfterWritingResult(t *testing.T) {
777 var buf bytes.Buffer
778 conn := NewConn(strings.NewReader(""), &buf)
779 conn.Handle("test/hook", func(context.Context, json.RawMessage) (any, error) {
780 return afterResponse{
781 result: map[string]string{"ok": "yes"},
782 after: func() {
783 _ = conn.Notify("test/notification", map[string]string{"after": "yes"})
784 },
785 }, nil
786 })
787
788 conn.serveRequest(context.Background(), json.RawMessage("1"), "test/hook", nil)
789
790 lines := strings.Split(strings.TrimSpace(buf.String()), "\n")
791 if len(lines) != 2 {
792 t.Fatalf("wrote %d frames, want 2: %q", len(lines), buf.String())
793 }
794 var response frame
795 if err := json.Unmarshal([]byte(lines[0]), &response); err != nil {
796 t.Fatalf("response frame: %v", err)
797 }
798 requireResponseFrame(t, response, 1)
799 var notification frame
800 if err := json.Unmarshal([]byte(lines[1]), &notification); err != nil {
801 t.Fatalf("notification frame: %v", err)
802 }
803 if notification.Method != "test/notification" || notification.ID != nil {
804 t.Fatalf("second frame = %+v, want notification", notification)
805 }
806 }
807
808 func TestServeAdvertisesCommandsAfterEverySessionOpenResponse(t *testing.T) {
809 sessionDir := t.TempDir()
810 factory := &commandFactory{
811 dir: sessionDir,
812 commands: []command.Command{{Name: "review", Description: "Review the target"}},
813 }
814 client, stop := startOrderedServer(t, factory)
815 defer stop()
816
817 client.send(t, 1, "initialize", InitializeParams{ProtocolVersion: 1})
818 requireResponseFrame(t, client.next(t), 1)
819
820 client.send(t, 2, "session/new", SessionNewParams{Cwd: t.TempDir()})
821 newResponse := client.next(t)
822 requireResponseFrame(t, newResponse, 2)
823 requireAvailableCommandsFrame(t, client.next(t))
824 var created SessionNewResult
825 if err := json.Unmarshal(newResponse.Result, &created); err != nil || created.SessionID == "" {
826 t.Fatalf("session/new result: %v (%q)", err, created.SessionID)
827 }
828
829 client.send(t, 3, "session/close", SessionCloseParams{SessionID: created.SessionID})
830 requireResponseFrame(t, client.next(t), 3)
831
832 persistedID := "ordered-session-open"
833 path := transcriptPath(sessionDir, persistedID)
834 if err := agent.NewSession("").Save(path); err != nil {
835 t.Fatalf("save transcript: %v", err)
836 }
837 now := time.Now().UTC()
838 if err := saveACPMeta(path, acpSessionMeta{
839 SessionID: persistedID,
840 Cwd: sessionDir,
841 CreatedAt: now,
842 UpdatedAt: now,
843 }); err != nil {
844 t.Fatalf("save ACP metadata: %v", err)
845 }
846
847 client.send(t, 4, "session/load", SessionLoadParams{SessionID: persistedID, Cwd: sessionDir})
848 requireResponseFrame(t, client.next(t), 4)
849 requireAvailableCommandsFrame(t, client.next(t))
850
851 client.send(t, 5, "session/close", SessionCloseParams{SessionID: persistedID})
852 requireResponseFrame(t, client.next(t), 5)
853
854 client.send(t, 6, "session/resume", SessionResumeParams{SessionID: persistedID, Cwd: sessionDir})
855 requireResponseFrame(t, client.next(t), 6)
856 requireAvailableCommandsFrame(t, client.next(t))
857 }
858
859 func TestServeSessionConfigSwitchesModelAndEffort(t *testing.T) {
860 factory := &configurableFactory{}
861 client, stop := startServer(t, factory)
862 defer stop()
863
864 client.call(t, "initialize", InitializeParams{ProtocolVersion: 1})
865 newResp := client.call(t, "session/new", SessionNewParams{Cwd: t.TempDir()})
866 var nr SessionNewResult
867 if err := json.Unmarshal(newResp.Result, &nr); err != nil {
868 t.Fatalf("session/new result: %v", err)
869 }
870 if nr.Models == nil || nr.Models.CurrentModelID != "fast" {
871 t.Fatalf("models = %+v, want current fast", nr.Models)
872 }
873 modelOpt, ok := findConfigOption(nr.ConfigOptions, "model")
874 if !ok || modelOpt.CurrentValue != "fast" {
875 t.Fatalf("model config = %+v, want current fast", modelOpt)
876 }
877 if got := factory.buildAt(t, 0).Model; got != "fast" {
878 t.Fatalf("initial build model = %q, want fast", got)
879 }
880
881 setModelResp := client.call(t, "session/set_config_option", SetSessionConfigOptionParams{
882 SessionID: nr.SessionID,
883 ConfigID: "model",
884 Value: "pro",
885 })
886 var modelSet SetSessionConfigOptionResult
887 if err := json.Unmarshal(setModelResp.Result, &modelSet); err != nil {
888 t.Fatalf("set model result: %v", err)
889 }
890 modelOpt, _ = findConfigOption(modelSet.ConfigOptions, "model")
891 if modelOpt.CurrentValue != "pro" {
892 t.Fatalf("model after set_config_option = %q, want pro", modelOpt.CurrentValue)
893 }
894 if got := factory.buildAt(t, 1).Model; got != "pro" {
895 t.Fatalf("second build model = %q, want pro", got)
896 }
897
898 setEffortResp := client.call(t, "session/set_config_option", SetSessionConfigOptionParams{
899 SessionID: nr.SessionID,
900 ConfigID: "effort",
901 Value: "high",
902 })
903 var effortSet SetSessionConfigOptionResult
904 if err := json.Unmarshal(setEffortResp.Result, &effortSet); err != nil {
905 t.Fatalf("set effort result: %v", err)
906 }
907 effortOpt, _ := findConfigOption(effortSet.ConfigOptions, "effort")
908 if effortOpt.CurrentValue != "high" {
909 t.Fatalf("effort after set_config_option = %q, want high", effortOpt.CurrentValue)
910 }
911 effortBuild := factory.buildAt(t, 2)
912 if effortBuild.Model != "pro" || effortBuild.EffortOverride == nil || *effortBuild.EffortOverride != "high" {
913 t.Fatalf("effort build = model:%q effort:%v, want pro/high", effortBuild.Model, effortBuild.EffortOverride)
914 }
915
916 setLegacyResp := client.call(t, "session/set_model", SetSessionModelParams{SessionID: nr.SessionID, ModelID: "fast"})
917 if setLegacyResp.Error != nil {
918 t.Fatalf("session/set_model errored: %+v", setLegacyResp.Error)
919 }
920 if got := factory.buildAt(t, 3).Model; got != "fast" {
921 t.Fatalf("legacy set_model build model = %q, want fast", got)
922 }
923 }
924
925 func TestServeSessionAxesStayIndependent(t *testing.T) {
926 type observed struct {
927 profile string
928 approval string
929 plan bool
930 goal string
931 }
932 seen := make(chan observed, 2)
933 factory := &configurableFactory{
934 withCtrl: func(_ context.Context, sink event.Sink, input string, p SessionParams, ctrl *control.Controller) error {
935 seen <- observed{
936 profile: p.RuntimeProfile,
937 approval: ctrl.ToolApprovalMode(),
938 plan: ctrl.PlanMode(),
939 goal: ctrl.Goal(),
940 }
941 ctrl.ClearGoal()
942 sink.Emit(event.Event{Kind: event.Text, Text: "done"})
943 return nil
944 },
945 }
946 client, stop := startServer(t, factory)
947 defer stop()
948
949 client.call(t, "initialize", InitializeParams{ProtocolVersion: 1})
950 newResp := client.call(t, "session/new", SessionNewParams{Cwd: t.TempDir()})
951 var nr SessionNewResult
952 if err := json.Unmarshal(newResp.Result, &nr); err != nil {
953 t.Fatalf("session/new result: %v", err)
954 }
955 work, ok := findConfigOption(nr.ConfigOptions, "work_mode")
956 if !ok || work.CurrentValue != "balanced" {
957 t.Fatalf("initial work mode = %+v, want balanced", work)
958 }
959 approval, ok := findConfigOption(nr.ConfigOptions, "tool_approval")
960 if !ok || approval.CurrentValue != control.ToolApprovalAsk {
961 t.Fatalf("initial tool approval = %+v, want ask", approval)
962 }
963
964 setWork := client.call(t, "session/set_config_option", SetSessionConfigOptionParams{
965 SessionID: nr.SessionID,
966 ConfigID: "work_mode",
967 Value: "delivery",
968 })
969 if setWork.Error != nil {
970 t.Fatalf("set work mode: %+v", setWork.Error)
971 }
972 if got := factory.buildAt(t, 1).RuntimeProfile; got != "delivery" {
973 t.Fatalf("rebuilt runtime profile = %q, want delivery", got)
974 }
975 buildsAfterWorkMode := factory.buildCount()
976
977 setApproval := client.call(t, "session/set_config_option", SetSessionConfigOptionParams{
978 SessionID: nr.SessionID,
979 ConfigID: "tool_approval",
980 Value: control.ToolApprovalAuto,
981 })
982 if setApproval.Error != nil {
983 t.Fatalf("set tool approval: %+v", setApproval.Error)
984 }
985 if got := factory.buildCount(); got != buildsAfterWorkMode {
986 t.Fatalf("tool approval rebuilt controller: builds=%d, want %d", got, buildsAfterWorkMode)
987 }
988
989 setGoal := client.call(t, "session/set_mode", SessionSetModeParams{SessionID: nr.SessionID, ModeID: sessionModeGoal})
990 if setGoal.Error != nil {
991 t.Fatalf("set goal mode: %+v", setGoal.Error)
992 }
993 promptCh := client.callAsync("session/prompt", SessionPromptParams{
994 SessionID: nr.SessionID,
995 Prompt: []ContentBlock{{Type: "text", Text: "ship the ACP profile switch"}},
996 })
997 _, promptResp := drainPrompt(t, client, promptCh)
998 if promptResp.Error != nil {
999 t.Fatalf("goal prompt: %+v", promptResp.Error)
1000 }
1001 goalObserved := <-seen
1002 if goalObserved.profile != "delivery" || goalObserved.approval != control.ToolApprovalAuto || goalObserved.plan || goalObserved.goal != "ship the ACP profile switch" {
1003 t.Fatalf("goal axes = %+v, want delivery + auto + goal", goalObserved)
1004 }
1005
1006 setPlan := client.call(t, "session/set_mode", SessionSetModeParams{SessionID: nr.SessionID, ModeID: sessionModePlan})
1007 if setPlan.Error != nil {
1008 t.Fatalf("set plan mode: %+v", setPlan.Error)
1009 }
1010 promptCh = client.callAsync("session/prompt", SessionPromptParams{
1011 SessionID: nr.SessionID,
1012 Prompt: []ContentBlock{{Type: "text", Text: "plan the follow-up"}},
1013 })
1014 _, promptResp = drainPrompt(t, client, promptCh)
1015 if promptResp.Error != nil {
1016 t.Fatalf("plan prompt: %+v", promptResp.Error)
1017 }
1018 planObserved := <-seen
1019 if planObserved.profile != "delivery" || planObserved.approval != control.ToolApprovalAuto || !planObserved.plan || planObserved.goal != "" {
1020 t.Fatalf("plan axes = %+v, want delivery + auto + plan", planObserved)
1021 }
1022 }
1023
1024 func TestServeLegacyModeAliasesRemainCompatible(t *testing.T) {
1025 type observed struct {
1026 approval string
1027 plan bool
1028 }
1029 seen := make(chan observed, 2)
1030 factory := &configurableFactory{
1031 withCtrl: func(_ context.Context, sink event.Sink, _ string, _ SessionParams, ctrl *control.Controller) error {
1032 seen <- observed{approval: ctrl.ToolApprovalMode(), plan: ctrl.PlanMode()}
1033 sink.Emit(event.Event{Kind: event.Text, Text: "done"})
1034 return nil
1035 },
1036 }
1037 client, stop := startServer(t, factory)
1038 defer stop()
1039 client.call(t, "initialize", InitializeParams{ProtocolVersion: 1})
1040 newResp := client.call(t, "session/new", SessionNewParams{Cwd: t.TempDir()})
1041 var nr SessionNewResult
1042 if err := json.Unmarshal(newResp.Result, &nr); err != nil {
1043 t.Fatalf("session/new result: %v", err)
1044 }
1045
1046 for _, tc := range []struct {
1047 mode string
1048 want string
1049 }{
1050 {mode: sessionModeLegacyDefault, want: control.ToolApprovalAsk},
1051 {mode: sessionModeLegacyAuto, want: control.ToolApprovalYolo},
1052 } {
1053 if resp := client.call(t, "session/set_mode", SessionSetModeParams{SessionID: nr.SessionID, ModeID: tc.mode}); resp.Error != nil {
1054 t.Fatalf("set legacy mode %q: %+v", tc.mode, resp.Error)
1055 }
1056 promptCh := client.callAsync("session/prompt", SessionPromptParams{
1057 SessionID: nr.SessionID,
1058 Prompt: []ContentBlock{{Type: "text", Text: "check legacy mode"}},
1059 })
1060 _, promptResp := drainPrompt(t, client, promptCh)
1061 if promptResp.Error != nil {
1062 t.Fatalf("prompt after legacy mode %q: %+v", tc.mode, promptResp.Error)
1063 }
1064 if got := <-seen; got.approval != tc.want || got.plan {
1065 t.Fatalf("legacy mode %q = %+v, want approval %q without plan", tc.mode, got, tc.want)
1066 }
1067 }
1068 }
1069
1070 func TestServeSessionAxesRestoreFromMetadata(t *testing.T) {
1071 dir := t.TempDir()
1072 sessionID := "axes-restore"
1073 path := transcriptPath(dir, sessionID)
1074 saved := agent.NewSession("")
1075 saved.Add(provider.Message{Role: provider.RoleUser, Content: "persist these axes"})
1076 if err := saved.Save(path); err != nil {
1077 t.Fatalf("save transcript: %v", err)
1078 }
1079 if err := saveACPMeta(path, acpSessionMeta{
1080 SessionID: sessionID,
1081 Cwd: dir,
1082 Model: "fast",
1083 RuntimeProfile: "delivery",
1084 ToolApprovalMode: control.ToolApprovalAuto,
1085 CollaborationMode: sessionModePlan,
1086 CreatedAt: time.Now().UTC(),
1087 UpdatedAt: time.Now().UTC(),
1088 }); err != nil {
1089 t.Fatalf("save ACP metadata: %v", err)
1090 }
1091
1092 seen := make(chan struct {
1093 approval string
1094 plan bool
1095 }, 1)
1096 reloadedFactory := &configurableFactory{
1097 dir: dir,
1098 withCtrl: func(_ context.Context, sink event.Sink, _ string, _ SessionParams, ctrl *control.Controller) error {
1099 seen <- struct {
1100 approval string
1101 plan bool
1102 }{approval: ctrl.ToolApprovalMode(), plan: ctrl.PlanMode()}
1103 sink.Emit(event.Event{Kind: event.Text, Text: "done"})
1104 return nil
1105 },
1106 }
1107 reloadedClient, stopReloaded := startServer(t, reloadedFactory)
1108 defer stopReloaded()
1109 reloadedClient.call(t, "initialize", InitializeParams{ProtocolVersion: 1})
1110 loadResp := reloadedClient.call(t, "session/load", SessionLoadParams{SessionID: sessionID, Cwd: dir})
1111 if loadResp.Error != nil {
1112 t.Fatalf("session/load: %+v", loadResp.Error)
1113 }
1114 var lr SessionLoadResult
1115 if err := json.Unmarshal(loadResp.Result, &lr); err != nil {
1116 t.Fatalf("session/load result: %v", err)
1117 }
1118 work, _ := findConfigOption(lr.ConfigOptions, "work_mode")
1119 approval, _ := findConfigOption(lr.ConfigOptions, "tool_approval")
1120 if work.CurrentValue != "delivery" || approval.CurrentValue != control.ToolApprovalAuto || lr.Modes == nil || lr.Modes.CurrentModeID != sessionModePlan {
1121 t.Fatalf("reloaded axes = work:%+v approval:%+v modes:%+v", work, approval, lr.Modes)
1122 }
1123 if got := reloadedFactory.buildAt(t, 0).RuntimeProfile; got != "delivery" {
1124 t.Fatalf("reloaded build profile = %q, want delivery", got)
1125 }
1126 promptCh := reloadedClient.callAsync("session/prompt", SessionPromptParams{
1127 SessionID: sessionID,
1128 Prompt: []ContentBlock{{Type: "text", Text: "verify restored controller"}},
1129 })
1130 _, promptResp := drainPrompt(t, reloadedClient, promptCh)
1131 if promptResp.Error != nil {
1132 t.Fatalf("reloaded prompt: %+v", promptResp.Error)
1133 }
1134 observed := <-seen
1135 if observed.approval != control.ToolApprovalAuto || !observed.plan {
1136 t.Fatalf("restored controller axes = %+v, want auto + plan", observed)
1137 }
1138 }
1139
1140 func TestServeSessionConfigQueuesDuringActivePrompt(t *testing.T) {
1141 started := make(chan struct{})
1142 release := make(chan struct{})
1143 var once sync.Once
1144 factory := &configurableFactory{
1145 behavior: func(ctx context.Context, sink event.Sink, input string, p SessionParams) error {
1146 once.Do(func() { close(started) })
1147 select {
1148 case <-release:
1149 case <-ctx.Done():
1150 return ctx.Err()
1151 }
1152 sink.Emit(event.Event{Kind: event.Text, Text: p.Model + ":" + input})
1153 return nil
1154 },
1155 }
1156 client, stop := startServer(t, factory)
1157 defer stop()
1158
1159 client.call(t, "initialize", InitializeParams{ProtocolVersion: 1})
1160 newResp := client.call(t, "session/new", SessionNewParams{Cwd: t.TempDir()})
1161 var nr SessionNewResult
1162 if err := json.Unmarshal(newResp.Result, &nr); err != nil {
1163 t.Fatalf("session/new result: %v", err)
1164 }
1165
1166 first := client.callAsync("session/prompt", SessionPromptParams{
1167 SessionID: nr.SessionID,
1168 Prompt: []ContentBlock{{Type: "text", Text: "first"}},
1169 })
1170 select {
1171 case <-started:
1172 case <-time.After(2 * time.Second):
1173 t.Fatal("prompt never started")
1174 }
1175
1176 setResp := client.call(t, "session/set_config_option", SetSessionConfigOptionParams{
1177 SessionID: nr.SessionID,
1178 ConfigID: "model",
1179 Value: "pro",
1180 })
1181 if setResp.Error != nil {
1182 t.Fatalf("set_config_option while running errored: %+v", setResp.Error)
1183 }
1184 var set SetSessionConfigOptionResult
1185 if err := json.Unmarshal(setResp.Result, &set); err != nil {
1186 t.Fatalf("set model result: %v", err)
1187 }
1188 modelOpt, _ := findConfigOption(set.ConfigOptions, "model")
1189 if modelOpt.CurrentValue != "pro" {
1190 t.Fatalf("queued model option = %q, want pro", modelOpt.CurrentValue)
1191 }
1192 if got := factory.buildCount(); got != 1 {
1193 t.Fatalf("build count while prompt is active = %d, want only initial build", got)
1194 }
1195
1196 close(release)
1197 _, resp := drainPrompt(t, client, first)
1198 if resp.Error != nil {
1199 t.Fatalf("first prompt errored: %+v", resp.Error)
1200 }
1201 if got := factory.buildAt(t, 1).Model; got != "pro" {
1202 t.Fatalf("queued rebuild model = %q, want pro", got)
1203 }
1204 }
1205
1206 func TestServeSessionConfigRejectsBackgroundJobsWhileIdle(t *testing.T) {
1207 dir := t.TempDir()
1208 factory := &configurableFactory{dir: dir, managers: []*jobs.Manager{}}
1209 client, stop := startServer(t, factory)
1210 defer stop()
1211
1212 client.call(t, "initialize", InitializeParams{ProtocolVersion: 1})
1213 newResp := client.call(t, "session/new", SessionNewParams{Cwd: t.TempDir()})
1214 var nr SessionNewResult
1215 if err := json.Unmarshal(newResp.Result, &nr); err != nil {
1216 t.Fatalf("session/new result: %v", err)
1217 }
1218
1219 jm := factory.managerAt(t, 0)
1220 release := make(chan struct{})
1221 var releaseOnce sync.Once
1222 started := make(chan struct{})
1223 sessionPath := transcriptPath(dir, nr.SessionID)
1224 jm.StartForSession(agent.BranchID(sessionPath), "bash", "server", func(ctx context.Context, _ io.Writer) (string, error) {
1225 close(started)
1226 select {
1227 case <-release:
1228 return "", nil
1229 case <-ctx.Done():
1230 return "", ctx.Err()
1231 }
1232 })
1233 defer func() {
1234 releaseOnce.Do(func() { close(release) })
1235 jm.Close()
1236 }()
1237 select {
1238 case <-started:
1239 case <-time.After(2 * time.Second):
1240 t.Fatal("background job never started")
1241 }
1242
1243 setResp := client.call(t, "session/set_config_option", SetSessionConfigOptionParams{
1244 SessionID: nr.SessionID,
1245 ConfigID: "model",
1246 Value: "pro",
1247 })
1248 if setResp.Error == nil || !strings.Contains(setResp.Error.Message, "stop background jobs") {
1249 t.Fatalf("set_config_option with background job error = %+v, want stop background jobs RPC error", setResp.Error)
1250 }
1251 legacyResp := client.call(t, "session/set_model", SetSessionModelParams{SessionID: nr.SessionID, ModelID: "pro"})
1252 if legacyResp.Error == nil || !strings.Contains(legacyResp.Error.Message, "stop background jobs") {
1253 t.Fatalf("set_model with background job error = %+v, want stop background jobs RPC error", legacyResp.Error)
1254 }
1255 if got := factory.buildCount(); got != 1 {
1256 t.Fatalf("build count after rejected switch = %d, want 1", got)
1257 }
1258 if running := jm.RunningForSession(agent.BranchID(sessionPath)); len(running) != 1 {
1259 t.Fatalf("running jobs after rejected switch = %+v, want original job still running", running)
1260 }
1261
1262 releaseOnce.Do(func() { close(release) })
1263 _ = jm.WaitForSession(context.Background(), agent.BranchID(sessionPath), nil, 5)
1264 if running := jm.RunningForSession(agent.BranchID(sessionPath)); len(running) != 0 {
1265 t.Fatalf("running jobs after release = %+v, want none before retry", running)
1266 }
1267
1268 retryResp := client.call(t, "session/set_config_option", SetSessionConfigOptionParams{
1269 SessionID: nr.SessionID,
1270 ConfigID: "model",
1271 Value: "pro",
1272 })
1273 if retryResp.Error != nil {
1274 t.Fatalf("retry set_config_option after jobs stopped errored: %+v", retryResp.Error)
1275 }
1276 var retry SetSessionConfigOptionResult
1277 if err := json.Unmarshal(retryResp.Result, &retry); err != nil {
1278 t.Fatalf("retry set_config_option result: %v", err)
1279 }
1280 modelOpt, _ := findConfigOption(retry.ConfigOptions, "model")
1281 if modelOpt.CurrentValue != "pro" {
1282 t.Fatalf("retry model currentValue = %q, want pro", modelOpt.CurrentValue)
1283 }
1284 if got := factory.buildCount(); got != 2 {
1285 t.Fatalf("build count after retry switch = %d, want rebuild", got)
1286 }
1287
1288 promptCh := client.callAsync("session/prompt", SessionPromptParams{
1289 SessionID: nr.SessionID,
1290 Prompt: []ContentBlock{{Type: "text", Text: "after-switch"}},
1291 })
1292 notifs, resp := drainPrompt(t, client, promptCh)
1293 if resp.Error != nil {
1294 t.Fatalf("prompt after retry switch errored: %+v", resp.Error)
1295 }
1296 var usedNewModel bool
1297 for _, n := range notifs {
1298 if text, ok := messageChunkText(t, n); ok && strings.Contains(text, "pro:after-switch") {
1299 usedNewModel = true
1300 break
1301 }
1302 }
1303 if !usedNewModel {
1304 t.Fatalf("prompt after retry did not use new model; notifications=%+v", notifs)
1305 }
1306 }
1307
1308 func TestQueuedRebuildPreservesControllerSideAxisDrift(t *testing.T) {
1309 started := make(chan struct{})
1310 release := make(chan struct{})
1311 seen := make(chan struct {
1312 profile string
1313 approval string
1314 plan bool
1315 }, 1)
1316 var first sync.Once
1317 factory := &configurableFactory{
1318 withCtrl: func(ctx context.Context, sink event.Sink, _ string, p SessionParams, ctrl *control.Controller) error {
1319 isFirst := false
1320 first.Do(func() { isFirst = true })
1321 if isFirst {
1322 close(started)
1323 select {
1324 case <-release:
1325 case <-ctx.Done():
1326 return ctx.Err()
1327 }
1328 // Simulate slash-command/controller-side state changes late in the
1329 // turn, after the client has queued a work-mode rebuild.
1330 ctrl.SetPlanMode(false)
1331 ctrl.SetToolApprovalMode(control.ToolApprovalAuto)
1332 } else {
1333 seen <- struct {
1334 profile string
1335 approval string
1336 plan bool
1337 }{p.RuntimeProfile, ctrl.ToolApprovalMode(), ctrl.PlanMode()}
1338 }
1339 sink.Emit(event.Event{Kind: event.Text, Text: "done"})
1340 return nil
1341 },
1342 }
1343 client, stop := startServer(t, factory)
1344 defer stop()
1345 client.call(t, "initialize", InitializeParams{ProtocolVersion: 1})
1346 newResp := client.call(t, "session/new", SessionNewParams{Cwd: t.TempDir()})
1347 var nr SessionNewResult
1348 if err := json.Unmarshal(newResp.Result, &nr); err != nil {
1349 t.Fatalf("session/new result: %v", err)
1350 }
1351 if resp := client.call(t, "session/set_mode", SessionSetModeParams{SessionID: nr.SessionID, ModeID: sessionModePlan}); resp.Error != nil {
1352 t.Fatalf("set plan mode: %+v", resp.Error)
1353 }
1354 firstPrompt := client.callAsync("session/prompt", SessionPromptParams{
1355 SessionID: nr.SessionID,
1356 Prompt: []ContentBlock{{Type: "text", Text: "first"}},
1357 })
1358 select {
1359 case <-started:
1360 case <-time.After(2 * time.Second):
1361 t.Fatal("first prompt did not start")
1362 }
1363 if resp := client.call(t, "session/set_config_option", SetSessionConfigOptionParams{
1364 SessionID: nr.SessionID,
1365 ConfigID: "work_mode",
1366 Value: "delivery",
1367 }); resp.Error != nil {
1368 t.Fatalf("queue work mode: %+v", resp.Error)
1369 }
1370 close(release)
1371 if _, resp := drainPrompt(t, client, firstPrompt); resp.Error != nil {
1372 t.Fatalf("first prompt: %+v", resp.Error)
1373 }
1374
1375 secondPrompt := client.callAsync("session/prompt", SessionPromptParams{
1376 SessionID: nr.SessionID,
1377 Prompt: []ContentBlock{{Type: "text", Text: "second"}},
1378 })
1379 if _, resp := drainPrompt(t, client, secondPrompt); resp.Error != nil {
1380 t.Fatalf("second prompt: %+v", resp.Error)
1381 }
1382 got := <-seen
1383 if got.profile != "delivery" || got.approval != control.ToolApprovalAuto || got.plan {
1384 t.Fatalf("rebuilt axes = %+v, want delivery + auto + normal", got)
1385 }
1386 }
1387
1388 func TestServeQueuedSessionConfigDiscardedWhenPromptLeavesBackgroundJob(t *testing.T) {
1389 dir := t.TempDir()
1390 releaseJob := make(chan struct{})
1391 releaseTurn := make(chan struct{})
1392 startedJob := make(chan struct{})
1393 startedTurn := make(chan struct{})
1394 var jobOnce sync.Once
1395 factory := &configurableFactory{dir: dir, managers: []*jobs.Manager{}}
1396 factory.behavior = func(ctx context.Context, sink event.Sink, input string, p SessionParams) error {
1397 if input == "first" {
1398 close(startedTurn)
1399 jm := factory.managerAt(t, 0)
1400 jobOnce.Do(func() {
1401 jm.StartForSession(jobs.SessionFromContext(ctx), "bash", "server", func(ctx context.Context, _ io.Writer) (string, error) {
1402 close(startedJob)
1403 select {
1404 case <-releaseJob:
1405 return "", nil
1406 case <-ctx.Done():
1407 return "", ctx.Err()
1408 }
1409 })
1410 })
1411 select {
1412 case <-startedJob:
1413 case <-time.After(2 * time.Second):
1414 t.Fatal("background job never started")
1415 }
1416 select {
1417 case <-releaseTurn:
1418 case <-ctx.Done():
1419 return ctx.Err()
1420 }
1421 }
1422 sink.Emit(event.Event{Kind: event.Text, Text: p.Model + ":" + input})
1423 return nil
1424 }
1425 client, stop := startServer(t, factory)
1426 defer stop()
1427
1428 client.call(t, "initialize", InitializeParams{ProtocolVersion: 1})
1429 newResp := client.call(t, "session/new", SessionNewParams{Cwd: t.TempDir()})
1430 var nr SessionNewResult
1431 if err := json.Unmarshal(newResp.Result, &nr); err != nil {
1432 t.Fatalf("session/new result: %v", err)
1433 }
1434
1435 first := client.callAsync("session/prompt", SessionPromptParams{
1436 SessionID: nr.SessionID,
1437 Prompt: []ContentBlock{{Type: "text", Text: "first"}},
1438 })
1439 select {
1440 case <-startedTurn:
1441 case <-time.After(2 * time.Second):
1442 t.Fatal("prompt never started")
1443 }
1444 setResp := client.call(t, "session/set_config_option", SetSessionConfigOptionParams{
1445 SessionID: nr.SessionID,
1446 ConfigID: "model",
1447 Value: "pro",
1448 })
1449 if setResp.Error != nil {
1450 t.Fatalf("set_config_option while prompt is running errored: %+v", setResp.Error)
1451 }
1452
1453 close(releaseTurn)
1454 notifs, resp := drainPrompt(t, client, first)
1455 if resp.Error != nil {
1456 t.Fatalf("first prompt errored: %+v", resp.Error)
1457 }
1458 warningIndex := -1
1459 for i, n := range notifs {
1460 if text, ok := messageChunkText(t, n); ok && strings.Contains(text, "stop background jobs") {
1461 warningIndex = i
1462 break
1463 }
1464 }
1465 if warningIndex < 0 {
1466 t.Fatalf("queued switch updates = %d, want warning mentioning background jobs", len(notifs))
1467 }
1468 var sawOldConfig bool
1469 for _, n := range notifs[warningIndex+1:] {
1470 if value, ok := configOptionValueFromUpdate(t, n, "model"); ok && value == "fast" {
1471 sawOldConfig = true
1472 break
1473 }
1474 }
1475 if !sawOldConfig {
1476 t.Fatalf("queued switch notifications after warning did not include model currentValue=fast: %+v", notifs[warningIndex+1:])
1477 }
1478
1479 close(releaseJob)
1480 jm := factory.managerAt(t, 0)
1481 _ = jm.WaitForSession(context.Background(), agent.BranchID(transcriptPath(dir, nr.SessionID)), nil, 5)
1482 second := client.callAsync("session/prompt", SessionPromptParams{
1483 SessionID: nr.SessionID,
1484 Prompt: []ContentBlock{{Type: "text", Text: "second"}},
1485 })
1486 _, resp = drainPrompt(t, client, second)
1487 if resp.Error != nil {
1488 t.Fatalf("second prompt errored: %+v", resp.Error)
1489 }
1490 if got := factory.buildCount(); got != 1 {
1491 t.Fatalf("build count after discarded queued switch = %d, want 1", got)
1492 }
1493 }
1494
1495 func TestServeSessionConfigRejectsPendingAsk(t *testing.T) {
1496 factory := &configurableFactory{
1497 withCtrl: func(ctx context.Context, _ event.Sink, _ string, _ SessionParams, ctrl *control.Controller) error {
1498 _, err := ctrl.Ask(ctx, []event.AskQuestion{{
1499 ID: "choice",
1500 Prompt: "Pick one",
1501 Options: []event.AskOption{{Label: "A"}, {Label: "B"}},
1502 }})
1503 return err
1504 },
1505 }
1506 client, stop := startServer(t, factory)
1507 defer stop()
1508
1509 client.call(t, "initialize", InitializeParams{ProtocolVersion: 1})
1510 newResp := client.call(t, "session/new", SessionNewParams{Cwd: t.TempDir()})
1511 var nr SessionNewResult
1512 if err := json.Unmarshal(newResp.Result, &nr); err != nil {
1513 t.Fatalf("session/new result: %v", err)
1514 }
1515
1516 promptCh := client.callAsync("session/prompt", SessionPromptParams{
1517 SessionID: nr.SessionID,
1518 Prompt: []ContentBlock{{Type: "text", Text: "ask"}},
1519 })
1520 var req frame
1521 select {
1522 case req = <-client.reqs:
1523 case <-time.After(2 * time.Second):
1524 t.Fatal("ask request was not sent to client")
1525 }
1526
1527 setResp := client.call(t, "session/set_config_option", SetSessionConfigOptionParams{
1528 SessionID: nr.SessionID,
1529 ConfigID: "model",
1530 Value: "pro",
1531 })
1532 if setResp.Error == nil || !strings.Contains(setResp.Error.Message, "pending") {
1533 t.Fatalf("set_config_option with pending ask error = %+v, want pending interaction RPC error", setResp.Error)
1534 }
1535 if got := factory.buildCount(); got != 1 {
1536 t.Fatalf("build count while ask is pending = %d, want 1", got)
1537 }
1538
1539 client.reply(req.ID, PermissionRequestResult{
1540 Outcome: PermissionOutcome{Outcome: "selected", OptionID: "choice:1"},
1541 })
1542 _, resp := drainPrompt(t, client, promptCh)
1543 if resp.Error != nil {
1544 t.Fatalf("prompt errored: %+v", resp.Error)
1545 }
1546 if got := factory.buildCount(); got != 1 {
1547 t.Fatalf("build count after answered ask = %d, want no queued rebuild", got)
1548 }
1549 }
1550
1551 func TestServeSessionConfigRebuildPreservesLifecycleHooks(t *testing.T) {
1552 factory := &configurableFactory{withHooks: true}
1553 client, stop := startServer(t, factory)
1554 defer stop()
1555
1556 client.call(t, "initialize", InitializeParams{ProtocolVersion: 1})
1557 newResp := client.call(t, "session/new", SessionNewParams{Cwd: t.TempDir()})
1558 var nr SessionNewResult
1559 if err := json.Unmarshal(newResp.Result, &nr); err != nil {
1560 t.Fatalf("session/new result: %v", err)
1561 }
1562
1563 promptCh := client.callAsync("session/prompt", SessionPromptParams{
1564 SessionID: nr.SessionID,
1565 Prompt: []ContentBlock{{Type: "text", Text: "one"}},
1566 })
1567 _, resp := drainPrompt(t, client, promptCh)
1568 if resp.Error != nil {
1569 t.Fatalf("first prompt errored: %+v", resp.Error)
1570 }
1571 if got := factory.hookEventsSnapshot(); len(got) != 1 || got[0] != hook.SessionStart {
1572 t.Fatalf("hook events after first prompt = %v, want [SessionStart]", got)
1573 }
1574
1575 setResp := client.call(t, "session/set_config_option", SetSessionConfigOptionParams{
1576 SessionID: nr.SessionID,
1577 ConfigID: "model",
1578 Value: "pro",
1579 })
1580 if setResp.Error != nil {
1581 t.Fatalf("set_config_option errored: %+v", setResp.Error)
1582 }
1583 if got := factory.hookEventsSnapshot(); len(got) != 1 || got[0] != hook.SessionStart {
1584 t.Fatalf("hook events after config rebuild = %v, want no lifecycle hook", got)
1585 }
1586
1587 promptCh = client.callAsync("session/prompt", SessionPromptParams{
1588 SessionID: nr.SessionID,
1589 Prompt: []ContentBlock{{Type: "text", Text: "two"}},
1590 })
1591 _, resp = drainPrompt(t, client, promptCh)
1592 if resp.Error != nil {
1593 t.Fatalf("second prompt errored: %+v", resp.Error)
1594 }
1595 if got := factory.hookEventsSnapshot(); len(got) != 1 || got[0] != hook.SessionStart {
1596 t.Fatalf("hook events after second prompt = %v, want no duplicate SessionStart", got)
1597 }
1598
1599 closeResp := client.call(t, "session/close", SessionCloseParams{SessionID: nr.SessionID})
1600 if closeResp.Error != nil {
1601 t.Fatalf("session/close errored: %+v", closeResp.Error)
1602 }
1603 if got := factory.hookEventsSnapshot(); len(got) != 2 || got[0] != hook.SessionStart || got[1] != hook.SessionEnd {
1604 t.Fatalf("hook events after close = %v, want [SessionStart SessionEnd]", got)
1605 }
1606 }
1607
1608 func TestServeSessionLoadFallsBackFromStaleSavedModel(t *testing.T) {
1609 dir := t.TempDir()
1610 cwd := t.TempDir()
1611 sessionID := "stale-model"
1612 path := transcriptPath(dir, sessionID)
1613 saved := agent.NewSession("")
1614 saved.Add(provider.Message{Role: provider.RoleUser, Content: "hello"})
1615 if err := saved.Save(path); err != nil {
1616 t.Fatal(err)
1617 }
1618 effort := "high"
1619 if err := saveACPMeta(path, acpSessionMeta{
1620 SessionID: sessionID,
1621 Cwd: cwd,
1622 Model: "missing/model",
1623 EffortOverride: &effort,
1624 CreatedAt: time.Now().UTC(),
1625 UpdatedAt: time.Now().UTC(),
1626 }); err != nil {
1627 t.Fatal(err)
1628 }
1629
1630 factory := &configurableFactory{dir: dir}
1631 client, stop := startServer(t, factory)
1632 defer stop()
1633
1634 client.call(t, "initialize", InitializeParams{ProtocolVersion: 1})
1635 loadResp := client.call(t, "session/load", SessionLoadParams{SessionID: sessionID, Cwd: cwd})
1636 if loadResp.Error != nil {
1637 t.Fatalf("session/load with stale saved model errored: %+v", loadResp.Error)
1638 }
1639 if got := factory.buildAt(t, 0).Model; got != "fast" {
1640 t.Fatalf("fallback build model = %q, want fast", got)
1641 }
1642 if got := factory.buildAt(t, 0).RuntimeProfile; got != "balanced" {
1643 t.Fatalf("old metadata runtime profile = %q, want balanced", got)
1644 }
1645 var loaded SessionLoadResult
1646 if err := json.Unmarshal(loadResp.Result, &loaded); err != nil {
1647 t.Fatalf("session/load result: %v", err)
1648 }
1649 approval, _ := findConfigOption(loaded.ConfigOptions, "tool_approval")
1650 if approval.CurrentValue != control.ToolApprovalAsk || loaded.Modes == nil || loaded.Modes.CurrentModeID != sessionModeNormal {
1651 t.Fatalf("old metadata axes = approval:%+v modes:%+v, want ask + normal", approval, loaded.Modes)
1652 }
1653 meta, ok, err := loadACPMeta(path)
1654 if err != nil || !ok {
1655 t.Fatalf("load rewritten meta = %v, ok=%v", err, ok)
1656 }
1657 if meta.Model != "fast" {
1658 t.Fatalf("rewritten meta model = %q, want fast", meta.Model)
1659 }
1660 }
1661
1662 func TestServeSessionLoadRejectsCleanupPending(t *testing.T) {
1663 dir := t.TempDir()
1664 cwd := t.TempDir()
1665 sessionID := "pending-load"
1666 path := transcriptPath(dir, sessionID)
1667 saved := agent.NewSession("")
1668 saved.Add(provider.Message{Role: provider.RoleUser, Content: "hello"})
1669 if err := saved.Save(path); err != nil {
1670 t.Fatal(err)
1671 }
1672 if err := saveACPMeta(path, acpSessionMeta{
1673 SessionID: sessionID,
1674 Cwd: cwd,
1675 CreatedAt: time.Now().UTC(),
1676 UpdatedAt: time.Now().UTC(),
1677 }); err != nil {
1678 t.Fatal(err)
1679 }
1680 if err := agent.MarkCleanupPending(path, "delete"); err != nil {
1681 t.Fatal(err)
1682 }
1683
1684 factory := &configurableFactory{dir: dir}
1685 client, stop := startServer(t, factory)
1686 defer stop()
1687
1688 client.call(t, "initialize", InitializeParams{ProtocolVersion: 1})
1689 loadResp := client.call(t, "session/load", SessionLoadParams{SessionID: sessionID, Cwd: cwd})
1690 if loadResp.Error == nil || !strings.Contains(loadResp.Error.Message, "unknown session") {
1691 t.Fatalf("session/load cleanup-pending error = %+v, want unknown session", loadResp.Error)
1692 }
1693 factory.mu.Lock()
1694 builds := append([]SessionParams(nil), factory.builds...)
1695 factory.mu.Unlock()
1696 if len(builds) != 0 {
1697 t.Fatalf("cleanup-pending load should not build a controller, got builds %+v", builds)
1698 }
1699 }
1700
1701 func TestServeCancel(t *testing.T) {
1702 started := make(chan struct{})
1703 factory := &fakeFactory{behavior: func(ctx context.Context, _ event.Sink, _ string) error {
1704 close(started)
1705 <-ctx.Done()
1706 return ctx.Err()
1707 }}
1708 client, stop := startServer(t, factory)
1709 defer stop()
1710
1711 client.call(t, "initialize", InitializeParams{ProtocolVersion: 1})
1712 newResp := client.call(t, "session/new", SessionNewParams{})
1713 var nr SessionNewResult
1714 json.Unmarshal(newResp.Result, &nr)
1715
1716 promptCh := client.callAsync("session/prompt", SessionPromptParams{
1717 SessionID: nr.SessionID,
1718 Prompt: []ContentBlock{{Type: "text", Text: "loop"}},
1719 })
1720
1721 select {
1722 case <-started:
1723 case <-time.After(2 * time.Second):
1724 t.Fatal("prompt never started")
1725 }
1726 client.notify("session/cancel", SessionCancelParams{SessionID: nr.SessionID})
1727
1728 select {
1729 case resp := <-promptCh:
1730 var pr SessionPromptResult
1731 json.Unmarshal(resp.Result, &pr)
1732 if pr.StopReason != StopCancelled {
1733 t.Errorf("stopReason = %q, want cancelled", pr.StopReason)
1734 }
1735 case <-time.After(2 * time.Second):
1736 t.Fatal("cancel did not end the prompt")
1737 }
1738 }
1739
1740 func TestServeSteerInjectsIntoActivePrompt(t *testing.T) {
1741 barrier := &steerBarrierTool{started: make(chan struct{}), release: make(chan struct{})}
1742 prov := testutil.NewMock("steer",
1743 testutil.Turn{ToolCalls: []provider.ToolCall{{ID: "call-1", Name: barrier.Name(), Arguments: `{}`}}},
1744 testutil.Turn{Text: "done"},
1745 )
1746 client, stop := startServer(t, &steerFactory{provider: prov, barrier: barrier})
1747 defer stop()
1748
1749 client.call(t, "initialize", InitializeParams{ProtocolVersion: 1})
1750 newResp := client.call(t, "session/new", SessionNewParams{})
1751 var nr SessionNewResult
1752 if err := json.Unmarshal(newResp.Result, &nr); err != nil {
1753 t.Fatalf("session/new: %v", err)
1754 }
1755
1756 promptCh := client.callAsync("session/prompt", SessionPromptParams{
1757 SessionID: nr.SessionID,
1758 Prompt: []ContentBlock{{Type: "text", Text: "start"}},
1759 })
1760 select {
1761 case <-barrier.started:
1762 case <-time.After(2 * time.Second):
1763 t.Fatal("prompt never reached the tool boundary")
1764 }
1765
1766 legacyResp := client.call(t, "session/steer", SessionSteerParams{
1767 SessionID: nr.SessionID,
1768 Prompt: []ContentBlock{{Type: "text", Text: "legacy route"}},
1769 })
1770 if legacyResp.Error == nil || legacyResp.Error.Code != ErrMethodNotFound {
1771 t.Fatalf("legacy session/steer = %+v, want method not found", legacyResp.Error)
1772 }
1773
1774 steerResp := client.call(t, sessionSteerMethod, SessionSteerParams{
1775 SessionID: nr.SessionID,
1776 Prompt: []ContentBlock{{Type: "text", Text: "use plan B"}},
1777 })
1778 if steerResp.Error != nil {
1779 t.Fatalf("%s errored: %+v", sessionSteerMethod, steerResp.Error)
1780 }
1781 close(barrier.release)
1782 _, promptResp := drainPrompt(t, client, promptCh)
1783 if promptResp.Error != nil {
1784 t.Fatalf("session/prompt errored: %+v", promptResp.Error)
1785 }
1786
1787 reqs := prov.Requests()
1788 if len(reqs) != 2 {
1789 t.Fatalf("provider requests = %d, want 2", len(reqs))
1790 }
1791 found := false
1792 for _, m := range reqs[1].Messages {
1793 if text, ok := agent.SteerText(m.Content); ok && text == "use plan B" {
1794 found = true
1795 break
1796 }
1797 }
1798 if !found {
1799 t.Fatalf("second provider request did not contain the steer: %+v", reqs[1].Messages)
1800 }
1801
1802 idleResp := client.call(t, sessionSteerMethod, SessionSteerParams{
1803 SessionID: nr.SessionID,
1804 Prompt: []ContentBlock{{Type: "text", Text: "too late"}},
1805 })
1806 if idleResp.Error == nil || idleResp.Error.Code != ErrInvalidRequest {
1807 t.Fatalf("idle %s = %+v, want invalid request", sessionSteerMethod, idleResp.Error)
1808 }
1809 }
1810
1811 func TestServePromptErrorIsNotReportedAsCancelled(t *testing.T) {
1812 factory := &fakeFactory{behavior: func(context.Context, event.Sink, string) error {
1813 return errors.New("provider failed")
1814 }}
1815 client, stop := startServer(t, factory)
1816 defer stop()
1817
1818 client.call(t, "initialize", InitializeParams{ProtocolVersion: 1})
1819 newResp := client.call(t, "session/new", SessionNewParams{})
1820 var nr SessionNewResult
1821 json.Unmarshal(newResp.Result, &nr)
1822
1823 promptCh := client.callAsync("session/prompt", SessionPromptParams{
1824 SessionID: nr.SessionID,
1825 Prompt: []ContentBlock{{Type: "text", Text: "fail"}},
1826 })
1827 _, resp := drainPrompt(t, client, promptCh)
1828 var pr SessionPromptResult
1829 if err := json.Unmarshal(resp.Result, &pr); err != nil {
1830 t.Fatalf("prompt result: %v", err)
1831 }
1832 if pr.StopReason != StopError {
1833 t.Errorf("stopReason = %q, want error", pr.StopReason)
1834 }
1835 }
1836
1837 func TestServeRejectsConcurrentPromptForSameSession(t *testing.T) {
1838 started := make(chan struct{})
1839 release := make(chan struct{})
1840 var once sync.Once
1841 factory := &fakeFactory{behavior: func(_ context.Context, _ event.Sink, _ string) error {
1842 once.Do(func() { close(started) })
1843 <-release
1844 return nil
1845 }}
1846 client, stop := startServer(t, factory)
1847 defer stop()
1848
1849 client.call(t, "initialize", InitializeParams{ProtocolVersion: 1})
1850 newResp := client.call(t, "session/new", SessionNewParams{})
1851 var nr SessionNewResult
1852 json.Unmarshal(newResp.Result, &nr)
1853
1854 first := client.callAsync("session/prompt", SessionPromptParams{
1855 SessionID: nr.SessionID,
1856 Prompt: []ContentBlock{{Type: "text", Text: "first"}},
1857 })
1858 select {
1859 case <-started:
1860 case <-time.After(2 * time.Second):
1861 t.Fatal("first prompt never started")
1862 }
1863
1864 second := client.call(t, "session/prompt", SessionPromptParams{
1865 SessionID: nr.SessionID,
1866 Prompt: []ContentBlock{{Type: "text", Text: "second"}},
1867 })
1868 if second.Error == nil {
1869 t.Fatal("second concurrent prompt should return an error")
1870 }
1871 if second.Error.Code != ErrInvalidRequest || !strings.Contains(second.Error.Message, "active prompt") {
1872 t.Fatalf("second prompt error = %+v, want active-prompt invalid request", second.Error)
1873 }
1874
1875 close(release)
1876 select {
1877 case resp := <-first:
1878 if resp.Error != nil {
1879 t.Fatalf("first prompt errored: %+v", resp.Error)
1880 }
1881 case <-time.After(2 * time.Second):
1882 t.Fatal("first prompt did not finish")
1883 }
1884 }
1885
1886 func TestServeSessionClose(t *testing.T) {
1887 factory := &fakeFactory{behavior: func(context.Context, event.Sink, string) error { return nil }}
1888 client, stop := startServer(t, factory)
1889 defer stop()
1890
1891 client.call(t, "initialize", InitializeParams{ProtocolVersion: 1})
1892 newResp := client.call(t, "session/new", SessionNewParams{})
1893 var nr SessionNewResult
1894 json.Unmarshal(newResp.Result, &nr)
1895
1896 closeResp := client.call(t, "session/close", SessionCloseParams{SessionID: nr.SessionID})
1897 if closeResp.Error != nil {
1898 t.Fatalf("session/close errored: %+v", closeResp.Error)
1899 }
1900
1901 promptResp := client.call(t, "session/prompt", SessionPromptParams{
1902 SessionID: nr.SessionID,
1903 Prompt: []ContentBlock{{Type: "text", Text: "after close"}},
1904 })
1905 if promptResp.Error == nil || !strings.Contains(promptResp.Error.Message, "unknown session") {
1906 t.Fatalf("prompt after close error = %+v, want unknown session", promptResp.Error)
1907 }
1908 }
1909
1910 func TestSessionDeleteWithStuckJobReturnsAfterSingleGrace(t *testing.T) {
1911 dir := t.TempDir()
1912 grace := time.Second
1913 maxElapsed := grace + 750*time.Millisecond
1914 factory := &teardownFactory{dir: dir, grace: grace}
1915 client, stop := startServer(t, factory)
1916 defer stop()
1917
1918 client.call(t, "initialize", InitializeParams{ProtocolVersion: 1})
1919 newResp := client.call(t, "session/new", SessionNewParams{Cwd: t.TempDir()})
1920 var nr SessionNewResult
1921 if err := json.Unmarshal(newResp.Result, &nr); err != nil || nr.SessionID == "" {
1922 t.Fatalf("session/new: %v (%q)", err, nr.SessionID)
1923 }
1924 path := transcriptPath(dir, nr.SessionID)
1925 if err := os.WriteFile(path, []byte(`{"role":"user","content":"hello"}`+"\n"), 0o644); err != nil {
1926 t.Fatalf("write transcript: %v", err)
1927 }
1928 releaseJob := startNonCooperativeACPJob(t, factory.lastManager(t), path)
1929 defer releaseJob()
1930
1931 start := time.Now()
1932 resp := client.call(t, "session/delete", SessionDeleteParams{SessionID: nr.SessionID})
1933 elapsed := time.Since(start)
1934 if resp.Error != nil {
1935 t.Fatalf("session/delete errored: %+v", resp.Error)
1936 }
1937 if elapsed > maxElapsed {
1938 t.Fatalf("session/delete took %s, want one teardown grace plus scheduling slack", elapsed)
1939 }
1940 if !agent.IsCleanupPending(path) {
1941 t.Fatalf("stuck ACP delete should mark cleanup pending")
1942 }
1943 if _, err := os.Stat(path); err != nil {
1944 t.Fatalf("stuck ACP transcript should remain until delayed cleanup: %v", err)
1945 }
1946 releaseJob()
1947 deadline := time.Now().Add(2 * time.Second)
1948 for agent.IsCleanupPending(path) {
1949 if time.Now().After(deadline) {
1950 t.Fatalf("cleanup-pending marker was not cleared after stuck job release")
1951 }
1952 time.Sleep(10 * time.Millisecond)
1953 }
1954 }
1955
1956 func TestServeRejectsPathLikeSessionID(t *testing.T) {
1957 factory := &fakeFactory{behavior: func(context.Context, event.Sink, string) error { return nil }}
1958 client, stop := startServer(t, factory)
1959 defer stop()
1960
1961 resp := client.call(t, "session/delete", SessionDeleteParams{SessionID: "../outside"})
1962 if resp.Error == nil {
1963 t.Fatal("session/delete with path-like sessionId should fail")
1964 }
1965 if resp.Error.Code != ErrInvalidParams || !strings.Contains(resp.Error.Message, "invalid sessionId") {
1966 t.Fatalf("session/delete error = %+v, want invalid sessionId", resp.Error)
1967 }
1968 }
1969
1970 func TestListACPMetasSkipsCleanupPending(t *testing.T) {
1971 dir := t.TempDir()
1972 visibleID := "visible"
1973 pendingID := "pending"
1974 for _, id := range []string{visibleID, pendingID} {
1975 path := transcriptPath(dir, id)
1976 if err := os.WriteFile(path, []byte(`{"role":"user","content":"hi"}`+"\n"), 0o644); err != nil {
1977 t.Fatal(err)
1978 }
1979 if err := saveACPMeta(path, acpSessionMeta{
1980 SessionID: id,
1981 Cwd: t.TempDir(),
1982 Title: id,
1983 CreatedAt: time.Now().UTC(),
1984 UpdatedAt: time.Now().UTC(),
1985 }); err != nil {
1986 t.Fatal(err)
1987 }
1988 }
1989 if err := agent.MarkCleanupPending(transcriptPath(dir, pendingID), "delete"); err != nil {
1990 t.Fatal(err)
1991 }
1992
1993 metas, err := listACPMetas(dir)
1994 if err != nil {
1995 t.Fatal(err)
1996 }
1997 if len(metas) != 1 || metas[0].SessionID != visibleID {
1998 t.Fatalf("listACPMetas = %+v, want only %q", metas, visibleID)
1999 }
2000 }
2001
2002 func TestDeleteSessionFilesDeletesOwnedSubagents(t *testing.T) {
2003 dir := t.TempDir()
2004 sessionPath := filepath.Join(dir, "session.jsonl")
2005 if err := os.WriteFile(sessionPath, []byte(`{"role":"user","content":"hi"}`+"\n"), 0o644); err != nil {
2006 t.Fatal(err)
2007 }
2008 ref := "sa_20260102_030405_000000000_aabbccddeeff"
2009 writeACPSubagentArtifact(t, dir, ref, agent.BranchID(sessionPath))
2010 jobsDir := jobs.ArtifactDir(sessionPath)
2011 if err := os.MkdirAll(jobsDir, 0o755); err != nil {
2012 t.Fatal(err)
2013 }
2014 if err := os.WriteFile(filepath.Join(jobsDir, "bash-1.log"), []byte("output"), 0o644); err != nil {
2015 t.Fatal(err)
2016 }
2017
2018 if err := deleteSessionFiles(sessionPath); err != nil {
2019 t.Fatalf("deleteSessionFiles: %v", err)
2020 }
2021 if _, err := os.Stat(filepath.Join(dir, "subagents", ref+".jsonl")); !os.IsNotExist(err) {
2022 t.Fatalf("subagent jsonl should be deleted, stat err = %v", err)
2023 }
2024 if _, err := os.Stat(filepath.Join(dir, "subagents", ref+".meta.json")); !os.IsNotExist(err) {
2025 t.Fatalf("subagent meta should be deleted, stat err = %v", err)
2026 }
2027 if _, err := os.Stat(jobsDir); !os.IsNotExist(err) {
2028 t.Fatalf("jobs sidecar should be deleted, stat err = %v", err)
2029 }
2030 }
2031
2032 func TestReconcileCleanupPendingDeletesACPMeta(t *testing.T) {
2033 dir := t.TempDir()
2034 sessionPath := transcriptPath(dir, "pending-acp")
2035 if err := os.WriteFile(sessionPath, []byte(`{"role":"user","content":"hi"}`+"\n"), 0o644); err != nil {
2036 t.Fatal(err)
2037 }
2038 if err := saveACPMeta(sessionPath, acpSessionMeta{Cwd: t.TempDir(), Model: "test-model"}); err != nil {
2039 t.Fatal(err)
2040 }
2041 jobsDir := jobs.ArtifactDir(sessionPath)
2042 if err := os.MkdirAll(jobsDir, 0o755); err != nil {
2043 t.Fatal(err)
2044 }
2045 if err := os.WriteFile(filepath.Join(jobsDir, "bash-1.log"), []byte("output"), 0o644); err != nil {
2046 t.Fatal(err)
2047 }
2048 if err := agent.MarkCleanupPending(sessionPath, "delete"); err != nil {
2049 t.Fatal(err)
2050 }
2051
2052 if err := ReconcileCleanupPending(dir); err != nil {
2053 t.Fatalf("ReconcileCleanupPending: %v", err)
2054 }
2055 for _, path := range []string{sessionPath, acpMetaPath(sessionPath), jobsDir, agent.CleanupPendingPath(sessionPath)} {
2056 if _, err := os.Stat(path); !os.IsNotExist(err) {
2057 t.Fatalf("%s still exists after reconciliation (err=%v)", path, err)
2058 }
2059 }
2060 }
2061
2062 func writeACPSubagentArtifact(t *testing.T, dir, ref, parentSession string) {
2063 t.Helper()
2064 subagentDir := filepath.Join(dir, "subagents")
2065 if err := os.MkdirAll(subagentDir, 0o755); err != nil {
2066 t.Fatal(err)
2067 }
2068 if err := os.WriteFile(filepath.Join(subagentDir, ref+".jsonl"), []byte(`{"role":"user","content":"sub"}`+"\n"), 0o644); err != nil {
2069 t.Fatal(err)
2070 }
2071 data, err := json.Marshal(agent.SubagentMeta{
2072 Ref: ref,
2073 Status: agent.SubagentCompleted,
2074 Kind: "task",
2075 Name: "task",
2076 ParentSession: parentSession,
2077 })
2078 if err != nil {
2079 t.Fatal(err)
2080 }
2081 if err := os.WriteFile(filepath.Join(subagentDir, ref+".meta.json"), data, 0o644); err != nil {
2082 t.Fatal(err)
2083 }
2084 }
2085
2086 func startNonCooperativeACPJob(t *testing.T, jm *jobs.Manager, sessionPath string) func() {
2087 t.Helper()
2088 started := make(chan struct{})
2089 release := make(chan struct{})
2090 jm.StartForSession(agent.BranchID(sessionPath), "bash", "stuck job", func(ctx context.Context, _ io.Writer) (string, error) {
2091 close(started)
2092 <-ctx.Done()
2093 <-release
2094 return "", ctx.Err()
2095 })
2096 select {
2097 case <-started:
2098 case <-time.After(2 * time.Second):
2099 t.Fatal("background job never started")
2100 }
2101 released := false
2102 return func() {
2103 if released {
2104 return
2105 }
2106 released = true
2107 close(release)
2108 }
2109 }
2110
2111 func TestServeUnknownMethod(t *testing.T) {
2112 factory := &fakeFactory{behavior: func(context.Context, event.Sink, string) error { return nil }}
2113 client, stop := startServer(t, factory)
2114 defer stop()
2115
2116 resp := client.call(t, "does/not/exist", nil)
2117 if resp.Error == nil {
2118 t.Fatal("expected an error response")
2119 }
2120 if resp.Error.Code != ErrMethodNotFound {
2121 t.Errorf("error code = %d, want %d", resp.Error.Code, ErrMethodNotFound)
2122 }
2123 }
2124
2124 lines GO