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