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