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