1package subagents
2
3import (
4 "context"
5 "encoding/json"
6 "fmt"
7 "os"
8 "os/exec"
9 "path/filepath"
10 "runtime"
11 "strings"
12 "testing"
13 "time"
14)
15
16func TestDeadlineShutdownCommandIdentifiesOrigin(t *testing.T) {
17 if runtime.GOOS == "windows" {
18 t.Skip("subagent inbox transport uses Unix-domain sockets")
19 }
20 path := filepath.Join(shortSocketDir(t), "deadline.sock")
21 listener, err := Listen(path)
22 if err != nil {
23 t.Fatal(err)
24 }
25 defer listener.Close()
26
27 a := &Agent{ID: "deadline-agent", InboxPath: path, inbox: NewInbox(path)}
28 defer a.inbox.Close()
29 r := &execRunner{agent: a, GracePeriod: time.Second}
30 ctx, cancel := context.WithDeadline(context.Background(), time.Now().Add(-time.Second))
31 defer cancel()
32 runnerDone := make(chan struct{})
33 watcherDone := make(chan struct{})
34 go func() {
35 r.stopOnContextDone(ctx, &exec.Cmd{}, runnerDone)
36 close(watcherDone)
37 }()
38
39 var command Envelope
40 select {
41 case line := <-listener.Lines():
42 command, err = ParseCommand(line)
43 if err != nil {
44 t.Fatal(err)
45 }
46 case <-time.After(time.Second):
47 t.Fatal("deadline shutdown command was not received")
48 }
49 close(runnerDone)
50 <-watcherDone
51
52 var payload AgentShutdownPayload
53 if err := command.DecodePayload(&payload); err != nil {
54 t.Fatal(err)
55 }
56 if payload.Origin != ShutdownOriginDeadline {
57 t.Fatalf("shutdown origin = %q, want %q", payload.Origin, ShutdownOriginDeadline)
58 }
59}
60
61// TestSubagentWorkerArgs locks in the exact flag set the subprocess
62// runner uses to start a subagent agent in daemon mode. Past
63// regressions in this area:
64//
65// - "--no-sess" instead of "--no-session" (old print-mode runner):
66// every spawned agent died with "unknown flag" before it could
67// talk to the model.
68//
69// - Forgetting --cwd: the child resolved tools against the parent
70// zut's working directory, defeating the whole point of the
71// worktree isolation.
72//
73// - Forgetting --session: a daemon-mode agent without a session
74// file would lose context between follow-up turns, making
75// "send another message" mostly useless.
76//
77// The test asserts the load-bearing pieces are present in plausible
78// positions. If a flag is renamed, update both the runner and this
79// test so we notice immediately.
80func TestCredentialStdinHelperProcess(t *testing.T) {
81 if os.Getenv("ZUT_SUBAGENT_CREDENTIAL_HELPER") != "1" {
82 return
83 }
84 if os.Getenv("ZUT_SUBAGENT_CREDENTIAL_STDIN") != "1" {
85 os.Exit(2)
86 }
87 var credential Credential
88 if err := json.NewDecoder(os.Stdin).Decode(&credential); err != nil {
89 os.Exit(3)
90 }
91 if credential.Value != "inherited-secret" || credential.Method != "apikey" {
92 os.Exit(4)
93 }
94 for _, arg := range os.Args {
95 if strings.Contains(arg, credential.Value) {
96 os.Exit(5)
97 }
98 }
99 for _, env := range os.Environ() {
100 if strings.Contains(env, credential.Value) {
101 os.Exit(6)
102 }
103 }
104 fmt.Println(`{"type":"agent_stopped","data":{"reason":"completed"}}`)
105 os.Exit(0)
106}
107
108// TestMessageDeltaStreamsAndReplaysWithoutDuplicatingFinalMessage verifies
109// that the durable event stream has the same visible result while live and
110// after history replay.
111func TestMessageDeltaStreamsAndReplaysWithoutDuplicatingFinalMessage(t *testing.T) {
112 events := []Event{
113 NewEvent("message.delta", map[string]any{"delta": "partial "}),
114 NewEvent("message.delta", map[string]any{"delta": "answer"}),
115 NewEvent("assistant_message", map[string]any{"content": []any{
116 map[string]any{"type": "text", "text": "partial answer"},
117 }}),
118 }
119
120 live := &Agent{transcriptLoaded: true}
121 for _, event := range events {
122 applyEventToSink(event, agentSink{a: live})
123 }
124 if got := strings.Join(live.Transcript(), "\n"); got != "partial answer" {
125 t.Fatalf("live transcript = %q, want final assistant message once", got)
126 }
127 if got := live.Snapshot().LastAssistant; got != "partial answer" {
128 t.Fatalf("live assistant output = %q, want %q", got, "partial answer")
129 }
130
131 replayed := &Agent{transcriptLoaded: true}
132 for _, event := range events {
133 replayEventTranscript(replayed, event)
134 }
135 if got := strings.Join(replayed.Transcript(), "\n"); got != "partial answer" {
136 t.Fatalf("replayed transcript = %q, want final assistant message once", got)
137 }
138}
139
140func TestMessageDeltaBoundsLiveProjection(t *testing.T) {
141 a := &Agent{maxOutputBytes: 32, maxOutputLines: 100, transcriptLoaded: true}
142 applyEventToSink(NewEvent("message.delta", map[string]any{"delta": strings.Repeat("x", 64)}), agentSink{a: a})
143 applyEventToSink(NewEvent("message.delta", map[string]any{"delta": "discarded"}), agentSink{a: a})
144
145 a.mu.Lock()
146 streamed := a.streamingAssistantText
147 truncated := a.streamingAssistantTruncated
148 a.mu.Unlock()
149 if !truncated {
150 t.Fatal("streamed output was not marked truncated")
151 }
152 if got := len([]byte(streamed)); got > 32 {
153 t.Fatalf("streamed output length = %d, want at most 32", got)
154 }
155 snapshot := a.Snapshot()
156 if got := snapshot.LastAssistant; got != streamed {
157 t.Fatalf("live assistant output = %q, want bounded stream %q", got, streamed)
158 }
159 if !snapshot.OutputTruncated {
160 t.Fatal("snapshot did not expose truncated streamed output")
161 }
162
163 events := []Event{
164 NewEvent(EventMessageDelta, map[string]any{"delta": "first partial"}),
165 NewEvent("turn_end", map[string]any{"stop": "cancelled"}),
166 NewEvent(EventMessageDelta, map[string]any{"delta": "second partial"}),
167 }
168 want := "first partial\nsecond partial"
169 assertProjection := func(t *testing.T, agent *Agent) {
170 t.Helper()
171 if got := strings.Join(agent.Transcript(), "\n"); got != want {
172 t.Fatalf("transcript = %q, want %q", got, want)
173 }
174 agent.mu.Lock()
175 truncated := agent.streamingAssistantTruncated
176 agent.mu.Unlock()
177 if truncated {
178 t.Fatal("fresh second-turn stream was incorrectly truncated")
179 }
180 }
181
182 live := &Agent{maxOutputBytes: 100, maxOutputLines: 100, transcriptLoaded: true}
183 for _, event := range events {
184 applyEventToSink(event, agentSink{a: live})
185 }
186 assertProjection(t, live)
187
188 replayed := &Agent{maxOutputBytes: 100, maxOutputLines: 100, transcriptLoaded: true}
189 replayTranscriptIntoAgent(replayed, events)
190 assertProjection(t, replayed)
191
192 legacy := &Agent{maxOutputBytes: 100, maxOutputLines: 100, transcriptLoaded: true}
193 replayEventsIntoAgent(legacy, events)
194 assertProjection(t, legacy)
195
196 truncatedTurn := &Agent{maxOutputBytes: 16, maxOutputLines: 100, transcriptLoaded: true}
197 applyEventToSink(NewEvent(EventMessageDelta, map[string]any{"delta": strings.Repeat("x", 64)}), agentSink{a: truncatedTurn})
198 applyEventToSink(NewEvent("turn_end", map[string]any{"stop": "cancelled"}), agentSink{a: truncatedTurn})
199 truncatedTurn.mu.Lock()
200 streamText := truncatedTurn.streamingAssistantText
201 streamTruncated := truncatedTurn.streamingAssistantTruncated
202 truncatedTurn.mu.Unlock()
203 if streamText != "" || streamTruncated {
204 t.Fatalf("stream state after turn boundary = (%q, %v), want empty and not truncated", streamText, streamTruncated)
205 }
206}
207
208func TestWorkerEnvironmentRedactsProviderSecrets(t *testing.T) {
209 t.Setenv("OPENAI_API_KEY", "openai-secret")
210 t.Setenv("CUSTOM_PROVIDER_API_KEY", "custom-secret")
211 t.Setenv("HF_TOKEN", "hf-secret")
212 t.Setenv("GITHUB_TOKEN", "github-secret")
213 t.Setenv("CUSTOM_PROVIDER_TOKEN", "token-secret")
214 t.Setenv("CUSTOM_PROVIDER_KEY", "key-secret")
215 t.Setenv("AWS_ACCESS_KEY_ID", "aws-access-secret")
216 t.Setenv("GOOGLE_APPLICATION_CREDENTIALS", "/tmp/service-account.json")
217 t.Setenv("ZUT_SUBAGENT_TEST_VALUE", "not-secret")
218 env := strings.Join(workerEnvironment("openai"), "\n")
219 for _, secret := range []string{
220 "OPENAI_API_KEY=openai-secret",
221 "CUSTOM_PROVIDER_API_KEY=custom-secret",
222 "HF_TOKEN=hf-secret",
223 "GITHUB_TOKEN=github-secret",
224 "CUSTOM_PROVIDER_TOKEN=token-secret",
225 "CUSTOM_PROVIDER_KEY=key-secret",
226 "AWS_ACCESS_KEY_ID=aws-access-secret",
227 "GOOGLE_APPLICATION_CREDENTIALS=/tmp/service-account.json",
228 } {
229 if strings.Contains(env, secret) {
230 t.Fatalf("worker environment leaked %s", secret)
231 }
232 }
233 if !strings.Contains(env, "ZUT_SUBAGENT_TEST_VALUE=not-secret") {
234 t.Fatal("worker environment dropped unrelated variables")
235 }
236
237 t.Setenv("AWS_BEARER_TOKEN_BEDROCK", "bedrock-secret")
238 bedrockEnv := strings.Join(workerEnvironment("amazon-bedrock"), "\n")
239 if !strings.Contains(bedrockEnv, "AWS_BEARER_TOKEN_BEDROCK=bedrock-secret") {
240 t.Fatal("Bedrock ambient credential chain was unexpectedly removed")
241 }
242 if !strings.Contains(bedrockEnv, "AWS_ACCESS_KEY_ID=aws-access-secret") {
243 t.Fatal("Bedrock AWS access-key credential was unexpectedly removed")
244 }
245}
246
247func TestExecRunnerTransfersCredentialOnlyOnStdin(t *testing.T) {
248 t.Setenv("ZUT_SUBAGENT_CREDENTIAL_HELPER", "1")
249 root := t.TempDir()
250 a := &Agent{
251 ID: "credential-test",
252 Dir: root,
253 Provider: "custom",
254 SessionPath: filepath.Join(root, "session.json"),
255 InboxPath: filepath.Join(root, "inbox"),
256 EventLogPath: filepath.Join(root, "events.jsonl"),
257 }
258 calls := 0
259 r := &execRunner{
260 agent: a,
261 Command: []string{os.Args[0], "-test.run=^TestCredentialStdinHelperProcess$"},
262 resolveCredential: func(context.Context, string) (Credential, error) {
263 calls++
264 return Credential{Value: "inherited-secret", Method: "apikey"}, nil
265 },
266 }
267 if err := r.Run(context.Background(), agentSink{a: a}); err != nil {
268 t.Fatal(err)
269 }
270 if calls != 1 {
271 t.Fatalf("credential resolver called %d times, want 1", calls)
272 }
273}
274
275func TestApplyEventPreservesMultilineRolesAndFinalAssistant(t *testing.T) {
276 a := &Agent{}
277 sink := agentSink{a: a}
278 applyEventToSink(NewEvent("user_message", map[string]any{"content": []any{
279 map[string]any{"type": "text", "text": "first line\n\nlast line"},
280 }}), sink)
281 applyEventToSink(NewEvent("assistant_message", map[string]any{"content": []any{
282 map[string]any{"type": "text", "text": "complete\nanswer"},
283 }}), sink)
284
285 snap := a.Snapshot()
286 wantLines := []string{"user: first line", "user: ", "user: last line", "complete", "answer"}
287 if strings.Join(snap.Lines, "\n") != strings.Join(wantLines, "\n") {
288 t.Fatalf("transcript = %#v, want %#v", snap.Lines, wantLines)
289 }
290 if snap.LastAssistant != "complete\nanswer" {
291 t.Fatalf("last assistant = %q", snap.LastAssistant)
292 }
293}
294
295func TestSubagentWorkerArgs(t *testing.T) {
296 args := subagentWorkerArgs(subagentWorkerArgsOpts{
297 Exe: "/path/to/zut",
298 Dir: "/tmp/worktree",
299 SessionPath: "/tmp/state/session.json",
300 InboxPath: "/tmp/state/in.sock",
301 Task: "do the thing",
302 Reasoning: "high",
303 Subagent: "reviewer",
304 TurnTimeout: 15 * time.Minute,
305 })
306 if len(args) < 7 {
307 t.Fatalf("argv unexpectedly short: %v", args)
308 }
309 if args[0] != "/path/to/zut" {
310 t.Fatalf("argv[0] = %q; want the binary path", args[0])
311 }
312 // The task must come last so anything that looks flag-like in
313 // the task body doesn't get interpreted as a flag.
314 if args[len(args)-1] != "do the thing" {
315 t.Fatalf("task should be last positional; got %v", args)
316 }
317
318 mustHave := map[string]string{
319 "--subagent-worker": "/tmp/state/in.sock",
320 "--session": "/tmp/state/session.json",
321 "--cwd": "/tmp/worktree",
322 "--reasoning": "high",
323 "--subagent": "reviewer",
324 "--subagent-turn-timeout": "15m0s",
325 }
326 for flag, value := range mustHave {
327 i := indexOf(args, flag)
328 if i < 0 {
329 t.Errorf("argv missing %q: %v", flag, args)
330 continue
331 }
332 if i+1 >= len(args) || args[i+1] != value {
333 t.Errorf("argv %q value = %q; want %q", flag, safeAt(args, i+1), value)
334 }
335 }
336
337 // Reject prior bad flags explicitly so a future revert is caught.
338 joined := strings.Join(args, " ")
339 for _, bad := range []string{"--print", "--no-sess ", "--no-session"} {
340 if strings.Contains(joined, bad) {
341 t.Fatalf("argv contains stale/wrong flag %q: %s", bad, joined)
342 }
343 }
344}
345
346func TestDefaultChildArgsPropagatesWebSearchPolicy(t *testing.T) {
347 allow := defaultChildArgs("zut", &Agent{Task: "task", WebSearchPolicy: WebSearchAllow}, "/session", "/inbox")
348 idx := indexOf(allow, "--web-search-policy")
349 if idx < 0 || safeAt(allow, idx+1) != "allow" {
350 t.Fatalf("allow argv = %v", allow)
351 }
352
353 // A legacy/hand-built Agent has no resolved parent policy and must not
354 // inherit the child process's default-enabled normal CLI setting.
355 deny := defaultChildArgs("zut", &Agent{Task: "task"}, "/session", "/inbox")
356 idx = indexOf(deny, "--web-search-policy")
357 if idx < 0 || safeAt(deny, idx+1) != "deny" {
358 t.Fatalf("deny argv = %v", deny)
359 }
360}
361
362func TestResolveSubagentExecutableUsesCurrentWhenAvailable(t *testing.T) {
363 got, err := resolveSubagentExecutable(context.Background(), "/current/bin/zut", "/old/bin/zut", func(_ context.Context, candidate string) (string, error) {
364 if candidate == "/current/bin/zut" {
365 return candidate, nil
366 }
367 return "", os.ErrNotExist
368 })
369 if err != nil {
370 t.Fatal(err)
371 }
372 if got != "/current/bin/zut" {
373 t.Fatalf("executable = %q, want current executable", got)
374 }
375}
376
377func TestResolveSubagentExecutableFallsBackToPath(t *testing.T) {
378 got, err := resolveSubagentExecutable(context.Background(), "/removed/.local/bin/zut", "/removed/.local/bin/zut", func(_ context.Context, candidate string) (string, error) {
379 if candidate == "zut" {
380 return "/active/go/bin/zut", nil
381 }
382 return "", os.ErrNotExist
383 })
384 if err != nil {
385 t.Fatal(err)
386 }
387 if got != "/active/go/bin/zut" {
388 t.Fatalf("executable = %q, want PATH fallback", got)
389 }
390}
391
392func TestResolveSubagentExecutableCancelsOtherLookupsAfterFirstSuccess(t *testing.T) {
393 slowStarted := make(chan struct{})
394 slowCanceled := make(chan struct{})
395 got, err := resolveSubagentExecutable(context.Background(), "/current/bin/zut", "/old/bin/zut", func(ctx context.Context, candidate string) (string, error) {
396 switch candidate {
397 case "/current/bin/zut":
398 close(slowStarted)
399 <-ctx.Done()
400 close(slowCanceled)
401 return "", ctx.Err()
402 case "zut":
403 <-slowStarted
404 return "/active/go/bin/zut", nil
405 default:
406 return "", os.ErrNotExist
407 }
408 })
409 if err != nil {
410 t.Fatal(err)
411 }
412 if got != "/active/go/bin/zut" {
413 t.Fatalf("executable = %q, want first successful lookup", got)
414 }
415 select {
416 case <-slowCanceled:
417 case <-time.After(time.Second):
418 t.Fatal("slower lookup was not canceled")
419 }
420}
421
422func TestSubagentWorkerArgsPropagatesProviderConnectionSettings(t *testing.T) {
423 args := subagentWorkerArgs(subagentWorkerArgsOpts{
424 Exe: "/zut",
425 Dir: "/work",
426 SessionPath: "/state/session.json",
427 InboxPath: "/state/inbox.sock",
428 BaseURL: "https://gateway.example.test/v1",
429 InsecureTLS: true,
430 })
431 if i := indexOf(args, "--base-url"); i < 0 || safeAt(args, i+1) != "https://gateway.example.test/v1" {
432 t.Fatalf("argv = %v, want --base-url with inherited endpoint", args)
433 }
434 if !containsArg(args, "--insecure") {
435 t.Fatalf("argv = %v, want inherited --insecure", args)
436 }
437 for _, arg := range args {
438 if strings.Contains(arg, "api-key") || strings.Contains(arg, "secret") {
439 t.Fatalf("argv contains credential material: %v", args)
440 }
441 }
442}
443
444func TestDefaultChildArgsPropagatesProviderConnectionSettings(t *testing.T) {
445 a := &Agent{
446 Dir: "/work",
447 BaseURL: "https://gateway.example.test/v1",
448 InsecureTLS: true,
449 Timeout: 15 * time.Minute,
450 }
451 args := defaultChildArgs("/zut", a, "/state/session.json", "/state/inbox.sock")
452 if i := indexOf(args, "--base-url"); i < 0 || safeAt(args, i+1) != a.BaseURL {
453 t.Fatalf("argv = %v, want Agent.BaseURL propagated", args)
454 }
455 if !containsArg(args, "--insecure") {
456 t.Fatalf("argv = %v, want Agent.InsecureTLS propagated", args)
457 }
458 if i := indexOf(args, "--subagent-turn-timeout"); i < 0 || safeAt(args, i+1) != a.Timeout.String() {
459 t.Fatalf("argv = %v, want Agent.Timeout propagated per turn", args)
460 }
461}
462
463// TestSubagentWorkerArgsEmptyTaskOmitsPositional makes sure that when the
464// agent is being adopted (no fresh task) we don't pass an empty
465// positional which the arg parser would treat as a real prompt.
466func TestSubagentWorkerArgsEmptyTaskOmitsPositional(t *testing.T) {
467 args := subagentWorkerArgs(subagentWorkerArgsOpts{
468 Exe: "/zut", Dir: "/wt", SessionPath: "/s.json", InboxPath: "/in.sock",
469 })
470 for _, a := range args {
471 if a == "" {
472 t.Fatalf("argv contains an empty positional: %v", args)
473 }
474 }
475 // last arg should be a real flag value, not a stray positional
476 if a := args[len(args)-1]; strings.HasPrefix(a, "--") {
477 t.Fatalf("argv ends on a flag with no value: %v", args)
478 }
479 for _, flag := range []string{"--reasoning", "--subagent"} {
480 if indexOf(args, flag) >= 0 {
481 t.Fatalf("empty child options should omit %s: %v", flag, args)
482 }
483 }
484}
485
486// TestDefaultChildArgsSpawnIncludesTask pins the spawn shape: a
487// fresh (non-resuming) Agent produces argv that ends with the
488// original task as a positional, so the child runs it as the
489// initial user turn.
490func TestDefaultChildArgsSpawnIncludesTask(t *testing.T) {
491 a := &Agent{Dir: "/wt", Task: "do thing"}
492 args := defaultChildArgs("/zut", a, "/s.json", "/in.sock")
493 if got := args[len(args)-1]; got != "do thing" {
494 t.Fatalf("spawn argv last = %q; want %q\n%v", got, "do thing", args)
495 }
496}
497
498// TestDefaultChildArgsResumeOmitsTask is the regression for the
499// "agent busy; send 'cancel' first" error: when an Agent is being
500// resumed (Resuming==true), the child argv MUST NOT include the
501// original Task as a positional. Otherwise the child fires the task
502// as a fresh user turn on every resume, racing with whatever the
503// user types next via the inbox.
504func TestDefaultChildArgsResumeOmitsTask(t *testing.T) {
505 a := &Agent{Dir: "/wt", Task: "do thing", Resuming: true}
506 args := defaultChildArgs("/zut", a, "/s.json", "/in.sock")
507 for _, v := range args {
508 if v == "do thing" {
509 t.Fatalf("resume argv contains the task; it would re-fire as a duplicate turn\n%v", args)
510 }
511 }
512 // And no trailing positional at all: every final token must be a
513 // recognized boolean flag or a value belonging to a preceding flag,
514 // never a stray empty string or the original task.
515 if got := args[len(args)-1]; got == "" {
516 t.Fatalf("resume argv ends with an empty positional: %v", args)
517 }
518}
519
520func TestDefaultChildArgsResumeUsesFollowUpPrompt(t *testing.T) {
521 a := &Agent{
522 Dir: "/wt",
523 Task: "review the change",
524 Resuming: true,
525 }
526 const followUp = "I applied your feedback. Please review it again."
527 a.setResumePrompt(followUp, time.Now())
528 args := defaultChildArgs("/zut", a, "/s.json", "/in.sock")
529 if got := args[len(args)-1]; got != followUp {
530 t.Fatalf("resume argv last = %q, want follow-up prompt %q\n%v", got, followUp, args)
531 }
532 for _, v := range args[:len(args)-1] {
533 if v == a.Task {
534 t.Fatalf("resume argv contains original task; it would duplicate the first turn\n%v", args)
535 }
536 }
537}
538
539func TestReadyDoesNotMarkInitialDelegatedTurnIdle(t *testing.T) {
540 a := &Agent{Task: "initial delegated task"}
541 a.setTurnState(TurnQueued, "")
542
543 if err := updateAgentFromEvent(a, NewEvent(EventAgentReady, map[string]any{"lifetime_turns": 0, "current_run_turns": 0})); err != nil {
544 t.Fatal(err)
545 }
546 if got := a.TurnState(); got != TurnQueued {
547 t.Fatalf("ready changed turn state to %s, want %s", got, TurnQueued)
548 }
549 if err := updateAgentFromEvent(a, NewEvent(EventTurnStarted, map[string]any{
550 "turn_id": "turn-1",
551 "lifetime_turns": 1,
552 "current_run_turns": 1,
553 })); err != nil {
554 t.Fatal(err)
555 }
556 if got := a.TurnState(); got != TurnRunning {
557 t.Fatalf("turn.started changed turn state to %s, want %s", got, TurnRunning)
558 }
559}
560
561func TestUpdateAgentFromEventIgnoresStaleRejectedFollowUp(t *testing.T) {
562 a := &Agent{}
563 a.setResumePrompt("current prompt", time.Now())
564 if err := updateAgentFromEvent(a, NewEvent("error", map[string]any{
565 "code": "turn_rejected", "reason": "busy", "command_id": "stale-command",
566 })); err != nil {
567 t.Fatal(err)
568 }
569 if prompt, _ := a.ResumePromptInfo(); prompt != "current prompt" {
570 t.Fatalf("stale rejection changed active prompt to %q", prompt)
571 }
572}
573
574func TestUpdateAgentFromEventRecoversRejectedFollowUp(t *testing.T) {
575 a := &Agent{}
576 a.setResumePrompt("retry me", time.Now())
577 a.setTurnState(TurnQueued, "turn-1")
578 if err := updateAgentFromEvent(a, NewEvent("error", map[string]any{
579 "code": "turn_rejected",
580 "reason": "busy",
581 "command_id": a.resumePromptCommandID(),
582 })); err != nil {
583 t.Fatal(err)
584 }
585 if got := a.TurnState(); got != TurnIdle {
586 t.Fatalf("rejected turn state = %s, want %s", got, TurnIdle)
587 }
588 if prompt, _ := a.ResumePromptInfo(); prompt != "" {
589 t.Fatalf("rejected prompt remained active: %q", prompt)
590 }
591 queued := a.resumePromptQueueSnapshot()
592 if len(queued) != 1 || queued[0].Prompt != "retry me" {
593 t.Fatalf("rejected prompt queue = %+v, want retry me", queued)
594 }
595}
596
597func TestUpdateAgentFromEventIgnoresProviderToolLoopTurnEnd(t *testing.T) {
598 a := &Agent{}
599 a.setTurnState(TurnRunning, "turn-1")
600 persisted := 0
601 a.persistFn = func(*Agent) error { persisted++; return nil }
602
603 updateAgentFromEvent(a, NewEvent("turn_end", map[string]any{"stop": "tool_use"}))
604
605 if got := a.TurnState(); got != TurnRunning {
606 t.Fatalf("turn state = %q, want %q", got, TurnRunning)
607 }
608 if got := a.CurrentTurnID(); got != "turn-1" {
609 t.Fatalf("turn ID = %q, want turn-1", got)
610 }
611 if persisted != 0 {
612 t.Fatalf("persist called %d times for provider/tool-loop turn_end, want 0", persisted)
613 }
614}
615
616func TestUpdateAgentFromEventPersistsTerminalTurnEndWithStep(t *testing.T) {
617 a := &Agent{}
618 persisted := 0
619 a.persistFn = func(*Agent) error { persisted++; return nil }
620
621 updateAgentFromEvent(a, NewEvent("turn_end", map[string]any{"step": float64(2), "error": "boom"}))
622
623 if got := a.TurnState(); got != TurnFailed {
624 t.Fatalf("turn state = %q, want %q", got, TurnFailed)
625 }
626 if persisted != 1 {
627 t.Fatalf("persist called %d times for terminal turn_end, want 1", persisted)
628 }
629}
630
631func TestNotifyPromptTurnEndIgnoresProviderToolLoopTurnEnd(t *testing.T) {
632 called := make(chan struct{}, 1)
633 a := &Agent{}
634 a.SetOnTurnEnd(func(step int, errMsg string) {
635 called <- struct{}{}
636 })
637
638 notifyPromptTurnEnd(a, NewEvent("turn_end", map[string]any{"stop": "tool_use"}))
639
640 select {
641 case <-called:
642 t.Fatal("OnTurnEnd fired for provider/tool-loop turn_end without step")
643 case <-time.After(50 * time.Millisecond):
644 }
645}
646
647func TestNotifyPromptTurnEndFiresForDaemonPromptCompletion(t *testing.T) {
648 type got struct {
649 step int
650 err string
651 }
652 called := make(chan got, 1)
653 a := &Agent{}
654 a.SetOnTurnEnd(func(step int, errMsg string) {
655 called <- got{step: step, err: errMsg}
656 })
657
658 notifyPromptTurnEnd(a, NewEvent("turn_end", map[string]any{"step": float64(2), "error": "boom"}))
659
660 select {
661 case g := <-called:
662 if g.step != 2 || g.err != "boom" {
663 t.Fatalf("callback = (%d, %q); want (2, boom)", g.step, g.err)
664 }
665 case <-time.After(time.Second):
666 t.Fatal("OnTurnEnd did not fire for prompt-level turn_end with step")
667 }
668}
669
670func TestSetOnTurnEndReplaysPendingNoticesInArrivalOrder(t *testing.T) {
671 a := &Agent{}
672 notifyPromptTurnEnd(a, NewEvent("turn_end", map[string]any{"step": float64(1)}))
673 notifyPromptTurnEnd(a, NewEvent("turn_end", map[string]any{"step": float64(2), "error": "boom"}))
674
675 type notice struct {
676 step int
677 err string
678 }
679 var got []notice
680 a.SetOnTurnEnd(func(step int, errMsg string) {
681 got = append(got, notice{step: step, err: errMsg})
682 })
683
684 want := []notice{{step: 1}, {step: 2, err: "boom"}}
685 if len(got) != len(want) {
686 t.Fatalf("replayed %d notices, want %d: %+v", len(got), len(want), got)
687 }
688 for i := range want {
689 if got[i] != want[i] {
690 t.Errorf("replayed notice %d = %+v, want %+v", i, got[i], want[i])
691 }
692 }
693}
694
695func TestUpdateAgentFromEventDoesNotCountProviderTurnStarts(t *testing.T) {
696 a := &Agent{}
697 a.setTurnCounts(4, 2)
698 a.setTurnState(TurnRunning, "turn-1")
699 persisted := 0
700 a.persistFn = func(*Agent) error { persisted++; return nil }
701
702 updateAgentFromEvent(a, NewEvent(EventTurnStarted, map[string]any{"step": float64(7), "nested_turn": true}))
703
704 if got := a.LifetimeTurnsValue(); got != 4 {
705 t.Fatalf("lifetime turns = %d, want 4", got)
706 }
707 if got := a.CurrentRunTurnsValue(); got != 2 {
708 t.Fatalf("current-run turns = %d, want 2", got)
709 }
710 if got := a.TurnState(); got != TurnRunning {
711 t.Fatalf("turn state = %s, want %s", got, TurnRunning)
712 }
713 if persisted != 0 {
714 t.Fatalf("provider turn start persisted %d times, want 0", persisted)
715 }
716}
717
718func TestUpdateAgentFromEventPersistsCanonicalTurnCounters(t *testing.T) {
719 stateDir := t.TempDir()
720 a := &Agent{ID: "counter-agent", stateDir: stateDir}
721 a.persistFn = func(agent *Agent) error {
722 if err := writeAgentMeta(stateDir, agent); err != nil {
723 t.Errorf("write agent metadata: %v", err)
724 return err
725 }
726 return nil
727 }
728
729 updateAgentFromEvent(a, NewEvent(EventTurnStarted, map[string]any{
730 "turn_id": "turn-3",
731 "lifetime_turns": 3,
732 "current_run_turns": 1,
733 }))
734
735 meta, err := readAgentMeta(stateDir)
736 if err != nil {
737 t.Fatal(err)
738 }
739 if meta.LifetimeTurns != 3 || meta.CurrentRunTurns != 1 {
740 t.Fatalf("persisted counters = (%d, %d), want (3, 1)", meta.LifetimeTurns, meta.CurrentRunTurns)
741 }
742 if got := a.Snapshot(); got.LifetimeTurns != 3 || got.CurrentRunTurns != 1 {
743 t.Fatalf("snapshot counters = (%d, %d), want (3, 1)", got.LifetimeTurns, got.CurrentRunTurns)
744 }
745}
746
747func TestRecordPersistenceErrorPreservesFirstFailure(t *testing.T) {
748 a := &Agent{}
749 a.recordPersistenceError(fmt.Errorf("first failure"))
750 a.recordPersistenceError(fmt.Errorf("later failure"))
751 if got := a.Snapshot().Err; !strings.Contains(got, "first failure") || strings.Contains(got, "later failure") {
752 t.Fatalf("recorded persistence error = %q, want only first failure", got)
753 }
754}
755
756func TestRestoreResumePromptPreservesCommandIdentity(t *testing.T) {
757 a := &Agent{}
758 a.setResumePrompt("old prompt", time.Now())
759 oldID := a.resumePromptCommandID()
760 previous := a.setResumePrompt("new prompt", time.Now())
761 a.restoreResumePrompt(previous)
762 if got := a.resumePromptCommandID(); got != oldID {
763 t.Fatalf("restored command ID = %q, want %q", got, oldID)
764 }
765}
766
767func TestEventMatchesPendingResumeRejectsIdentityLessRejection(t *testing.T) {
768 a := &Agent{resumePromptID: "command-1"}
769 if eventMatchesPendingResume(a, Event{Type: "error", Data: map[string]any{"code": "turn_rejected"}}) {
770 t.Fatal("identity-less turn rejection matched a pending prompt")
771 }
772 if !eventMatchesPendingResume(a, Event{Type: EventTurnStarted, Data: map[string]any{}}) {
773 t.Fatal("identity-less legacy turn.started did not match a pending prompt")
774 }
775 if !eventMatchesPendingResume(a, Event{Type: "error", Data: map[string]any{"code": "turn_rejected", "command_id": "command-1"}}) {
776 t.Fatal("matching turn rejection did not match a pending prompt")
777 }
778}
779
780func TestEventCounterRejectsNegativeFractionalAndOverflowValues(t *testing.T) {
781 cases := []struct {
782 name string
783 value any
784 ok bool
785 }{
786 {name: "valid integer", value: float64(3), ok: true},
787 {name: "negative integer", value: int64(-1)},
788 {name: "fractional float", value: 1.5},
789 {name: "negative float", value: -1.0},
790 {name: "fractional json number", value: json.Number("1.5")},
791 {name: "overflow float", value: float64(^uint(0) >> 1)},
792 }
793 for _, tc := range cases {
794 t.Run(tc.name, func(t *testing.T) {
795 got, ok := eventCounter(map[string]any{"counter": tc.value}, "counter")
796 if ok != tc.ok {
797 t.Fatalf("eventCounter ok = %t, want %t (value %v, got %d)", ok, tc.ok, tc.value, got)
798 }
799 })
800 }
801}
802
803func indexOf(xs []string, want string) int {
804 for i, x := range xs {
805 if x == want {
806 return i
807 }
808 }
809 return -1
810}
811
812func safeAt(xs []string, i int) string {
813 if i < 0 || i >= len(xs) {
814 return ""
815 }
816 return xs[i]
817}