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