1package agent
2
3import (
4 "context"
5 "encoding/json"
6 "errors"
7 "fmt"
8 "net/http"
9 "net/http/httptest"
10 "os"
11 "path/filepath"
12 "reflect"
13 "strings"
14 "sync"
15 "sync/atomic"
16 "testing"
17 "time"
18
19 "github.com/bnema/zut/packages/agent/subagents"
20 "github.com/bnema/zut/packages/core"
21 "github.com/bnema/zut/packages/provider"
22)
23
24func TestResidentChildRunnerWithoutStepLimitRunsPastFormerCap(t *testing.T) {
25 server, requests := residentToolLoopServer(t, 51)
26 runner, _, _ := newResidentChildTestRunner(t, "unlimited-child", Args{
27 Provider: "openai", Model: "gpt-4o", APIKey: "synthetic", BaseURL: server.URL,
28 CWD: t.TempDir(), NoContextFiles: true, NoSkill: true, DisableRepetitionGuard: true,
29 })
30 if err := runner.Run(t.Context(), "review"); err != nil {
31 t.Fatalf("runner error = %v, want success past the former 50-step cap", err)
32 }
33 if got := requests.Load(); got != 52 {
34 t.Fatalf("provider requests = %d, want 51 tool turns and one answer", got)
35 }
36}
37
38func TestResidentChildRunnerEnforcesExplicitParentStepLimit(t *testing.T) {
39 server, requests := residentToolLoopServer(t, 200)
40 runner, _, _ := newResidentChildTestRunner(t, "limited-child", Args{
41 Provider: "openai", Model: "gpt-4o", APIKey: "synthetic", BaseURL: server.URL,
42 MaxSteps: 3, CWD: t.TempDir(), NoContextFiles: true, NoSkill: true,
43 })
44 err := runner.Run(t.Context(), "review")
45 if !errors.Is(err, core.ErrMaxSteps) || !strings.Contains(err.Error(), "3") {
46 t.Fatalf("runner error = %v, want step limit 3", err)
47 }
48 if got := requests.Load(); got != 3 {
49 t.Fatalf("provider requests = %d, want 3", got)
50 }
51}
52
53// residentToolLoopServer answers the first toolCalls requests with a distinct
54// read call and then returns a substantive final answer.
55func residentToolLoopServer(t *testing.T, toolCalls int) (*httptest.Server, *atomic.Int32) {
56 t.Helper()
57 var requests atomic.Int32
58 server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
59 number := requests.Add(1)
60 w.Header().Set("content-type", "text/event-stream")
61 if int(number) <= toolCalls {
62 writeOpenAIChunk(t, w, map[string]any{
63 "choices": []any{map[string]any{
64 "index": 0,
65 "delta": map[string]any{"tool_calls": []any{map[string]any{
66 "index": 0, "id": fmt.Sprintf("call-%d", number), "type": "function",
67 "function": map[string]string{"name": "read", "arguments": `{"path":"x"}`},
68 }}},
69 "finish_reason": "tool_calls",
70 }},
71 "usage": map[string]int{"prompt_tokens": 10, "completion_tokens": 1},
72 })
73 return
74 }
75 writeOpenAIChunk(t, w, map[string]any{
76 "choices": []any{map[string]any{"index": 0, "delta": map[string]string{"content": "final result"}, "finish_reason": "stop"}},
77 "usage": map[string]int{"prompt_tokens": 10, "completion_tokens": 1},
78 })
79 }))
80 t.Cleanup(server.Close)
81 return server, &requests
82}
83
84// newResidentChildTestRunner builds one journaled resident child runner with
85// its own accepted spec.
86func newResidentChildTestRunner(t *testing.T, childID string, args Args) (subagents.ResidentRuntime, *subagents.ResidentJournal, subagents.ResidentChildSpec) {
87 t.Helper()
88 spec := subagents.ResidentChildSpec{
89 ID: childID, SessionID: childID + "-session", Provider: "openai", Model: "gpt-4o",
90 Tools: []string{"read"},
91 }
92 journal, err := subagents.OpenResidentJournal(t.TempDir(), spec.ID)
93 if err != nil {
94 t.Fatal(err)
95 }
96 t.Cleanup(func() { _ = journal.Close() })
97 if err := journal.Accept(spec, "review"); err != nil {
98 t.Fatal(err)
99 }
100 runtime, err := newResidentChildRunner(args, spec, journal)
101 if err != nil {
102 t.Fatal(err)
103 }
104 return runtime, journal, spec
105}
106
107func TestResidentChildRunnerInjectsContextReminderIntoNextRequest(t *testing.T) {
108 var firstUsage atomic.Int64
109 var captured []string
110 var mu sync.Mutex
111 var requests atomic.Int32
112 server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
113 number := requests.Add(1)
114 var body map[string]any
115 if err := json.NewDecoder(r.Body).Decode(&body); err != nil {
116 t.Errorf("decode request: %v", err)
117 }
118 encoded, _ := json.Marshal(body["messages"])
119 mu.Lock()
120 captured = append(captured, string(encoded))
121 mu.Unlock()
122 w.Header().Set("content-type", "text/event-stream")
123 usage := map[string]int{"prompt_tokens": 10, "completion_tokens": 1}
124 text := "second result"
125 if number == 1 {
126 text = "first result"
127 usage = map[string]int{"prompt_tokens": int(firstUsage.Load()), "completion_tokens": 1}
128 }
129 writeOpenAIChunk(t, w, map[string]any{
130 "choices": []any{map[string]any{"index": 0, "delta": map[string]string{"content": text}, "finish_reason": "stop"}},
131 "usage": usage,
132 })
133 }))
134 defer server.Close()
135
136 args := Args{Provider: "openai", Model: "gpt-4o", APIKey: "synthetic", BaseURL: server.URL, CWD: t.TempDir(), NoContextFiles: true, NoSkill: true}
137 resolved, err := Resolve(args, true)
138 if err != nil {
139 t.Fatal(err)
140 }
141 if resolved.ContextWindow <= 0 {
142 t.Fatalf("resolved context window = %d, want positive", resolved.ContextWindow)
143 }
144 // Cross the 85% threshold on the first turn.
145 firstUsage.Store(int64(resolved.ContextWindow * 86 / 100))
146 spec := subagents.ResidentChildSpec{
147 ID: "nudge-child", SessionID: "nudge-session", Provider: "openai", Model: "gpt-4o",
148 Tools: []string{"read"},
149 }
150 journal, err := subagents.OpenResidentJournal(t.TempDir(), spec.ID)
151 if err != nil {
152 t.Fatal(err)
153 }
154 t.Cleanup(func() { _ = journal.Close() })
155 if err := journal.Accept(spec, "review"); err != nil {
156 t.Fatal(err)
157 }
158 runner, err := newResidentChildRunner(args, spec, journal)
159 if err != nil {
160 t.Fatal(err)
161 }
162 if err := runner.Run(t.Context(), "review"); err != nil {
163 t.Fatal(err)
164 }
165 if err := runner.Run(t.Context(), "follow up"); err != nil {
166 t.Fatal(err)
167 }
168 if got := requests.Load(); got != 2 {
169 t.Fatalf("provider requests = %d, want 2", got)
170 }
171 mu.Lock()
172 defer mu.Unlock()
173 if len(captured) != 2 {
174 t.Fatalf("captured requests = %d, want 2", len(captured))
175 }
176 if strings.Contains(captured[0], "Context usage is at") {
177 t.Fatalf("first request carries a reminder before any usage: %s", captured[0])
178 }
179 want := fmt.Sprintf("Context usage is at %d%%", firstUsage.Load()*100/int64(resolved.ContextWindow))
180 if !strings.Contains(captured[1], want) {
181 t.Fatalf("second request missing %q: %s", want, captured[1])
182 }
183 // The reminder travels as developer context in the live request; the
184 // durable transcript keeps only finalized history.
185 messages, err := subagents.ReadResidentTranscriptMessages(journal.Dir())
186 if err != nil {
187 t.Fatal(err)
188 }
189 var texts []string
190 for _, message := range messages {
191 for _, content := range message.Content {
192 if block, ok := content.(provider.TextBlock); ok {
193 texts = append(texts, block.Text)
194 }
195 }
196 }
197 joined := strings.Join(texts, "\n")
198 for _, want := range []string{"first result", "second result"} {
199 if !strings.Contains(joined, want) {
200 t.Fatalf("journal history missing %q: %q", want, joined)
201 }
202 }
203 if strings.Contains(joined, "Context usage is at") {
204 t.Fatalf("journal transcript carries the reminder: %q", joined)
205 }
206}
207
208func TestResidentChildRegistryUsesExactToolListAndForbidsDelegation(t *testing.T) {
209 catalogue := core.Registry{
210 "read": nil,
211 "bash": nil,
212 "subagent": nil,
213 "update_goal": nil,
214 "plan": nil,
215 }
216 registry, err := residentChildRegistry(catalogue, []string{"read"})
217 if err != nil {
218 t.Fatal(err)
219 }
220 if len(registry) != 1 {
221 t.Fatalf("registry = %#v", registry)
222 }
223 for _, name := range []string{"subagent", "update_goal", "plan", "missing"} {
224 _, err := residentChildRegistry(catalogue, []string{name})
225 if err == nil || !strings.Contains(err.Error(), name) {
226 t.Fatalf("residentChildRegistry(%q) error = %v", name, err)
227 }
228 }
229}
230
231func TestConfigureResidentContextNudgeRemindsOncePerBand(t *testing.T) {
232 agent := core.NewAgent(nil, "model", "system", core.Registry{"read": nil})
233 configureResidentContextNudge(agent, 1_000)
234
235 check := func(usage provider.Usage, wantReminder bool) string {
236 t.Helper()
237 agent.SeedLastTurnUsage(usage)
238 allowed, reason, contextText := agent.BeforeTurnContext(t.Context(), 1)
239 if !allowed || reason != "" {
240 t.Fatalf("nudge blocked turn: allowed=%t reason=%q", allowed, reason)
241 }
242 if wantReminder && !strings.Contains(contextText, "Context usage is at ") {
243 t.Fatalf("missing reminder for usage %#v: %q", usage, contextText)
244 }
245 if !wantReminder && contextText != "" {
246 t.Fatalf("unexpected reminder for usage %#v: %q", usage, contextText)
247 }
248 return contextText
249 }
250
251 // Below the first band: silent, including cache-heavy usage.
252 check(provider.Usage{InputTokens: 849}, false)
253 check(provider.Usage{InputTokens: 400, CacheReadTokens: 400, CacheWriteTokens: 49}, false)
254 // Each band fires once, carrying the computed percent of the crossing turn.
255 first := check(provider.Usage{InputTokens: 850}, true)
256 if !strings.Contains(first, "85%") {
257 t.Fatalf("reminder percent = %q, want 85%%", first)
258 }
259 check(provider.Usage{InputTokens: 860}, false)
260 second := check(provider.Usage{InputTokens: 900}, true)
261 if !strings.Contains(second, "90%") {
262 t.Fatalf("reminder percent = %q, want 90%%", second)
263 }
264 check(provider.Usage{InputTokens: 920}, false)
265 third := check(provider.Usage{InputTokens: 950}, true)
266 if !strings.Contains(third, "95%") {
267 t.Fatalf("reminder percent = %q, want 95%%", third)
268 }
269 // Past the last band the ladder is spent.
270 check(provider.Usage{InputTokens: 999}, false)
271}
272
273func TestConfigureResidentContextNudgeIgnoresShrinkingUsage(t *testing.T) {
274 agent := core.NewAgent(nil, "model", "system", core.Registry{"read": nil})
275 configureResidentContextNudge(agent, 1_000)
276
277 check := func(usage provider.Usage, wantReminder bool) string {
278 t.Helper()
279 agent.SeedLastTurnUsage(usage)
280 allowed, reason, contextText := agent.BeforeTurnContext(t.Context(), 1)
281 if !allowed || reason != "" {
282 t.Fatalf("nudge blocked turn: allowed=%t reason=%q", allowed, reason)
283 }
284 if wantReminder != (contextText != "") {
285 t.Fatalf("reminder for usage %#v = %q, want reminder=%t", usage, contextText, wantReminder)
286 }
287 return contextText
288 }
289
290 // A shrinking prompt never delivers a band it did not cross.
291 check(provider.Usage{InputTokens: 700}, false)
292 check(provider.Usage{InputTokens: 500}, false)
293 jumped := check(provider.Usage{InputTokens: 910}, true)
294 if !strings.Contains(jumped, "91%") {
295 t.Fatalf("reminder percent = %q, want 91%%", jumped)
296 }
297}
298
299func TestConfigureResidentContextNudgeHandlesDirectJumpAndUnknownWindow(t *testing.T) {
300 agent := core.NewAgent(nil, "model", "system", core.Registry{"read": nil})
301 configureResidentContextNudge(agent, 1_000)
302 agent.SeedLastTurnUsage(provider.Usage{InputTokens: 970})
303 allowed, _, contextText := agent.BeforeTurnContext(t.Context(), 1)
304 if !allowed || !strings.Contains(contextText, "97%") {
305 t.Fatalf("jump reminder = allowed=%t context=%q, want one 97%% reminder", allowed, contextText)
306 }
307
308 unknown := core.NewAgent(nil, "model", "system", core.Registry{"read": nil})
309 configureResidentContextNudge(unknown, 0)
310 unknown.SeedLastTurnUsage(provider.Usage{InputTokens: 970})
311 if allowed, reason, contextText := unknown.BeforeTurnContext(t.Context(), 1); !allowed || reason != "" || contextText != "" {
312 t.Fatalf("unknown window context = allowed=%t reason=%q context=%q, want silent", allowed, reason, contextText)
313 }
314}
315
316func TestResidentChildRunnerKeepsToolsAcrossTurns(t *testing.T) {
317 var requests atomic.Int32
318 server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
319 requestNumber := requests.Add(1)
320 var body map[string]any
321 if err := json.NewDecoder(r.Body).Decode(&body); err != nil {
322 t.Errorf("decode request: %v", err)
323 }
324 w.Header().Set("content-type", "text/event-stream")
325 if requestNumber == 1 {
326 if got := requestMaxTokens(t, body); got != 16_384 {
327 t.Errorf("initial max_tokens = %d, want 16384", got)
328 }
329 writeOpenAIChunk(t, w, map[string]any{
330 "choices": []any{map[string]any{
331 "index": 0,
332 "delta": map[string]any{"tool_calls": []any{map[string]any{
333 "index": 0, "id": "call-1", "type": "function",
334 "function": map[string]string{"name": "read", "arguments": `{"path":"x"}`},
335 }}},
336 "finish_reason": "tool_calls",
337 }},
338 "usage": map[string]int{"prompt_tokens": 900, "completion_tokens": 0},
339 })
340 return
341 }
342 if tools, ok := body["tools"].([]any); !ok || len(tools) == 0 {
343 t.Errorf("near-limit request tools = %#v, want registered tools", body["tools"])
344 }
345 if got := requestMaxTokens(t, body); got <= 100 {
346 t.Errorf("near-limit max_tokens = %d, want model output limit", got)
347 }
348 writeOpenAIChunk(t, w, map[string]any{
349 "choices": []any{map[string]any{"index": 0, "delta": map[string]string{"content": "final result"}, "finish_reason": "stop"}},
350 "usage": map[string]int{"prompt_tokens": 0, "completion_tokens": 50},
351 })
352 }))
353 defer server.Close()
354
355 runner, journal := newResidentTestRunner(t, server.URL)
356 if err := runner.Run(t.Context(), "review"); err != nil {
357 t.Fatal(err)
358 }
359 if got := requests.Load(); got != 2 {
360 t.Fatalf("provider requests = %d, want 2", got)
361 }
362 if summary := latestResidentAssistantText(t, journal); summary != "final result" {
363 t.Fatalf("latest summary = %q", summary)
364 }
365}
366
367func TestResidentChildRunnerRunsHighUsageFollowUpWithoutBudget(t *testing.T) {
368 var requests atomic.Int32
369 server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
370 requests.Add(1)
371 var body map[string]any
372 if err := json.NewDecoder(r.Body).Decode(&body); err != nil {
373 t.Errorf("decode request: %v", err)
374 }
375 if got := requestMaxTokens(t, body); got != 16_384 {
376 t.Errorf("initial max_tokens = %d, want 16384", got)
377 }
378 w.Header().Set("content-type", "text/event-stream")
379 writeOpenAIChunk(t, w, map[string]any{
380 "choices": []any{map[string]any{"index": 0, "delta": map[string]string{"content": "best available result"}, "finish_reason": "stop"}},
381 "usage": map[string]int{"prompt_tokens": 1_200, "completion_tokens": 0},
382 })
383 }))
384 defer server.Close()
385
386 runner, journal := newResidentTestRunner(t, server.URL)
387 if err := runner.Run(t.Context(), "review"); err != nil {
388 t.Fatalf("runner error = %v, want terminal result", err)
389 }
390 if got := requests.Load(); got != 1 {
391 t.Fatalf("provider requests = %d, want 1", got)
392 }
393 if summary := latestResidentAssistantText(t, journal); summary != "best available result" {
394 t.Fatalf("latest summary = %q", summary)
395 }
396 // Without a cumulative budget, a high-usage follow-up runs normally.
397 if err := runner.Run(t.Context(), "continue"); err != nil {
398 t.Fatalf("follow-up error = %v, want success", err)
399 }
400 if got := requests.Load(); got != 2 {
401 t.Fatalf("provider requests = %d, want 2", got)
402 }
403}
404
405func TestResidentChildRunnerResumesWithRetainedHistory(t *testing.T) {
406 var requests atomic.Int32
407 server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
408 request := requests.Add(1)
409 w.Header().Set("content-type", "text/event-stream")
410 if request == 1 {
411 writeOpenAIChunk(t, w, map[string]any{
412 "choices": []any{map[string]any{"index": 0, "delta": map[string]string{"content": "first result"}, "finish_reason": "stop"}},
413 "usage": map[string]int{"prompt_tokens": 1200, "completion_tokens": 0},
414 })
415 return
416 }
417 var body map[string]any
418 if err := json.NewDecoder(r.Body).Decode(&body); err != nil {
419 t.Error(err)
420 }
421 encoded, _ := json.Marshal(body["messages"])
422 if !strings.Contains(string(encoded), "first result") {
423 t.Errorf("resume lost prior messages: %s", encoded)
424 }
425 writeOpenAIChunk(t, w, map[string]any{
426 "choices": []any{map[string]any{"index": 0, "delta": map[string]string{"content": "recovered result"}, "finish_reason": "stop"}},
427 "usage": map[string]int{"prompt_tokens": 20, "completion_tokens": 10},
428 })
429 }))
430 defer server.Close()
431 cwd := t.TempDir()
432 manager := subagents.NewResidentManager(t.TempDir(), func(spec subagents.ResidentChildSpec, journal *subagents.ResidentJournal) (subagents.ResidentRuntime, error) {
433 return newResidentChildRunner(Args{Provider: "openai", Model: "gpt-4o", APIKey: "synthetic", BaseURL: server.URL, CWD: cwd, NoContextFiles: true, NoSkill: true}, spec, journal)
434 })
435 t.Cleanup(func() { _ = manager.Close(context.Background()) })
436 completed := make(chan subagents.ResidentCompletion, 2)
437 manager.SetCompletionObserver(func(c subagents.ResidentCompletion) bool { completed <- c; return true })
438 spec := subagents.ResidentChildSpec{ID: "resume-child", SessionID: "session", Provider: "openai", Model: "gpt-4o"}
439 if _, err := manager.Spawn(t.Context(), spec, "review"); err != nil {
440 t.Fatal(err)
441 }
442 for turn := 0; turn < 2; turn++ {
443 select {
444 case completion := <-completed:
445 if completion.Err != nil {
446 t.Fatal(completion.Err)
447 }
448 case <-time.After(5 * time.Second):
449 t.Fatal("no completion")
450 }
451 if turn == 0 {
452 if err := manager.Resume(t.Context(), spec.ID, "follow up using previous results"); err != nil {
453 t.Fatal(err)
454 }
455 }
456 }
457 snapshot, _ := manager.SnapshotFor(spec.ID)
458 if requests.Load() != 2 || snapshot.Usage.InputTokens != 1220 {
459 t.Fatalf("requests=%d, snapshot=%#v", requests.Load(), snapshot)
460 }
461}
462
463func newResidentTestRunner(t *testing.T, baseURL string) (subagents.ResidentRuntime, *subagents.ResidentJournal) {
464 t.Helper()
465 runner, journal, _ := newResidentChildTestRunner(t, "resident-child", Args{
466 Provider: "openai", Model: "gpt-4o", APIKey: "synthetic", BaseURL: baseURL,
467 CWD: t.TempDir(), NoContextFiles: true, NoSkill: true,
468 })
469 return runner, journal
470}
471
472func requestMaxTokens(t *testing.T, body map[string]any) int {
473 t.Helper()
474 value, ok := body["max_tokens"].(float64)
475 if !ok {
476 t.Errorf("request max_tokens = %#v", body["max_tokens"])
477 return 0
478 }
479 return int(value)
480}
481
482func latestResidentAssistantText(t *testing.T, journal *subagents.ResidentJournal) string {
483 t.Helper()
484 messages, err := subagents.ReadResidentTranscriptMessages(journal.Dir())
485 if err != nil {
486 t.Fatal(err)
487 }
488 for index := len(messages) - 1; index >= 0; index-- {
489 if messages[index].Role != provider.RoleAssistant {
490 continue
491 }
492 var text strings.Builder
493 for _, content := range messages[index].Content {
494 if block, ok := content.(provider.TextBlock); ok {
495 text.WriteString(block.Text)
496 }
497 }
498 return text.String()
499 }
500 return ""
501}
502
503func writeOpenAIChunk(t *testing.T, w http.ResponseWriter, chunk map[string]any) {
504 t.Helper()
505 data, err := json.Marshal(chunk)
506 if err != nil {
507 t.Fatal(err)
508 }
509 _, _ = fmt.Fprintf(w, "data: %s\n\ndata: [DONE]\n\n", data)
510}
511
512func TestResidentChildArgsPreserveDurableProfileInheritance(t *testing.T) {
513 no := false
514 next := residentChildArgs(Args{Orchestrate: true, NoSkill: false, NoContextFiles: false, BaseURL: "https://stale.example/v1", InsecureTLS: false}, "openai", subagents.ResidentChildSpec{
515 Provider: "openai", BaseURL: "https://current.example/v1", InsecureTLS: true, Model: "gpt-5", Workspace: "/repo/child",
516 InheritSkills: &no, InheritProjectContext: &no,
517 })
518 if next.Provider != "openai" || next.BaseURL != "https://current.example/v1" || !next.InsecureTLS || next.Model != "gpt-5" || next.CWD != "/repo/child" || !next.NoSkill || !next.NoContextFiles || next.Orchestrate || !next.ResidentChild {
519 t.Fatalf("resident child args = %#v", next)
520 }
521}
522
523func TestResidentChildArgsDoNotForwardCLIKeyAcrossProviders(t *testing.T) {
524 parent := Args{Provider: "openai", APIKey: "parent-key"}
525 child := subagents.ResidentChildSpec{Provider: "anthropic", Model: "claude"}
526 if next := residentChildArgs(parent, "openai", child); next.APIKey != "" {
527 t.Fatalf("cross-provider child inherited API key %q", next.APIKey)
528 }
529 child.Provider = "openai"
530 if next := residentChildArgs(parent, "openai", child); next.APIKey != "parent-key" {
531 t.Fatalf("same-provider child API key = %q, want parent key", next.APIKey)
532 }
533}
534
535// residentOverflowScript scripts the provider responses of a child that hits
536// the context window: leading tool-call turns, one overflow, one compaction
537// summary, then continued tool turns or the final text.
538type residentOverflowScript struct {
539 toolCallRequests int
540 continuationToolCalls int
541 secondOverflow bool
542 finalText string
543 onOverflow func()
544 // toolUsageTokens is the prompt size reported by tool-call responses; zero
545 // reports a small prompt.
546 toolUsageTokens int
547 // summaryUsageTokens, when positive, is the prompt size reported by the
548 // compaction summary response.
549 summaryUsageTokens int
550 // continuationUsageTokens is the prompt size reported by tool turns after a
551 // successful compaction; zero reuses toolUsageTokens.
552 continuationUsageTokens int
553
554 requests atomic.Int32
555 mu sync.Mutex
556 bodies []string
557}
558
559func (s *residentOverflowScript) start(t *testing.T) *httptest.Server {
560 t.Helper()
561 server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
562 number := int(s.requests.Add(1))
563 var body map[string]any
564 if err := json.NewDecoder(r.Body).Decode(&body); err != nil {
565 t.Errorf("decode request: %v", err)
566 }
567 encoded, _ := json.Marshal(body["messages"])
568 s.mu.Lock()
569 s.bodies = append(s.bodies, string(encoded))
570 s.mu.Unlock()
571
572 overflowAt := s.toolCallRequests + 1
573 switch {
574 case number <= s.toolCallRequests:
575 w.Header().Set("content-type", "text/event-stream")
576 writeOpenAIChunk(t, w, residentToolCallChunk(number, s.toolUsageTokens))
577 case number == overflowAt:
578 if s.onOverflow != nil {
579 s.onOverflow()
580 }
581 w.WriteHeader(http.StatusRequestEntityTooLarge)
582 _, _ = w.Write([]byte(`{"error":{"message":"input exceeds the context window","code":"context_length_exceeded"}}`))
583 case number == overflowAt+1:
584 w.Header().Set("content-type", "text/event-stream")
585 writeOpenAIChunk(t, w, residentSummaryChunk(s.summaryUsageTokens))
586 case s.secondOverflow:
587 w.WriteHeader(http.StatusRequestEntityTooLarge)
588 _, _ = w.Write([]byte(`{"error":{"message":"input exceeds the context window","code":"context_length_exceeded"}}`))
589 case number < overflowAt+2+s.continuationToolCalls:
590 w.Header().Set("content-type", "text/event-stream")
591 continuationUsage := s.continuationUsageTokens
592 if continuationUsage <= 0 {
593 continuationUsage = s.toolUsageTokens
594 }
595 writeOpenAIChunk(t, w, residentToolCallChunk(number, continuationUsage))
596 default:
597 w.Header().Set("content-type", "text/event-stream")
598 writeOpenAIChunk(t, w, residentFinalChunk(s.finalText))
599 }
600 }))
601 t.Cleanup(server.Close)
602 return server
603}
604
605func residentToolCallChunk(number, usageTokens int) map[string]any {
606 if usageTokens <= 0 {
607 usageTokens = 10
608 }
609 return map[string]any{
610 "choices": []any{map[string]any{
611 "index": 0,
612 "delta": map[string]any{"tool_calls": []any{map[string]any{
613 "index": 0, "id": fmt.Sprintf("call-%d", number), "type": "function",
614 "function": map[string]string{"name": "read", "arguments": `{"path":"x"}`},
615 }}},
616 "finish_reason": "tool_calls",
617 }},
618 "usage": map[string]int{"prompt_tokens": usageTokens, "completion_tokens": 1},
619 }
620}
621
622func residentSummaryChunk(usageTokens int) map[string]any {
623 chunk := map[string]any{
624 "choices": []any{map[string]any{"index": 0, "delta": map[string]string{"content": "compacted digest"}, "finish_reason": "stop"}},
625 }
626 if usageTokens > 0 {
627 chunk["usage"] = map[string]int{"prompt_tokens": usageTokens, "completion_tokens": 2}
628 }
629 return chunk
630}
631
632func residentFinalChunk(text string) map[string]any {
633 return map[string]any{
634 "choices": []any{map[string]any{"index": 0, "delta": map[string]string{"content": text}, "finish_reason": "stop"}},
635 "usage": map[string]int{"prompt_tokens": 12, "completion_tokens": 5},
636 }
637}
638
639func (s *residentOverflowScript) capturedBodies() []string {
640 s.mu.Lock()
641 defer s.mu.Unlock()
642 return append([]string(nil), s.bodies...)
643}
644
645func residentArguments(t *testing.T, baseURL string) Args {
646 t.Helper()
647 return Args{
648 Provider: "openai", Model: "gpt-4o", APIKey: "synthetic", BaseURL: baseURL,
649 CWD: t.TempDir(), NoContextFiles: true, NoSkill: true,
650 }
651}
652
653func countResidentCheckpoints(t *testing.T, journal *subagents.ResidentJournal) int {
654 t.Helper()
655 records, err := subagents.ReadResidentJournal(filepath.Join(journal.Dir(), "transcript.jsonl"))
656 if err != nil {
657 t.Fatal(err)
658 }
659 checkpoints := 0
660 for _, record := range records {
661 if record.Type == "child.compacted" {
662 checkpoints++
663 }
664 }
665 return checkpoints
666}
667
668func TestResidentChildRunnerRecoversFromContextOverflowOnce(t *testing.T) {
669 script := &residentOverflowScript{toolCallRequests: 2, finalText: "overflow recovered result"}
670 server := script.start(t)
671 runner, journal, spec := newResidentChildTestRunner(t, "overflow-child", residentArguments(t, server.URL))
672 if err := journal.RecordTurnStarted(spec, "turn-1"); err != nil {
673 t.Fatal(err)
674 }
675 runErr := runner.Run(t.Context(), "review")
676 if err := journal.RecordTurnFinished(spec, "turn-1", runErr); err != nil {
677 t.Fatal(err)
678 }
679 if runErr != nil {
680 t.Fatalf("runner error = %v, want one recovered turn", runErr)
681 }
682 if got := script.requests.Load(); got != 5 {
683 t.Fatalf("provider requests = %d, want 3 prompt, 1 compaction, and 1 continuation", got)
684 }
685 result, err := journal.Result()
686 if err != nil {
687 t.Fatal(err)
688 }
689 if result.State != subagents.ResidentIdle || result.Summary != "overflow recovered result" {
690 t.Fatalf("result = %#v, want the recovered final text", result)
691 }
692 records, err := subagents.ReadResidentJournal(filepath.Join(journal.Dir(), "transcript.jsonl"))
693 if err != nil {
694 t.Fatal(err)
695 }
696 checkpoints := 0
697 for _, record := range records {
698 if record.Type != "child.compacted" {
699 continue
700 }
701 checkpoints++
702 if record.TurnID != "turn-1" {
703 t.Fatalf("checkpoint turn ID = %q, want the active accepted turn", record.TurnID)
704 }
705 }
706 if checkpoints != 1 {
707 t.Fatalf("compaction checkpoints = %d, want 1", checkpoints)
708 }
709 // Resume replay starts from the checkpoint: the digest replaces the
710 // pre-overflow tool exchange and the continuation request carried it.
711 messages, err := subagents.ReadResidentTranscriptMessages(journal.Dir())
712 if err != nil {
713 t.Fatal(err)
714 }
715 if len(messages) != 3 || !strings.HasPrefix(residentMessageText(messages[0]), "## Context Summary (compacted)") {
716 t.Fatalf("resumed transcript = %#v, want the checkpoint plus the kept tail", messages)
717 }
718 bodies := script.capturedBodies()
719 if len(bodies) != 5 || !strings.Contains(bodies[4], "compacted digest") {
720 t.Fatalf("continuation request = %q, want the compacted transcript", bodies[len(bodies)-1])
721 }
722 if strings.Contains(bodies[4], "call-1") {
723 t.Fatalf("continuation request kept the dropped tool exchange: %q", bodies[4])
724 }
725}
726
727func TestResidentChildRunnerDoesNotRecoverTwicePerTurn(t *testing.T) {
728 script := &residentOverflowScript{toolCallRequests: 2, secondOverflow: true, finalText: "unreached"}
729 server := script.start(t)
730 runner, journal, _ := newResidentChildTestRunner(t, "double-overflow", residentArguments(t, server.URL))
731 err := runner.Run(t.Context(), "review")
732 if err == nil || !provider.IsContextOverflowError(err) {
733 t.Fatalf("runner error = %v, want the second overflow to stay terminal", err)
734 }
735 if got := script.requests.Load(); got != 5 {
736 t.Fatalf("provider requests = %d, want no second compaction", got)
737 }
738 if checkpoints := countResidentCheckpoints(t, journal); checkpoints != 1 {
739 t.Fatalf("compaction checkpoints = %d, want 1", checkpoints)
740 }
741}
742
743func TestResidentChildRunnerDoesNotReplenishExplicitStepLimit(t *testing.T) {
744 script := &residentOverflowScript{toolCallRequests: 3, finalText: "unreached"}
745 server := script.start(t)
746 args := residentArguments(t, server.URL)
747 args.MaxSteps = 4
748 runner, journal, _ := newResidentChildTestRunner(t, "limited-overflow", args)
749 err := runner.Run(t.Context(), "review")
750 if err == nil || !provider.IsContextOverflowError(err) {
751 t.Fatalf("runner error = %v, want the overflow", err)
752 }
753 if got := script.requests.Load(); got != 5 {
754 t.Fatalf("provider requests = %d, want the compaction attempt only", got)
755 }
756 // The checkpoint is still journaled so an explicit resume starts small,
757 // but the spent allowance forbids continuing this turn.
758 if checkpoints := countResidentCheckpoints(t, journal); checkpoints != 1 {
759 t.Fatalf("compaction checkpoints = %d, want 1", checkpoints)
760 }
761 for _, body := range script.capturedBodies() {
762 if strings.Contains(body, "compacted digest") {
763 t.Fatalf("recovery continued after the step allowance was spent: %q", body)
764 }
765 }
766}
767
768func TestResidentChildRunnerFailsTurnWhenCheckpointCannotPersist(t *testing.T) {
769 root := t.TempDir()
770 spec := subagents.ResidentChildSpec{
771 ID: "persist-failure", SessionID: "persist-failure-session", Provider: "openai", Model: "gpt-4o",
772 Tools: []string{"read"},
773 }
774 journal, err := subagents.OpenResidentJournal(root, spec.ID)
775 if err != nil {
776 t.Fatal(err)
777 }
778 t.Cleanup(func() { _ = journal.Close() })
779 if err := journal.Accept(spec, "review"); err != nil {
780 t.Fatal(err)
781 }
782 // The journal is closed at the exact overflow boundary, so the recovery
783 // cannot persist its checkpoint.
784 script := &residentOverflowScript{
785 toolCallRequests: 2,
786 finalText: "unreached",
787 onOverflow: func() { _ = journal.Close() },
788 }
789 server := script.start(t)
790 runner, err := newResidentChildRunner(residentArguments(t, server.URL), spec, journal)
791 if err != nil {
792 t.Fatal(err)
793 }
794 if err := runner.Run(t.Context(), "review"); err == nil {
795 t.Fatal("runner recovered without persisting its checkpoint")
796 }
797 if got := script.requests.Load(); got != 4 {
798 t.Fatalf("provider requests = %d, want the compaction attempt only", got)
799 }
800 data, readErr := os.ReadFile(filepath.Join(journal.Dir(), "transcript.jsonl"))
801 if readErr != nil {
802 t.Fatal(readErr)
803 }
804 if strings.Contains(string(data), `"type":"child.compacted"`) {
805 t.Fatal("a failed recovery recorded a checkpoint")
806 }
807
808 // A later reconstructed child resumes the pre-compaction transcript.
809 reopened, err := subagents.OpenResidentJournal(root, spec.ID)
810 if err != nil {
811 t.Fatal(err)
812 }
813 t.Cleanup(func() { _ = reopened.Close() })
814 runner, err = newResidentChildRunner(residentArguments(t, server.URL), spec, reopened)
815 if err != nil {
816 t.Fatal(err)
817 }
818 if err := runner.Run(t.Context(), "verify"); err != nil {
819 t.Fatalf("follow-up error = %v", err)
820 }
821 bodies := script.capturedBodies()
822 if last := bodies[len(bodies)-1]; !strings.Contains(last, "call-1") {
823 t.Fatalf("follow-up request = %q, want the pre-compaction transcript", last)
824 }
825}
826
827func TestRecoverResidentContextOverflowRestoresTranscript(t *testing.T) {
828 server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
829 w.Header().Set("content-type", "text/event-stream")
830 writeOpenAIChunk(t, w, map[string]any{
831 "choices": []any{map[string]any{"index": 0, "delta": map[string]string{"content": "compacted digest"}, "finish_reason": "stop"}},
832 "usage": map[string]int{"prompt_tokens": 5, "completion_tokens": 2},
833 })
834 }))
835 defer server.Close()
836 resolved, err := Resolve(residentArguments(t, server.URL), true)
837 if err != nil {
838 t.Fatal(err)
839 }
840 agent := core.NewAgent(resolved.NewClient(), resolved.Model, "", resolved.ToolRegistry)
841 before := []provider.Message{
842 {Role: provider.RoleUser, Content: []provider.Content{provider.TextBlock{Text: "task"}}},
843 {Role: provider.RoleAssistant, Content: []provider.Content{provider.ToolCallBlock{ID: "call-1", Name: "read", Arguments: json.RawMessage(`{"path":"x"}`)}}},
844 {Role: provider.RoleTool, Content: []provider.Content{provider.ToolResultBlock{CallID: "call-1", Content: []provider.Content{provider.TextBlock{Text: "missing"}}}}},
845 {Role: provider.RoleUser, Content: []provider.Content{provider.TextBlock{Text: "keep working"}}},
846 }
847 agent.SetMessages(before)
848
849 journal, err := subagents.OpenResidentJournal(t.TempDir(), "restore-child")
850 if err != nil {
851 t.Fatal(err)
852 }
853 if err := journal.Accept(subagents.ResidentChildSpec{ID: "restore-child", SessionID: "restore-session", Provider: "openai", Model: "gpt-4o"}, "task"); err != nil {
854 t.Fatal(err)
855 }
856 if err := journal.Close(); err != nil {
857 t.Fatal(err)
858 }
859
860 overflow := errors.New("openai: http 413: context_length_exceeded")
861 err = recoverResidentContextOverflow(t.Context(), agent, journal, func(core.AgentEvent) {}, 0, 4, overflow)
862 if !errors.Is(err, overflow) || !strings.Contains(err.Error(), "persist resident child compaction") {
863 t.Fatalf("recovery error = %v, want a checkpoint persistence failure", err)
864 }
865 if !reflect.DeepEqual(agent.Messages(), before) {
866 t.Fatalf("transcript = %#v, want the pre-compaction transcript restored", agent.Messages())
867 }
868}
869
870func residentMessageText(message provider.Message) string {
871 var text strings.Builder
872 for _, content := range message.Content {
873 if block, ok := content.(provider.TextBlock); ok {
874 text.WriteString(block.Text)
875 }
876 }
877 return text.String()
878}
879
880func TestResidentChildRunnerIgnoresCompactionUsageInReminder(t *testing.T) {
881 // The summarization request reports the pre-compaction prompt size. If that
882 // gauge armed the next reminder, the first continuation step would claim the
883 // compacted transcript sits at the top of the window. The continuation's own
884 // usage must still be able to arm a truthful reminder.
885 script := &residentOverflowScript{
886 toolCallRequests: 2,
887 continuationToolCalls: 1,
888 finalText: "overflow recovered result",
889 summaryUsageTokens: 1 << 20,
890 continuationUsageTokens: 1 << 20,
891 }
892 server := script.start(t)
893 runner, _, _ := newResidentChildTestRunner(t, "usage-child", residentArguments(t, server.URL))
894 if err := runner.Run(t.Context(), "review"); err != nil {
895 t.Fatalf("runner error = %v, want one recovered turn", err)
896 }
897 bodies := script.capturedBodies()
898 if len(bodies) != 6 {
899 t.Fatalf("provider requests = %d, want 3 prompt, 1 compaction, and 2 continued steps", len(bodies))
900 }
901 if strings.Contains(bodies[4], "Context usage is at") {
902 t.Fatalf("first continuation step carries a reminder from the compaction request: %q", bodies[4])
903 }
904 if !strings.Contains(bodies[5], "Context usage is at") {
905 t.Fatalf("second continuation step lost the reminder: %q", bodies[5])
906 }
907}
908
909func TestResidentCheckpointMessagesDropsOnlyHostContext(t *testing.T) {
910 internal := provider.Message{
911 Role: provider.RoleDeveloper,
912 Meta: map[string]string{"internal_context": "true"},
913 Content: []provider.Content{provider.TextBlock{Text: "Context usage is at 96% of the model window."}},
914 }
915 importedNote := provider.Message{
916 Role: provider.RoleDeveloper,
917 Content: []provider.Content{provider.TextBlock{Text: "imported developer note"}},
918 }
919 task := provider.Message{Role: provider.RoleUser, Content: []provider.Content{provider.TextBlock{Text: "task"}}}
920 kept := residentCheckpointMessages([]provider.Message{internal, importedNote, task})
921 if len(kept) != 2 || kept[0].Role != provider.RoleDeveloper || kept[1].Role != provider.RoleUser {
922 t.Fatalf("checkpoint messages = %#v, want host context dropped and other messages kept", kept)
923 }
924 if !core.IsInternalContextMessage(internal) || core.IsInternalContextMessage(importedNote) {
925 t.Fatal("IsInternalContextMessage misclassified a developer message")
926 }
927}
928
929func TestResidentChildRunnerKeepsHostContextOutOfCheckpoint(t *testing.T) {
930 // Tool turns report a prompt size past the top band, so the child holds a
931 // host reminder when the overflow arrives.
932 script := &residentOverflowScript{
933 toolCallRequests: 2,
934 finalText: "overflow recovered result",
935 toolUsageTokens: 1 << 20,
936 }
937 server := script.start(t)
938 runner, journal, _ := newResidentChildTestRunner(t, "checkpoint-context", residentArguments(t, server.URL))
939 if err := runner.Run(t.Context(), "review"); err != nil {
940 t.Fatalf("runner error = %v, want one recovered turn", err)
941 }
942 messages, err := subagents.ReadResidentTranscriptMessages(journal.Dir())
943 if err != nil {
944 t.Fatal(err)
945 }
946 joined := joinedResidentText(messages)
947 if !strings.Contains(joined, "compacted digest") {
948 t.Fatalf("resumed transcript = %q, want the checkpoint digest", joined)
949 }
950 if strings.Contains(joined, "Context usage is at") {
951 t.Fatalf("checkpoint replayed a stale host reminder: %q", joined)
952 }
953}
954
955func TestResidentChildRunnerKeepsRemainingStepAllowanceAcrossRecovery(t *testing.T) {
956 // Two tool turns and the overflow leave two of the five allowed steps. The
957 // continuation must receive exactly those, not a fresh limit.
958 script := &residentOverflowScript{
959 toolCallRequests: 2,
960 continuationToolCalls: 10,
961 finalText: "unreached",
962 }
963 server := script.start(t)
964 args := residentArguments(t, server.URL)
965 args.MaxSteps = 5
966 runner, journal, _ := newResidentChildTestRunner(t, "partial-allowance", args)
967 err := runner.Run(t.Context(), "review")
968 if !errors.Is(err, core.ErrMaxSteps) {
969 t.Fatalf("runner error = %v, want the continuation to exhaust the remaining allowance", err)
970 }
971 if got := script.requests.Load(); got != 6 {
972 t.Fatalf("provider requests = %d, want 2 tool turns, the overflow, the compaction, and 2 more steps", got)
973 }
974 if checkpoints := countResidentCheckpoints(t, journal); checkpoints != 1 {
975 t.Fatalf("compaction checkpoints = %d, want 1", checkpoints)
976 }
977}
978
979func TestResidentChildRunnerDoesNotCompactWithoutHistory(t *testing.T) {
980 // The overflow arrives on the first request, so the transcript is too short
981 // to summarize without replacing the task itself.
982 script := &residentOverflowScript{toolCallRequests: 0, finalText: "unreached"}
983 server := script.start(t)
984 runner, journal, _ := newResidentChildTestRunner(t, "short-overflow", residentArguments(t, server.URL))
985 err := runner.Run(t.Context(), "review")
986 if err == nil || !provider.IsContextOverflowError(err) || !strings.Contains(err.Error(), "context_length_exceeded") {
987 t.Fatalf("runner error = %v, want the original overflow", err)
988 }
989 if got := script.requests.Load(); got != 1 {
990 t.Fatalf("provider requests = %d, want no compaction attempt", got)
991 }
992 if checkpoints := countResidentCheckpoints(t, journal); checkpoints != 0 {
993 t.Fatalf("compaction checkpoints = %d, want none without history", checkpoints)
994 }
995}
996
997func joinedResidentText(messages []provider.Message) string {
998 var text strings.Builder
999 for _, message := range messages {
1000 text.WriteString(residentMessageText(message))
1001 text.WriteString("\n")
1002 }
1003 return text.String()
1004}