detached 3f37538

feat(subagents): compact context proactively

commit
3f37538bc1a572957890f2683598ab0b4b37bb9b
author
brice · · 2026-08-08 06:27 UTC
parent
ab30a57
signature
GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email
13 files · +973 -75
M README.md +1 -1
@@ -409,7 +409,7 @@ Opens a dialog with every persistent setting. `up`/`down` to navigate, `enter` o
409409 - **fast mode** — request the provider's fast tier where supported. It currently uses OpenAI's Fast service tier for OpenAI, OpenAI Responses, and OpenAI Codex models. Off by default; toggle it with `/fast` or `/settings`; changes apply on the next model call and persist as `fast_mode`. Other providers currently return an unsupported-provider error. Fast mode may cost more and depends on the selected model/account.
410410 - **lsp in main session** — enable the built-in `lsp` tool and code diagnostics for the main agent. Enabled by default; changes persist as `lsp_enabled`.
411411 - **lsp in sub-agents** — allow newly spawned background sub-agents to use the built-in `lsp` tool. Enabled by default; changes persist as `subagent_lsp_enabled` and apply when a child starts.
412-- **auto-compact threshold** — choose `off`, `70%`, `80%`, `85%` (default), `90%`, or `95%` of the model's advertised context window. The selected percentage controls automatic compaction before and after interactive turns and persists as `auto_compact_threshold`. Structural unfinished tails and truncated output continue as before. After a successful after-turn threshold compaction, zut may make at most two hidden, bounded recovery continuations when the assistant's visible text appears to promise future work; clear completed text settles normally, and a newer queued user request wins. An interrupted interactive session retains an outstanding handoff when resumed, including after portable export and import. Manual `/compact` does not silently start work. `off` disables percentage-based triggers but keeps manual `/compact` and automatic recovery from context-window and payload-too-large responses. Truncated-output continuation remains a separate priority rule.
412+- **auto-compact threshold** — choose `off`, `70%`, `80%`, `85%` (default), `90%`, or `95%` of the model's advertised context window. The selected percentage controls automatic compaction before and after interactive turns and before sub-agent follow-up turns, and persists as `auto_compact_threshold`. Context accounting includes uncached input plus provider-reported cache reads and writes, so cached turns still trigger compaction. Structural unfinished tails and truncated output continue as before. After a successful after-turn threshold compaction, zut may make at most two hidden, bounded recovery continuations when the assistant's visible text appears to promise future work; clear completed text settles normally, and a newer queued user request wins. An interrupted interactive session retains an outstanding handoff when resumed, including after portable export and import. Manual `/compact` does not silently start work. `off` disables percentage-based triggers but keeps manual `/compact` and automatic recovery from context-window and payload-too-large responses. Truncated-output continuation remains a separate priority rule.
413413 - **jail new sessions by default**: start every new agent with tools confined to its working directory. Off by default. The setting applies to interactive, print, JSON, RPC, and background-agent runs, persists as `jail_by_default`, and immediately updates the current interactive session. `/jail` and `/unjail` remain session-scoped overrides and do not change this default.
414414 - **compact transcript rendering**: reduce visual chrome in the chat transcript. Tool calls render as a quiet header plus indented output instead of a bordered box, and sent messages render without padded background bubbles. Off by default. Changes apply immediately and persist to `config.json` as `compact_mode`.
415415 - **show loaded resources at startup**: list the active Zutfile agent (for `zut run`), loaded `AGENTS.md` paths, extensions, and user-installed skills in compact sections above the transcript. Built-in skills are omitted. Off by default. Changes apply immediately and persist to `config.json` as `show_instructions_at_startup`.
@@ -409,7 +409,7 @@ Opens a dialog with every persistent setting. `up`/`down` to navigate, `enter` o
409 - **fast mode** — request the provider's fast tier where supported. It currently uses OpenAI's Fast service tier for OpenAI, OpenAI Responses, and OpenAI Codex models. Off by default; toggle it with `/fast` or `/settings`; changes apply on the next model call and persist as `fast_mode`. Other providers currently return an unsupported-provider error. Fast mode may cost more and depends on the selected model/account.409 - **fast mode** — request the provider's fast tier where supported. It currently uses OpenAI's Fast service tier for OpenAI, OpenAI Responses, and OpenAI Codex models. Off by default; toggle it with `/fast` or `/settings`; changes apply on the next model call and persist as `fast_mode`. Other providers currently return an unsupported-provider error. Fast mode may cost more and depends on the selected model/account.
410 - **lsp in main session** — enable the built-in `lsp` tool and code diagnostics for the main agent. Enabled by default; changes persist as `lsp_enabled`.410 - **lsp in main session** — enable the built-in `lsp` tool and code diagnostics for the main agent. Enabled by default; changes persist as `lsp_enabled`.
411 - **lsp in sub-agents** — allow newly spawned background sub-agents to use the built-in `lsp` tool. Enabled by default; changes persist as `subagent_lsp_enabled` and apply when a child starts.411 - **lsp in sub-agents** — allow newly spawned background sub-agents to use the built-in `lsp` tool. Enabled by default; changes persist as `subagent_lsp_enabled` and apply when a child starts.
412-- **auto-compact threshold** — choose `off`, `70%`, `80%`, `85%` (default), `90%`, or `95%` of the model's advertised context window. The selected percentage controls automatic compaction before and after interactive turns and persists as `auto_compact_threshold`. Structural unfinished tails and truncated output continue as before. After a successful after-turn threshold compaction, zut may make at most two hidden, bounded recovery continuations when the assistant's visible text appears to promise future work; clear completed text settles normally, and a newer queued user request wins. An interrupted interactive session retains an outstanding handoff when resumed, including after portable export and import. Manual `/compact` does not silently start work. `off` disables percentage-based triggers but keeps manual `/compact` and automatic recovery from context-window and payload-too-large responses. Truncated-output continuation remains a separate priority rule.412+- **auto-compact threshold** — choose `off`, `70%`, `80%`, `85%` (default), `90%`, or `95%` of the model's advertised context window. The selected percentage controls automatic compaction before and after interactive turns and before sub-agent follow-up turns, and persists as `auto_compact_threshold`. Context accounting includes uncached input plus provider-reported cache reads and writes, so cached turns still trigger compaction. Structural unfinished tails and truncated output continue as before. After a successful after-turn threshold compaction, zut may make at most two hidden, bounded recovery continuations when the assistant's visible text appears to promise future work; clear completed text settles normally, and a newer queued user request wins. An interrupted interactive session retains an outstanding handoff when resumed, including after portable export and import. Manual `/compact` does not silently start work. `off` disables percentage-based triggers but keeps manual `/compact` and automatic recovery from context-window and payload-too-large responses. Truncated-output continuation remains a separate priority rule.
413 - **jail new sessions by default**: start every new agent with tools confined to its working directory. Off by default. The setting applies to interactive, print, JSON, RPC, and background-agent runs, persists as `jail_by_default`, and immediately updates the current interactive session. `/jail` and `/unjail` remain session-scoped overrides and do not change this default.413 - **jail new sessions by default**: start every new agent with tools confined to its working directory. Off by default. The setting applies to interactive, print, JSON, RPC, and background-agent runs, persists as `jail_by_default`, and immediately updates the current interactive session. `/jail` and `/unjail` remain session-scoped overrides and do not change this default.
414 - **compact transcript rendering**: reduce visual chrome in the chat transcript. Tool calls render as a quiet header plus indented output instead of a bordered box, and sent messages render without padded background bubbles. Off by default. Changes apply immediately and persist to `config.json` as `compact_mode`.414 - **compact transcript rendering**: reduce visual chrome in the chat transcript. Tool calls render as a quiet header plus indented output instead of a bordered box, and sent messages render without padded background bubbles. Off by default. Changes apply immediately and persist to `config.json` as `compact_mode`.
415 - **show loaded resources at startup**: list the active Zutfile agent (for `zut run`), loaded `AGENTS.md` paths, extensions, and user-installed skills in compact sections above the transcript. Built-in skills are omitted. Off by default. Changes apply immediately and persist to `config.json` as `show_instructions_at_startup`.415 - **show loaded resources at startup**: list the active Zutfile agent (for `zut run`), loaded `AGENTS.md` paths, extensions, and user-installed skills in compact sections above the transcript. Built-in skills are omitted. Off by default. Changes apply immediately and persist to `config.json` as `show_instructions_at_startup`.
M packages/agent/build.go +14 -5
@@ -45,6 +45,11 @@ type Resolved struct {
4545 MaxSteps int
4646 Sandbox *tools.Sandbox
4747
48+ // ContextWindow and MaxOutput retain the effective model metadata,
49+ // including metadata synthesized for valid open-catalog local and routed
50+ // models that provider.FindModel cannot rediscover later.
51+ ContextWindow int
52+
4853 // MaxOutput is the resolved model's maximum output-token budget
4954 // (from the catalog). Passed to the agent so each turn requests
5055 // the model's full output capacity instead of the provider's
@@ -684,11 +689,13 @@ func Resolve(args Args, requireCred bool) (Resolved, error) {
684689 if err != nil {
685690 if cfg, ok := provider.CustomProviders()[provName]; ok {
686691 resolvedModel = provider.Model{
687- Provider: provName,
688- ID: model,
689- DisplayName: model,
690- BaseURL: cfg.BaseURL,
691- Source: "user",
692+ Provider: provName,
693+ ID: model,
694+ DisplayName: model,
695+ ContextWindow: 128000,
696+ MaxOutput: 16384,
697+ BaseURL: cfg.BaseURL,
698+ Source: "user",
692699 }
693700 err = nil
694701 }
@@ -922,6 +929,7 @@ func Resolve(args Args, requireCred bool) (Resolved, error) {
922929 ToolSummary: summaries,
923930 SystemPrompt: sys,
924931 MaxSteps: max,
932+ ContextWindow: resolvedModel.ContextWindow,
925933 MaxOutput: resolvedModel.MaxOutput,
926934 Sandbox: sandbox,
927935 SkillTool: skillTool,
@@ -1243,6 +1251,7 @@ func (r *Resolved) UseSandbox(s *tools.Sandbox) {
12431251 func (r Resolved) NewAgent() *core.Agent {
12441252 a := core.NewAgent(r.NewClient(), r.Model, r.SystemPrompt, r.ToolRegistry)
12451253 a.MaxSteps = r.MaxSteps
1254+ a.ContextWindow = r.ContextWindow
12461255 a.MaxTokens = r.MaxOutput
12471256 a.Reasoning = r.Reasoning
12481257 a.Temperature = r.Temperature
@@ -45,6 +45,11 @@ type Resolved struct {
45 MaxSteps int45 MaxSteps int
46 Sandbox *tools.Sandbox46 Sandbox *tools.Sandbox
47 47
48+ // ContextWindow and MaxOutput retain the effective model metadata,
49+ // including metadata synthesized for valid open-catalog local and routed
50+ // models that provider.FindModel cannot rediscover later.
51+ ContextWindow int
52+
48 // MaxOutput is the resolved model's maximum output-token budget53 // MaxOutput is the resolved model's maximum output-token budget
49 // (from the catalog). Passed to the agent so each turn requests54 // (from the catalog). Passed to the agent so each turn requests
50 // the model's full output capacity instead of the provider's55 // the model's full output capacity instead of the provider's
@@ -684,11 +689,13 @@ func Resolve(args Args, requireCred bool) (Resolved, error) {
684 if err != nil {689 if err != nil {
685 if cfg, ok := provider.CustomProviders()[provName]; ok {690 if cfg, ok := provider.CustomProviders()[provName]; ok {
686 resolvedModel = provider.Model{691 resolvedModel = provider.Model{
687- Provider: provName,692+ Provider: provName,
688- ID: model,693+ ID: model,
689- DisplayName: model,694+ DisplayName: model,
690- BaseURL: cfg.BaseURL,695+ ContextWindow: 128000,
691- Source: "user",696+ MaxOutput: 16384,
697+ BaseURL: cfg.BaseURL,
698+ Source: "user",
692 }699 }
693 err = nil700 err = nil
694 }701 }
@@ -922,6 +929,7 @@ func Resolve(args Args, requireCred bool) (Resolved, error) {
922 ToolSummary: summaries,929 ToolSummary: summaries,
923 SystemPrompt: sys,930 SystemPrompt: sys,
924 MaxSteps: max,931 MaxSteps: max,
932+ ContextWindow: resolvedModel.ContextWindow,
925 MaxOutput: resolvedModel.MaxOutput,933 MaxOutput: resolvedModel.MaxOutput,
926 Sandbox: sandbox,934 Sandbox: sandbox,
927 SkillTool: skillTool,935 SkillTool: skillTool,
@@ -1243,6 +1251,7 @@ func (r *Resolved) UseSandbox(s *tools.Sandbox) {
1243 func (r Resolved) NewAgent() *core.Agent {1251 func (r Resolved) NewAgent() *core.Agent {
1244 a := core.NewAgent(r.NewClient(), r.Model, r.SystemPrompt, r.ToolRegistry)1252 a := core.NewAgent(r.NewClient(), r.Model, r.SystemPrompt, r.ToolRegistry)
1245 a.MaxSteps = r.MaxSteps1253 a.MaxSteps = r.MaxSteps
1254+ a.ContextWindow = r.ContextWindow
1246 a.MaxTokens = r.MaxOutput1255 a.MaxTokens = r.MaxOutput
1247 a.Reasoning = r.Reasoning1256 a.Reasoning = r.Reasoning
1248 a.Temperature = r.Temperature1257 a.Temperature = r.Temperature
M packages/agent/build_test.go +10 -1
@@ -745,6 +745,9 @@ func TestResolveOpenRouterPreservesSavedRoutedModelID(t *testing.T) {
745745 if r.MaxOutput != 64000 {
746746 t.Fatalf("MaxOutput = %d, want synthetic gateway default 64000", r.MaxOutput)
747747 }
748+ if r.ContextWindow != 1_000_000 || r.NewAgent().ContextWindow != 1_000_000 {
749+ t.Fatalf("resolved/agent context window = %d/%d, want synthetic gateway metadata", r.ContextWindow, r.NewAgent().ContextWindow)
750+ }
748751 }
749752
750753 func TestResolveGatewayPlainUnknownModelFallsBack(t *testing.T) {
@@ -856,6 +859,9 @@ func TestResolveLlamaCPPUsesRouterInferenceURL(t *testing.T) {
856859 if r.Credential != "local" || !r.HasCredential() {
857860 t.Fatalf("credential = %q", r.Credential)
858861 }
862+ if r.ContextWindow != 128000 || r.NewAgent().ContextWindow != 128000 {
863+ t.Fatalf("resolved/agent context window = %d/%d, want synthesized local metadata", r.ContextWindow, r.NewAgent().ContextWindow)
864+ }
859865 if got := r.NewClient().Name(); got != provider.LlamaCPPProviderID {
860866 t.Fatalf("client name = %q, want %q", got, provider.LlamaCPPProviderID)
861867 }
@@ -888,7 +894,7 @@ func TestResolveCustomProviderModelBaseURLBeatsProviderBaseURL(t *testing.T) {
888894 "baseUrl": "https://provider.example.com/v1",
889895 "api": "openai",
890896 "models": [
891- {"id": "fast", "baseUrl": "https://model.example.com/v1"}
897+ {"id": "fast", "baseUrl": "https://model.example.com/v1", "contextWindow": 65536}
892898 ]
893899 }
894900 }
@@ -910,6 +916,9 @@ func TestResolveCustomProviderModelBaseURLBeatsProviderBaseURL(t *testing.T) {
910916 if r.BaseURL != "https://model.example.com/v1" {
911917 t.Fatalf("BaseURL = %q, want model-level baseUrl", r.BaseURL)
912918 }
919+ if r.ContextWindow != 65536 || r.NewAgent().ContextWindow != 65536 {
920+ t.Fatalf("resolved/agent context window = %d/%d, want retained user-model metadata", r.ContextWindow, r.NewAgent().ContextWindow)
921+ }
913922 }
914923
915924 func TestCustomProviderUsesOpenAIResponsesAPI(t *testing.T) {
@@ -745,6 +745,9 @@ func TestResolveOpenRouterPreservesSavedRoutedModelID(t *testing.T) {
745 if r.MaxOutput != 64000 {745 if r.MaxOutput != 64000 {
746 t.Fatalf("MaxOutput = %d, want synthetic gateway default 64000", r.MaxOutput)746 t.Fatalf("MaxOutput = %d, want synthetic gateway default 64000", r.MaxOutput)
747 }747 }
748+ if r.ContextWindow != 1_000_000 || r.NewAgent().ContextWindow != 1_000_000 {
749+ t.Fatalf("resolved/agent context window = %d/%d, want synthetic gateway metadata", r.ContextWindow, r.NewAgent().ContextWindow)
750+ }
748 }751 }
749 752
750 func TestResolveGatewayPlainUnknownModelFallsBack(t *testing.T) {753 func TestResolveGatewayPlainUnknownModelFallsBack(t *testing.T) {
@@ -856,6 +859,9 @@ func TestResolveLlamaCPPUsesRouterInferenceURL(t *testing.T) {
856 if r.Credential != "local" || !r.HasCredential() {859 if r.Credential != "local" || !r.HasCredential() {
857 t.Fatalf("credential = %q", r.Credential)860 t.Fatalf("credential = %q", r.Credential)
858 }861 }
862+ if r.ContextWindow != 128000 || r.NewAgent().ContextWindow != 128000 {
863+ t.Fatalf("resolved/agent context window = %d/%d, want synthesized local metadata", r.ContextWindow, r.NewAgent().ContextWindow)
864+ }
859 if got := r.NewClient().Name(); got != provider.LlamaCPPProviderID {865 if got := r.NewClient().Name(); got != provider.LlamaCPPProviderID {
860 t.Fatalf("client name = %q, want %q", got, provider.LlamaCPPProviderID)866 t.Fatalf("client name = %q, want %q", got, provider.LlamaCPPProviderID)
861 }867 }
@@ -888,7 +894,7 @@ func TestResolveCustomProviderModelBaseURLBeatsProviderBaseURL(t *testing.T) {
888 "baseUrl": "https://provider.example.com/v1",894 "baseUrl": "https://provider.example.com/v1",
889 "api": "openai",895 "api": "openai",
890 "models": [896 "models": [
891- {"id": "fast", "baseUrl": "https://model.example.com/v1"}897+ {"id": "fast", "baseUrl": "https://model.example.com/v1", "contextWindow": 65536}
892 ]898 ]
893 }899 }
894 }900 }
@@ -910,6 +916,9 @@ func TestResolveCustomProviderModelBaseURLBeatsProviderBaseURL(t *testing.T) {
910 if r.BaseURL != "https://model.example.com/v1" {916 if r.BaseURL != "https://model.example.com/v1" {
911 t.Fatalf("BaseURL = %q, want model-level baseUrl", r.BaseURL)917 t.Fatalf("BaseURL = %q, want model-level baseUrl", r.BaseURL)
912 }918 }
919+ if r.ContextWindow != 65536 || r.NewAgent().ContextWindow != 65536 {
920+ t.Fatalf("resolved/agent context window = %d/%d, want retained user-model metadata", r.ContextWindow, r.NewAgent().ContextWindow)
921+ }
913 }922 }
914 923
915 func TestCustomProviderUsesOpenAIResponsesAPI(t *testing.T) {924 func TestCustomProviderUsesOpenAIResponsesAPI(t *testing.T) {
M packages/agent/cli.go +53 -16
@@ -942,7 +942,12 @@ func runPrintMode(ctx context.Context, args Args, version string) error {
942942 if recovery.Compacted {
943943 transcriptStart = recovery.OutputStart
944944 }
945- WriteNewTranscript(ag, sess, transcriptStart)
945+ if persistErr := WriteNewTranscript(ag, sess, transcriptStart); persistErr != nil {
946+ if err != nil {
947+ return errors.Join(err, persistErr)
948+ }
949+ return persistErr
950+ }
946951 if err != nil {
947952 return err
948953 }
@@ -1016,7 +1021,12 @@ func runStreamMode(ctx context.Context, args Args, version string) error {
10161021 if recovery.Compacted {
10171022 transcriptStart = recovery.OutputStart
10181023 }
1019- WriteNewTranscript(ag, sess, transcriptStart)
1024+ if persistErr := WriteNewTranscript(ag, sess, transcriptStart); persistErr != nil {
1025+ if err != nil {
1026+ return errors.Join(err, persistErr)
1027+ }
1028+ return persistErr
1029+ }
10201030 return err
10211031 }
10221032
@@ -1131,7 +1141,12 @@ func runJSONMode(ctx context.Context, args Args, version string) error {
11311141 if recovery.Compacted {
11321142 transcriptStart = recovery.OutputStart
11331143 }
1134- WriteNewTranscript(ag, sess, transcriptStart)
1144+ if persistErr := WriteNewTranscript(ag, sess, transcriptStart); persistErr != nil {
1145+ if err != nil {
1146+ return errors.Join(err, persistErr)
1147+ }
1148+ return persistErr
1149+ }
11351150 return err
11361151 }
11371152
@@ -1881,7 +1896,10 @@ func runInteractive(ctx context.Context, args Args, version string) error {
18811896 // than the public callback, which would acquire the read lock held
18821897 // by this transition.
18831898 if oldSess != nil {
1884- writeNewTranscriptLocked(currentAg, oldSess, sessBaselineMsgs)
1899+ if err := writeNewTranscriptLocked(currentAg, oldSess, sessBaselineMsgs); err != nil {
1900+ persistMu.Unlock()
1901+ return fmt.Errorf("flush current session: %w", err)
1902+ }
18851903 }
18861904 currentMessageCount := len(currentAg.Messages())
18871905 currentCost := currentAg.Cost()
@@ -2116,7 +2134,12 @@ func runInteractive(ctx context.Context, args Args, version string) error {
21162134 var newStates map[string]json.RawMessage
21172135 persistMu.Lock()
21182136 if sess != nil {
2119- writeNewTranscriptLocked(currentAg, sess, sessBaselineMsgs)
2137+ if err := writeNewTranscriptLocked(currentAg, sess, sessBaselineMsgs); err != nil {
2138+ persistMu.Unlock()
2139+ _ = newSess.Close()
2140+ rollback()
2141+ return fmt.Errorf("flush current session: %w", err)
2142+ }
21202143 _ = sess.Close()
21212144 }
21222145 sess = newSess
@@ -2424,7 +2447,10 @@ func runInteractive(ctx context.Context, args Args, version string) error {
24242447 if sess == nil {
24252448 return
24262449 }
2427- writeNewTranscriptLocked(currentAg, sess, sessBaselineMsgs)
2450+ if err := writeNewTranscriptLocked(currentAg, sess, sessBaselineMsgs); err != nil {
2451+ iv.ReportError(fmt.Errorf("flush session: %w", err))
2452+ return
2453+ }
24282454 sessBaselineMsgs = len(currentAg.Messages())
24292455 },
24302456 SessionTransition: newSessionTransition(&sessionTransitionMu),
@@ -2544,8 +2570,11 @@ func runInteractive(ctx context.Context, args Args, version string) error {
25442570 if finalAg != nil {
25452571 persistMu.Lock()
25462572 if sess != nil {
2547- writeNewTranscriptLocked(finalAg, sess, sessBaselineMsgs)
2548- sessBaselineMsgs = len(finalAg.Messages())
2573+ if err := writeNewTranscriptLocked(finalAg, sess, sessBaselineMsgs); err != nil {
2574+ fmt.Fprintf(os.Stderr, "flush session: %v\n", err)
2575+ } else {
2576+ sessBaselineMsgs = len(finalAg.Messages())
2577+ }
25492578 _ = sess.Close()
25502579 sess = nil
25512580 }
@@ -2573,8 +2602,11 @@ func runInteractive(ctx context.Context, args Args, version string) error {
25732602 if finalAg := iv.Agent(); finalAg != nil {
25742603 persistMu.Lock()
25752604 if sess != nil {
2576- writeNewTranscriptLocked(finalAg, sess, sessBaselineMsgs)
2577- sessBaselineMsgs = len(finalAg.Messages())
2605+ if err := writeNewTranscriptLocked(finalAg, sess, sessBaselineMsgs); err != nil {
2606+ runErr = errors.Join(runErr, fmt.Errorf("flush session: %w", err))
2607+ } else {
2608+ sessBaselineMsgs = len(finalAg.Messages())
2609+ }
25782610 }
25792611 persistMu.Unlock()
25802612 }
@@ -2671,24 +2703,29 @@ func pickSession(root, cwd string) (string, error) {
26712703 // agent's transcript to the session. Used by callers that don't hold
26722704 // the persistMu (non-interactive print/json modes which run a single
26732705 // turn under their own goroutine).
2674-func WriteNewTranscript(ag *core.Agent, sess *core.Session, from int) {
2675- writeNewTranscriptLocked(ag, sess, from)
2706+func WriteNewTranscript(ag *core.Agent, sess *core.Session, from int) error {
2707+ return writeNewTranscriptLocked(ag, sess, from)
26762708 }
26772709
26782710 // writeNewTranscriptLocked is the same as WriteNewTranscript. The
26792711 // suffix marks that interactive callers must hold persistMu when
26802712 // invoking it so concurrent appends from the agent loop don't race
26812713 // with this catch-up flush.
2682-func writeNewTranscriptLocked(ag *core.Agent, sess *core.Session, from int) {
2714+func writeNewTranscriptLocked(ag *core.Agent, sess *core.Session, from int) error {
26832715 if sess == nil || ag == nil {
2684- return
2716+ return nil
26852717 }
26862718 msgs := ag.Messages()
26872719 for i := from; i < len(msgs); i++ {
2688- _ = sess.AppendMessage(msgs[i])
2720+ if err := sess.AppendMessage(msgs[i]); err != nil {
2721+ return fmt.Errorf("append transcript message %d: %w", i, err)
2722+ }
26892723 }
26902724 cum := ag.Cost()
2691- _ = sess.AppendUsage(cum, cum)
2725+ if err := sess.AppendUsage(cum, cum); err != nil {
2726+ return fmt.Errorf("append transcript usage: %w", err)
2727+ }
2728+ return nil
26922729 }
26932730
26942731 func readAllStdin() (string, error) {
@@ -942,7 +942,12 @@ func runPrintMode(ctx context.Context, args Args, version string) error {
942 if recovery.Compacted {942 if recovery.Compacted {
943 transcriptStart = recovery.OutputStart943 transcriptStart = recovery.OutputStart
944 }944 }
945- WriteNewTranscript(ag, sess, transcriptStart)945+ if persistErr := WriteNewTranscript(ag, sess, transcriptStart); persistErr != nil {
946+ if err != nil {
947+ return errors.Join(err, persistErr)
948+ }
949+ return persistErr
950+ }
946 if err != nil {951 if err != nil {
947 return err952 return err
948 }953 }
@@ -1016,7 +1021,12 @@ func runStreamMode(ctx context.Context, args Args, version string) error {
1016 if recovery.Compacted {1021 if recovery.Compacted {
1017 transcriptStart = recovery.OutputStart1022 transcriptStart = recovery.OutputStart
1018 }1023 }
1019- WriteNewTranscript(ag, sess, transcriptStart)1024+ if persistErr := WriteNewTranscript(ag, sess, transcriptStart); persistErr != nil {
1025+ if err != nil {
1026+ return errors.Join(err, persistErr)
1027+ }
1028+ return persistErr
1029+ }
1020 return err1030 return err
1021 }1031 }
1022 1032
@@ -1131,7 +1141,12 @@ func runJSONMode(ctx context.Context, args Args, version string) error {
1131 if recovery.Compacted {1141 if recovery.Compacted {
1132 transcriptStart = recovery.OutputStart1142 transcriptStart = recovery.OutputStart
1133 }1143 }
1134- WriteNewTranscript(ag, sess, transcriptStart)1144+ if persistErr := WriteNewTranscript(ag, sess, transcriptStart); persistErr != nil {
1145+ if err != nil {
1146+ return errors.Join(err, persistErr)
1147+ }
1148+ return persistErr
1149+ }
1135 return err1150 return err
1136 }1151 }
1137 1152
@@ -1881,7 +1896,10 @@ func runInteractive(ctx context.Context, args Args, version string) error {
1881 // than the public callback, which would acquire the read lock held1896 // than the public callback, which would acquire the read lock held
1882 // by this transition.1897 // by this transition.
1883 if oldSess != nil {1898 if oldSess != nil {
1884- writeNewTranscriptLocked(currentAg, oldSess, sessBaselineMsgs)1899+ if err := writeNewTranscriptLocked(currentAg, oldSess, sessBaselineMsgs); err != nil {
1900+ persistMu.Unlock()
1901+ return fmt.Errorf("flush current session: %w", err)
1902+ }
1885 }1903 }
1886 currentMessageCount := len(currentAg.Messages())1904 currentMessageCount := len(currentAg.Messages())
1887 currentCost := currentAg.Cost()1905 currentCost := currentAg.Cost()
@@ -2116,7 +2134,12 @@ func runInteractive(ctx context.Context, args Args, version string) error {
2116 var newStates map[string]json.RawMessage2134 var newStates map[string]json.RawMessage
2117 persistMu.Lock()2135 persistMu.Lock()
2118 if sess != nil {2136 if sess != nil {
2119- writeNewTranscriptLocked(currentAg, sess, sessBaselineMsgs)2137+ if err := writeNewTranscriptLocked(currentAg, sess, sessBaselineMsgs); err != nil {
2138+ persistMu.Unlock()
2139+ _ = newSess.Close()
2140+ rollback()
2141+ return fmt.Errorf("flush current session: %w", err)
2142+ }
2120 _ = sess.Close()2143 _ = sess.Close()
2121 }2144 }
2122 sess = newSess2145 sess = newSess
@@ -2424,7 +2447,10 @@ func runInteractive(ctx context.Context, args Args, version string) error {
2424 if sess == nil {2447 if sess == nil {
2425 return2448 return
2426 }2449 }
2427- writeNewTranscriptLocked(currentAg, sess, sessBaselineMsgs)2450+ if err := writeNewTranscriptLocked(currentAg, sess, sessBaselineMsgs); err != nil {
2451+ iv.ReportError(fmt.Errorf("flush session: %w", err))
2452+ return
2453+ }
2428 sessBaselineMsgs = len(currentAg.Messages())2454 sessBaselineMsgs = len(currentAg.Messages())
2429 },2455 },
2430 SessionTransition: newSessionTransition(&sessionTransitionMu),2456 SessionTransition: newSessionTransition(&sessionTransitionMu),
@@ -2544,8 +2570,11 @@ func runInteractive(ctx context.Context, args Args, version string) error {
2544 if finalAg != nil {2570 if finalAg != nil {
2545 persistMu.Lock()2571 persistMu.Lock()
2546 if sess != nil {2572 if sess != nil {
2547- writeNewTranscriptLocked(finalAg, sess, sessBaselineMsgs)2573+ if err := writeNewTranscriptLocked(finalAg, sess, sessBaselineMsgs); err != nil {
2548- sessBaselineMsgs = len(finalAg.Messages())2574+ fmt.Fprintf(os.Stderr, "flush session: %v\n", err)
2575+ } else {
2576+ sessBaselineMsgs = len(finalAg.Messages())
2577+ }
2549 _ = sess.Close()2578 _ = sess.Close()
2550 sess = nil2579 sess = nil
2551 }2580 }
@@ -2573,8 +2602,11 @@ func runInteractive(ctx context.Context, args Args, version string) error {
2573 if finalAg := iv.Agent(); finalAg != nil {2602 if finalAg := iv.Agent(); finalAg != nil {
2574 persistMu.Lock()2603 persistMu.Lock()
2575 if sess != nil {2604 if sess != nil {
2576- writeNewTranscriptLocked(finalAg, sess, sessBaselineMsgs)2605+ if err := writeNewTranscriptLocked(finalAg, sess, sessBaselineMsgs); err != nil {
2577- sessBaselineMsgs = len(finalAg.Messages())2606+ runErr = errors.Join(runErr, fmt.Errorf("flush session: %w", err))
2607+ } else {
2608+ sessBaselineMsgs = len(finalAg.Messages())
2609+ }
2578 }2610 }
2579 persistMu.Unlock()2611 persistMu.Unlock()
2580 }2612 }
@@ -2671,24 +2703,29 @@ func pickSession(root, cwd string) (string, error) {
2671 // agent's transcript to the session. Used by callers that don't hold2703 // agent's transcript to the session. Used by callers that don't hold
2672 // the persistMu (non-interactive print/json modes which run a single2704 // the persistMu (non-interactive print/json modes which run a single
2673 // turn under their own goroutine).2705 // turn under their own goroutine).
2674-func WriteNewTranscript(ag *core.Agent, sess *core.Session, from int) {2706+func WriteNewTranscript(ag *core.Agent, sess *core.Session, from int) error {
2675- writeNewTranscriptLocked(ag, sess, from)2707+ return writeNewTranscriptLocked(ag, sess, from)
2676 }2708 }
2677 2709
2678 // writeNewTranscriptLocked is the same as WriteNewTranscript. The2710 // writeNewTranscriptLocked is the same as WriteNewTranscript. The
2679 // suffix marks that interactive callers must hold persistMu when2711 // suffix marks that interactive callers must hold persistMu when
2680 // invoking it so concurrent appends from the agent loop don't race2712 // invoking it so concurrent appends from the agent loop don't race
2681 // with this catch-up flush.2713 // with this catch-up flush.
2682-func writeNewTranscriptLocked(ag *core.Agent, sess *core.Session, from int) {2714+func writeNewTranscriptLocked(ag *core.Agent, sess *core.Session, from int) error {
2683 if sess == nil || ag == nil {2715 if sess == nil || ag == nil {
2684- return2716+ return nil
2685 }2717 }
2686 msgs := ag.Messages()2718 msgs := ag.Messages()
2687 for i := from; i < len(msgs); i++ {2719 for i := from; i < len(msgs); i++ {
2688- _ = sess.AppendMessage(msgs[i])2720+ if err := sess.AppendMessage(msgs[i]); err != nil {
2721+ return fmt.Errorf("append transcript message %d: %w", i, err)
2722+ }
2689 }2723 }
2690 cum := ag.Cost()2724 cum := ag.Cost()
2691- _ = sess.AppendUsage(cum, cum)2725+ if err := sess.AppendUsage(cum, cum); err != nil {
2726+ return fmt.Errorf("append transcript usage: %w", err)
2727+ }
2728+ return nil
2692 }2729 }
2693 2730
2694 func readAllStdin() (string, error) {2731 func readAllStdin() (string, error) {
M packages/agent/modes/auto_compact_test.go +4 -4
@@ -28,8 +28,8 @@ func TestNormalizeAutoCompactThreshold(t *testing.T) {
2828
2929 for _, tt := range tests {
3030 t.Run(tt.name, func(t *testing.T) {
31- if got := normalizeAutoCompactThreshold(tt.value); got != tt.want {
32- t.Fatalf("normalizeAutoCompactThreshold(%v) = %d, want %d", tt.value, got, tt.want)
31+ if got := NormalizeAutoCompactThreshold(tt.value); got != tt.want {
32+ t.Fatalf("NormalizeAutoCompactThreshold(%v) = %d, want %d", tt.value, got, tt.want)
3333 }
3434 })
3535 }
@@ -353,8 +353,8 @@ func TestShouldAutoCompactUsesConfiguredThreshold(t *testing.T) {
353353
354354 for _, tt := range tests {
355355 t.Run(tt.name, func(t *testing.T) {
356- if got := shouldAutoCompact(tt.used, tt.window, tt.threshold); got != tt.want {
357- t.Fatalf("shouldAutoCompact(%d, %d, %d) = %t, want %t", tt.used, tt.window, tt.threshold, got, tt.want)
356+ if got := ShouldAutoCompact(tt.used, tt.window, tt.threshold); got != tt.want {
357+ t.Fatalf("ShouldAutoCompact(%d, %d, %d) = %t, want %t", tt.used, tt.window, tt.threshold, got, tt.want)
358358 }
359359 })
360360 }
@@ -28,8 +28,8 @@ func TestNormalizeAutoCompactThreshold(t *testing.T) {
28 28
29 for _, tt := range tests {29 for _, tt := range tests {
30 t.Run(tt.name, func(t *testing.T) {30 t.Run(tt.name, func(t *testing.T) {
31- if got := normalizeAutoCompactThreshold(tt.value); got != tt.want {31+ if got := NormalizeAutoCompactThreshold(tt.value); got != tt.want {
32- t.Fatalf("normalizeAutoCompactThreshold(%v) = %d, want %d", tt.value, got, tt.want)32+ t.Fatalf("NormalizeAutoCompactThreshold(%v) = %d, want %d", tt.value, got, tt.want)
33 }33 }
34 })34 })
35 }35 }
@@ -353,8 +353,8 @@ func TestShouldAutoCompactUsesConfiguredThreshold(t *testing.T) {
353 353
354 for _, tt := range tests {354 for _, tt := range tests {
355 t.Run(tt.name, func(t *testing.T) {355 t.Run(tt.name, func(t *testing.T) {
356- if got := shouldAutoCompact(tt.used, tt.window, tt.threshold); got != tt.want {356+ if got := ShouldAutoCompact(tt.used, tt.window, tt.threshold); got != tt.want {
357- t.Fatalf("shouldAutoCompact(%d, %d, %d) = %t, want %t", tt.used, tt.window, tt.threshold, got, tt.want)357+ t.Fatalf("ShouldAutoCompact(%d, %d, %d) = %t, want %t", tt.used, tt.window, tt.threshold, got, tt.want)
358 }358 }
359 })359 })
360 }360 }
M packages/agent/modes/context_recovery.go +86 -19
@@ -17,6 +17,13 @@ type ContextRecoveryOptions struct {
1717 // transcript and prevents the continuation request.
1818 PersistCompaction func([]provider.Message) error
1919
20+ // ContextWindow and AutoCompactThreshold enable proactive compaction before
21+ // Prompt. The threshold uses the same supported percentage values and 85%
22+ // default as interactive mode; zero disables it. Context usage includes
23+ // uncached input, cache reads, and cache writes from the latest request.
24+ ContextWindow int
25+ AutoCompactThreshold *int
26+
2027 // SuppressInitialOverflowEvent hides the recoverable first-attempt
2128 // EvTurnEnd overflow from sinks whose protocol cannot retract a terminal
2229 // error frame, such as JSONL. A retry overflow is never suppressed.
@@ -31,19 +38,54 @@ type ContextRecoveryResult struct {
3138 Compacted bool
3239 }
3340
34-// PromptWithContextRecovery appends prompt and runs it once. If that provider
35-// request exceeds its context window, it compacts the existing transcript and
36-// continues the already-appended prompt exactly once. Cancellation, a
37-// one-message transcript, a compaction failure, and a second overflow remain
38-// terminal outcomes.
41+// PromptWithContextRecovery proactively compacts a transcript whose latest
42+// request reached the configured context percentage, then appends prompt and
43+// runs it once. If that provider request still exceeds its context window, it
44+// compacts the existing transcript and continues the already-appended prompt
45+// exactly once. Cancellation, a compaction failure, and a second overflow
46+// remain terminal outcomes.
3947 func PromptWithContextRecovery(ctx context.Context, ag *core.Agent, prompt string, images []provider.ImageBlock, sink func(core.AgentEvent), options ContextRecoveryOptions) (result ContextRecoveryResult, err error) {
4048 result.OutputStart = len(ag.Messages())
4149 if sink == nil {
4250 sink = func(core.AgentEvent) {}
4351 }
4452
53+ usage := ag.LastTurnUsage()
54+ contextUsed := usage.InputTokens + usage.CacheReadTokens + usage.CacheWriteTokens
55+ threshold := NormalizeAutoCompactThreshold(options.AutoCompactThreshold)
56+ proactiveSummaryOverflow := false
57+ if ShouldAutoCompact(contextUsed, options.ContextWindow, threshold) {
58+ messages := ag.Messages()
59+ // A sole transcript message is commonly the worker's only task.
60+ // Replacing it before it has run loses the exact instruction.
61+ if len(messages) > 1 {
62+ if err := ctx.Err(); err != nil {
63+ return result, err
64+ }
65+ compacted, compactErr := compactContext(ctx, ag, messages, options.PersistCompaction)
66+ if compactErr != nil {
67+ if ctxErr := ctx.Err(); ctxErr != nil {
68+ return result, ctxErr
69+ }
70+ if !provider.IsContextOverflowError(compactErr) {
71+ return result, fmt.Errorf("proactively compact transcript: %w", compactErr)
72+ }
73+ // The summary request can itself exceed a provider limit. The
74+ // transcript is unchanged, so make one normal prompt attempt. If
75+ // that also overflows, do not repeat the doomed summary request.
76+ proactiveSummaryOverflow = true
77+ } else {
78+ result.OutputStart = len(compacted)
79+ result.Compacted = true
80+ if err := ctx.Err(); err != nil {
81+ return result, err
82+ }
83+ }
84+ }
85+ }
86+
4587 firstSink := sink
46- if options.SuppressInitialOverflowEvent {
88+ if options.SuppressInitialOverflowEvent && !proactiveSummaryOverflow {
4789 firstSink = func(event core.AgentEvent) {
4890 if turnEnd, ok := event.(core.EvTurnEnd); ok && provider.IsContextOverflowError(turnEnd.Err) {
4991 return
@@ -52,33 +94,58 @@ func PromptWithContextRecovery(ctx context.Context, ag *core.Agent, prompt strin
5294 }
5395 }
5496
55- if err = ag.Prompt(ctx, prompt, images, firstSink); err == nil || ctx.Err() != nil || !provider.IsContextOverflowError(err) {
97+ if err = ag.Prompt(ctx, prompt, images, firstSink); err == nil || ctx.Err() != nil || !provider.IsContextOverflowError(err) || proactiveSummaryOverflow {
5698 return result, err
5799 }
58100
101+ if ctxErr := ctx.Err(); ctxErr != nil {
102+ return result, ctxErr
103+ }
59104 messages := ag.Messages()
60105 if len(messages) <= 1 {
61106 // The failed prompt is the only message. Summarizing it would replace
62107 // the user's task with an LLM-generated paraphrase.
63108 return result, err
64109 }
65- keepTail := min(4, len(messages)-1)
66- if _, compactErr := ag.Compact(ctx, keepTail, nil); compactErr != nil {
110+ compacted, compactErr := compactContext(ctx, ag, messages, options.PersistCompaction)
111+ if compactErr != nil {
67112 return result, fmt.Errorf("compact transcript after initial overflow: %w", compactErr)
68113 }
69114
115+ result.OutputStart = len(compacted)
116+ result.Compacted = true
117+ if ctxErr := ctx.Err(); ctxErr != nil {
118+ return result, ctxErr
119+ }
120+ return result, ag.Continue(ctx, sink)
121+}
122+
123+func compactContext(ctx context.Context, ag *core.Agent, messages []provider.Message, persist func([]provider.Message) error) ([]provider.Message, error) {
124+ if err := ctx.Err(); err != nil {
125+ return nil, err
126+ }
127+ keepTail := min(4, len(messages)-1)
128+ if _, err := ag.Compact(ctx, keepTail, nil); err != nil {
129+ return nil, err
130+ }
131+
70132 compacted := ag.Messages()
71- if options.PersistCompaction != nil {
72- if persistErr := options.PersistCompaction(compacted); persistErr != nil {
73- // The session did not receive the compaction checkpoint, so restore
74- // the original history plus the already-appended prompt rather than
75- // let a later transcript flush make the session inconsistent.
133+ if err := ctx.Err(); err != nil {
134+ ag.SetMessages(messages)
135+ return nil, err
136+ }
137+ if persist != nil {
138+ if err := ctx.Err(); err != nil {
76139 ag.SetMessages(messages)
77- return result, fmt.Errorf("persist compacted transcript: %w", persistErr)
140+ return nil, err
141+ }
142+ if err := persist(compacted); err != nil {
143+ // The session did not receive the compaction checkpoint. Restore the
144+ // exact pre-compaction transcript so a later suffix flush cannot mix
145+ // the old durable generation with the new in-memory generation.
146+ ag.SetMessages(messages)
147+ return nil, fmt.Errorf("persist compacted transcript: %w", err)
78148 }
79149 }
80-
81- result.OutputStart = len(compacted)
82- result.Compacted = true
83- return result, ag.Continue(ctx, sink)
150+ return compacted, nil
84151 }
@@ -17,6 +17,13 @@ type ContextRecoveryOptions struct {
17 // transcript and prevents the continuation request.17 // transcript and prevents the continuation request.
18 PersistCompaction func([]provider.Message) error18 PersistCompaction func([]provider.Message) error
19 19
20+ // ContextWindow and AutoCompactThreshold enable proactive compaction before
21+ // Prompt. The threshold uses the same supported percentage values and 85%
22+ // default as interactive mode; zero disables it. Context usage includes
23+ // uncached input, cache reads, and cache writes from the latest request.
24+ ContextWindow int
25+ AutoCompactThreshold *int
26+
20 // SuppressInitialOverflowEvent hides the recoverable first-attempt27 // SuppressInitialOverflowEvent hides the recoverable first-attempt
21 // EvTurnEnd overflow from sinks whose protocol cannot retract a terminal28 // EvTurnEnd overflow from sinks whose protocol cannot retract a terminal
22 // error frame, such as JSONL. A retry overflow is never suppressed.29 // error frame, such as JSONL. A retry overflow is never suppressed.
@@ -31,19 +38,54 @@ type ContextRecoveryResult struct {
31 Compacted bool38 Compacted bool
32 }39 }
33 40
34-// PromptWithContextRecovery appends prompt and runs it once. If that provider41+// PromptWithContextRecovery proactively compacts a transcript whose latest
35-// request exceeds its context window, it compacts the existing transcript and42+// request reached the configured context percentage, then appends prompt and
36-// continues the already-appended prompt exactly once. Cancellation, a43+// runs it once. If that provider request still exceeds its context window, it
37-// one-message transcript, a compaction failure, and a second overflow remain44+// compacts the existing transcript and continues the already-appended prompt
38-// terminal outcomes.45+// exactly once. Cancellation, a compaction failure, and a second overflow
46+// remain terminal outcomes.
39 func PromptWithContextRecovery(ctx context.Context, ag *core.Agent, prompt string, images []provider.ImageBlock, sink func(core.AgentEvent), options ContextRecoveryOptions) (result ContextRecoveryResult, err error) {47 func PromptWithContextRecovery(ctx context.Context, ag *core.Agent, prompt string, images []provider.ImageBlock, sink func(core.AgentEvent), options ContextRecoveryOptions) (result ContextRecoveryResult, err error) {
40 result.OutputStart = len(ag.Messages())48 result.OutputStart = len(ag.Messages())
41 if sink == nil {49 if sink == nil {
42 sink = func(core.AgentEvent) {}50 sink = func(core.AgentEvent) {}
43 }51 }
44 52
53+ usage := ag.LastTurnUsage()
54+ contextUsed := usage.InputTokens + usage.CacheReadTokens + usage.CacheWriteTokens
55+ threshold := NormalizeAutoCompactThreshold(options.AutoCompactThreshold)
56+ proactiveSummaryOverflow := false
57+ if ShouldAutoCompact(contextUsed, options.ContextWindow, threshold) {
58+ messages := ag.Messages()
59+ // A sole transcript message is commonly the worker's only task.
60+ // Replacing it before it has run loses the exact instruction.
61+ if len(messages) > 1 {
62+ if err := ctx.Err(); err != nil {
63+ return result, err
64+ }
65+ compacted, compactErr := compactContext(ctx, ag, messages, options.PersistCompaction)
66+ if compactErr != nil {
67+ if ctxErr := ctx.Err(); ctxErr != nil {
68+ return result, ctxErr
69+ }
70+ if !provider.IsContextOverflowError(compactErr) {
71+ return result, fmt.Errorf("proactively compact transcript: %w", compactErr)
72+ }
73+ // The summary request can itself exceed a provider limit. The
74+ // transcript is unchanged, so make one normal prompt attempt. If
75+ // that also overflows, do not repeat the doomed summary request.
76+ proactiveSummaryOverflow = true
77+ } else {
78+ result.OutputStart = len(compacted)
79+ result.Compacted = true
80+ if err := ctx.Err(); err != nil {
81+ return result, err
82+ }
83+ }
84+ }
85+ }
86+
45 firstSink := sink87 firstSink := sink
46- if options.SuppressInitialOverflowEvent {88+ if options.SuppressInitialOverflowEvent && !proactiveSummaryOverflow {
47 firstSink = func(event core.AgentEvent) {89 firstSink = func(event core.AgentEvent) {
48 if turnEnd, ok := event.(core.EvTurnEnd); ok && provider.IsContextOverflowError(turnEnd.Err) {90 if turnEnd, ok := event.(core.EvTurnEnd); ok && provider.IsContextOverflowError(turnEnd.Err) {
49 return91 return
@@ -52,33 +94,58 @@ func PromptWithContextRecovery(ctx context.Context, ag *core.Agent, prompt strin
52 }94 }
53 }95 }
54 96
55- if err = ag.Prompt(ctx, prompt, images, firstSink); err == nil || ctx.Err() != nil || !provider.IsContextOverflowError(err) {97+ if err = ag.Prompt(ctx, prompt, images, firstSink); err == nil || ctx.Err() != nil || !provider.IsContextOverflowError(err) || proactiveSummaryOverflow {
56 return result, err98 return result, err
57 }99 }
58 100
101+ if ctxErr := ctx.Err(); ctxErr != nil {
102+ return result, ctxErr
103+ }
59 messages := ag.Messages()104 messages := ag.Messages()
60 if len(messages) <= 1 {105 if len(messages) <= 1 {
61 // The failed prompt is the only message. Summarizing it would replace106 // The failed prompt is the only message. Summarizing it would replace
62 // the user's task with an LLM-generated paraphrase.107 // the user's task with an LLM-generated paraphrase.
63 return result, err108 return result, err
64 }109 }
65- keepTail := min(4, len(messages)-1)110+ compacted, compactErr := compactContext(ctx, ag, messages, options.PersistCompaction)
66- if _, compactErr := ag.Compact(ctx, keepTail, nil); compactErr != nil {111+ if compactErr != nil {
67 return result, fmt.Errorf("compact transcript after initial overflow: %w", compactErr)112 return result, fmt.Errorf("compact transcript after initial overflow: %w", compactErr)
68 }113 }
69 114
115+ result.OutputStart = len(compacted)
116+ result.Compacted = true
117+ if ctxErr := ctx.Err(); ctxErr != nil {
118+ return result, ctxErr
119+ }
120+ return result, ag.Continue(ctx, sink)
121+}
122+
123+func compactContext(ctx context.Context, ag *core.Agent, messages []provider.Message, persist func([]provider.Message) error) ([]provider.Message, error) {
124+ if err := ctx.Err(); err != nil {
125+ return nil, err
126+ }
127+ keepTail := min(4, len(messages)-1)
128+ if _, err := ag.Compact(ctx, keepTail, nil); err != nil {
129+ return nil, err
130+ }
131+
70 compacted := ag.Messages()132 compacted := ag.Messages()
71- if options.PersistCompaction != nil {133+ if err := ctx.Err(); err != nil {
72- if persistErr := options.PersistCompaction(compacted); persistErr != nil {134+ ag.SetMessages(messages)
73- // The session did not receive the compaction checkpoint, so restore135+ return nil, err
74- // the original history plus the already-appended prompt rather than136+ }
75- // let a later transcript flush make the session inconsistent.137+ if persist != nil {
138+ if err := ctx.Err(); err != nil {
76 ag.SetMessages(messages)139 ag.SetMessages(messages)
77- return result, fmt.Errorf("persist compacted transcript: %w", persistErr)140+ return nil, err
141+ }
142+ if err := persist(compacted); err != nil {
143+ // The session did not receive the compaction checkpoint. Restore the
144+ // exact pre-compaction transcript so a later suffix flush cannot mix
145+ // the old durable generation with the new in-memory generation.
146+ ag.SetMessages(messages)
147+ return nil, fmt.Errorf("persist compacted transcript: %w", err)
78 }148 }
79 }149 }
80-150+ return compacted, nil
81- result.OutputStart = len(compacted)
82- result.Compacted = true
83- return result, ag.Continue(ctx, sink)
84 }151 }
M packages/agent/modes/context_recovery_test.go +307 -0
@@ -6,6 +6,8 @@ import (
66 "context"
77 "encoding/json"
88 "errors"
9+ "reflect"
10+ "strings"
911 "sync"
1012 "testing"
1113
@@ -22,6 +24,311 @@ type hostRecoveryClient struct {
2224
2325 func (c *hostRecoveryClient) Name() string { return "host-context-recovery-test" }
2426
27+type proactiveContextClient struct {
28+ mu sync.Mutex
29+ requests []provider.Request
30+ summaryErr error
31+ onSummary func()
32+ normalOverflow bool
33+}
34+
35+func (c *proactiveContextClient) Name() string { return "proactive-context-test" }
36+
37+func (c *proactiveContextClient) Stream(_ context.Context, req provider.Request) (<-chan provider.Event, error) {
38+ c.mu.Lock()
39+ c.requests = append(c.requests, req)
40+ c.mu.Unlock()
41+
42+ events := make(chan provider.Event, 2)
43+ go func() {
44+ defer close(events)
45+ if strings.Contains(req.System, "context summarization assistant") {
46+ if c.onSummary != nil {
47+ c.onSummary()
48+ }
49+ if c.summaryErr != nil {
50+ events <- provider.EventDone{Stop: provider.StopError, Err: c.summaryErr}
51+ return
52+ }
53+ events <- provider.EventTextDelta{Delta: "condensed history"}
54+ events <- provider.EventDone{Stop: provider.StopEnd}
55+ return
56+ }
57+ if c.normalOverflow {
58+ events <- provider.EventDone{Stop: provider.StopError, Err: errors.New("context window exceeded")}
59+ return
60+ }
61+ events <- provider.EventDone{Stop: provider.StopEnd, Message: provider.Message{
62+ Role: provider.RoleAssistant,
63+ Content: []provider.Content{provider.TextBlock{Text: "completed"}},
64+ }}
65+ }()
66+ return events, nil
67+}
68+
69+func TestPromptWithContextRecoveryProactivelyCompactsCacheOnlyUsage(t *testing.T) {
70+ client := &proactiveContextClient{}
71+ agent := core.NewAgent(client, "model", "system", nil)
72+ history := []provider.Message{
73+ {Role: provider.RoleUser, Content: []provider.Content{provider.TextBlock{Text: "one"}}},
74+ {Role: provider.RoleAssistant, Content: []provider.Content{provider.TextBlock{Text: "two"}}},
75+ {Role: provider.RoleUser, Content: []provider.Content{provider.TextBlock{Text: "three"}}},
76+ {Role: provider.RoleAssistant, Content: []provider.Content{provider.TextBlock{Text: "four"}}},
77+ {Role: provider.RoleUser, Content: []provider.Content{provider.TextBlock{Text: "five"}}},
78+ {Role: provider.RoleAssistant, Content: []provider.Content{provider.TextBlock{Text: "six"}}},
79+ }
80+ agent.SetMessages(history)
81+ agent.SeedLastTurnUsage(provider.Usage{CacheReadTokens: 85})
82+
83+ var persisted [][]provider.Message
84+ result, err := PromptWithContextRecovery(context.Background(), agent, "continue", nil, nil, ContextRecoveryOptions{
85+ ContextWindow: 100,
86+ PersistCompaction: func(messages []provider.Message) error {
87+ persisted = append(persisted, messages)
88+ return nil
89+ },
90+ })
91+ if err != nil {
92+ t.Fatalf("PromptWithContextRecovery: %v", err)
93+ }
94+ if !result.Compacted || len(persisted) != 1 {
95+ t.Fatalf("compacted/persisted = %t/%d, want true/1", result.Compacted, len(persisted))
96+ }
97+ if got := recoveredAssistantText(agent.Messages()[result.OutputStart:]); got != "completed" {
98+ t.Fatalf("output after compacted boundary = %q, want completed", got)
99+ }
100+
101+ client.mu.Lock()
102+ requests := append([]provider.Request(nil), client.requests...)
103+ client.mu.Unlock()
104+ if len(requests) != 2 {
105+ t.Fatalf("provider requests = %d, want compaction then prompt", len(requests))
106+ }
107+ if !recoveryRequestContainsText(requests[0], "<conversation>") || !recoveryRequestContainsText(requests[0], "one") {
108+ t.Fatalf("first request was not compaction over old history: %#v", requests[0])
109+ }
110+ if got := requestUserTextCount(requests[1], "continue"); got != 1 {
111+ t.Fatalf("prompt count after proactive compaction = %d, want 1", got)
112+ }
113+}
114+
115+func TestPromptWithContextRecoveryHonorsDisabledThreshold(t *testing.T) {
116+ client := &proactiveContextClient{}
117+ agent := core.NewAgent(client, "model", "system", nil)
118+ agent.SetMessages([]provider.Message{{Role: provider.RoleUser, Content: []provider.Content{provider.TextBlock{Text: "history"}}}})
119+ agent.SeedLastTurnUsage(provider.Usage{InputTokens: 100, CacheReadTokens: 100})
120+ disabled := 0
121+
122+ result, err := PromptWithContextRecovery(context.Background(), agent, "continue", nil, nil, ContextRecoveryOptions{
123+ ContextWindow: 100,
124+ AutoCompactThreshold: &disabled,
125+ })
126+ if err != nil {
127+ t.Fatalf("PromptWithContextRecovery: %v", err)
128+ }
129+ if result.Compacted {
130+ t.Fatal("disabled threshold unexpectedly compacted")
131+ }
132+ client.mu.Lock()
133+ requests := append([]provider.Request(nil), client.requests...)
134+ client.mu.Unlock()
135+ if len(requests) != 1 || requestUserTextCount(requests[0], "continue") != 1 {
136+ t.Fatalf("requests with disabled threshold = %#v, want one direct prompt", requests)
137+ }
138+}
139+
140+func TestPromptWithContextRecoveryRestoresProactiveCompactionOnPersistenceFailure(t *testing.T) {
141+ client := &proactiveContextClient{}
142+ agent := core.NewAgent(client, "model", "system", nil)
143+ history := []provider.Message{
144+ {Role: provider.RoleUser, Content: []provider.Content{provider.TextBlock{Text: "original"}}},
145+ {Role: provider.RoleAssistant, Content: []provider.Content{provider.TextBlock{Text: "answer"}}},
146+ }
147+ agent.SetMessages(history)
148+ agent.SeedLastTurnUsage(provider.Usage{CacheWriteTokens: 90})
149+
150+ _, err := PromptWithContextRecovery(context.Background(), agent, "must not send", nil, nil, ContextRecoveryOptions{
151+ ContextWindow: 100,
152+ PersistCompaction: func([]provider.Message) error {
153+ return errors.New("disk full")
154+ },
155+ })
156+ if err == nil || !strings.Contains(err.Error(), "persist compacted transcript: disk full") {
157+ t.Fatalf("error = %v, want persistence failure", err)
158+ }
159+ if got := agent.Messages(); !reflect.DeepEqual(got, history) {
160+ t.Fatalf("transcript after persistence failure = %#v, want %#v", got, history)
161+ }
162+ client.mu.Lock()
163+ requestCount := len(client.requests)
164+ client.mu.Unlock()
165+ if requestCount != 1 {
166+ t.Fatalf("provider requests = %d, want compaction only", requestCount)
167+ }
168+}
169+
170+func TestPromptWithContextRecoveryDoesNotProactivelyCompactSoleTask(t *testing.T) {
171+ client := &proactiveContextClient{}
172+ agent := core.NewAgent(client, "model", "system", nil)
173+ task := provider.Message{Role: provider.RoleUser, Content: []provider.Content{provider.TextBlock{Text: "exact sole task"}}}
174+ agent.SetMessages([]provider.Message{task})
175+ agent.SeedLastTurnUsage(provider.Usage{CacheReadTokens: 95})
176+
177+ result, err := PromptWithContextRecovery(context.Background(), agent, "continue", nil, nil, ContextRecoveryOptions{ContextWindow: 100})
178+ if err != nil {
179+ t.Fatalf("PromptWithContextRecovery: %v", err)
180+ }
181+ if result.Compacted {
182+ t.Fatal("sole task was proactively compacted")
183+ }
184+ client.mu.Lock()
185+ requests := append([]provider.Request(nil), client.requests...)
186+ client.mu.Unlock()
187+ if len(requests) != 1 || !recoveryRequestContainsText(requests[0], "exact sole task") {
188+ t.Fatalf("requests = %#v, want one direct request retaining the sole task", requests)
189+ }
190+}
191+
192+func TestPromptWithContextRecoveryFallsBackWhenProactiveSummaryOverflows(t *testing.T) {
193+ client := &proactiveContextClient{summaryErr: errors.New("context window exceeded while summarizing")}
194+ agent := core.NewAgent(client, "model", "system", nil)
195+ history := []provider.Message{
196+ {Role: provider.RoleUser, Content: []provider.Content{provider.TextBlock{Text: "history"}}},
197+ {Role: provider.RoleAssistant, Content: []provider.Content{provider.TextBlock{Text: "answer"}}},
198+ }
199+ agent.SetMessages(history)
200+ agent.SeedLastTurnUsage(provider.Usage{InputTokens: 90})
201+
202+ result, err := PromptWithContextRecovery(context.Background(), agent, "same task", nil, nil, ContextRecoveryOptions{ContextWindow: 100})
203+ if err != nil {
204+ t.Fatalf("PromptWithContextRecovery: %v", err)
205+ }
206+ if result.Compacted {
207+ t.Fatal("overflowing proactive summary reported compaction")
208+ }
209+ client.mu.Lock()
210+ requests := append([]provider.Request(nil), client.requests...)
211+ client.mu.Unlock()
212+ if len(requests) != 2 {
213+ t.Fatalf("provider requests = %d, want one summary and one normal prompt", len(requests))
214+ }
215+ if got := requestUserTextCount(requests[1], "same task"); got != 1 {
216+ t.Fatalf("normal recovery prompt count = %d, want 1", got)
217+ }
218+}
219+
220+func TestPromptWithContextRecoveryStopsBeforeContinuationWhenPersistenceCancels(t *testing.T) {
221+ ctx, cancel := context.WithCancel(context.Background())
222+ client := &proactiveContextClient{}
223+ agent := core.NewAgent(client, "model", "system", nil)
224+ history := []provider.Message{
225+ {Role: provider.RoleUser, Content: []provider.Content{provider.TextBlock{Text: "history"}}},
226+ {Role: provider.RoleAssistant, Content: []provider.Content{provider.TextBlock{Text: "answer"}}},
227+ }
228+ agent.SetMessages(history)
229+ agent.SeedLastTurnUsage(provider.Usage{CacheWriteTokens: 90})
230+ persisted := 0
231+
232+ result, err := PromptWithContextRecovery(ctx, agent, "must not run", nil, nil, ContextRecoveryOptions{
233+ ContextWindow: 100,
234+ PersistCompaction: func([]provider.Message) error {
235+ persisted++
236+ cancel()
237+ return nil
238+ },
239+ })
240+ if !errors.Is(err, context.Canceled) {
241+ t.Fatalf("error = %v, want context cancellation", err)
242+ }
243+ if !result.Compacted || persisted != 1 {
244+ t.Fatalf("compaction result/persistence = %v/%d, want true/1", result.Compacted, persisted)
245+ }
246+ client.mu.Lock()
247+ requestCount := len(client.requests)
248+ client.mu.Unlock()
249+ if requestCount != 1 {
250+ t.Fatalf("provider requests = %d, want summary only", requestCount)
251+ }
252+}
253+
254+func TestPromptWithContextRecoveryRollsBackWhenCanceledAfterSummary(t *testing.T) {
255+ ctx, cancel := context.WithCancel(context.Background())
256+ client := &proactiveContextClient{onSummary: cancel}
257+ agent := core.NewAgent(client, "model", "system", nil)
258+ history := []provider.Message{
259+ {Role: provider.RoleUser, Content: []provider.Content{provider.TextBlock{Text: "history"}}},
260+ {Role: provider.RoleAssistant, Content: []provider.Content{provider.TextBlock{Text: "answer"}}},
261+ }
262+ agent.SetMessages(history)
263+ agent.SeedLastTurnUsage(provider.Usage{CacheWriteTokens: 90})
264+ persisted := 0
265+
266+ _, err := PromptWithContextRecovery(ctx, agent, "must not run", nil, nil, ContextRecoveryOptions{
267+ ContextWindow: 100,
268+ PersistCompaction: func([]provider.Message) error {
269+ persisted++
270+ return nil
271+ },
272+ })
273+ if !errors.Is(err, context.Canceled) {
274+ t.Fatalf("error = %v, want context cancellation", err)
275+ }
276+ if persisted != 0 {
277+ t.Fatalf("persistence calls = %d, want 0", persisted)
278+ }
279+ if got := agent.Messages(); !reflect.DeepEqual(got, history) {
280+ t.Fatalf("transcript after cancellation = %#v, want %#v", got, history)
281+ }
282+ client.mu.Lock()
283+ requestCount := len(client.requests)
284+ client.mu.Unlock()
285+ if requestCount != 1 {
286+ t.Fatalf("provider requests = %d, want summary only", requestCount)
287+ }
288+}
289+
290+func TestPromptWithContextRecoveryRepairsToolPairAtCompactionBoundary(t *testing.T) {
291+ client := &proactiveContextClient{}
292+ agent := core.NewAgent(client, "model", "system", nil)
293+ agent.SetMessages([]provider.Message{
294+ {Role: provider.RoleUser, Content: []provider.Content{provider.TextBlock{Text: "old"}}},
295+ {Role: provider.RoleAssistant, Content: []provider.Content{provider.ToolCallBlock{ID: "split-call", Name: "read"}}},
296+ {Role: provider.RoleTool, Content: []provider.Content{provider.ToolResultBlock{CallID: "split-call", Content: []provider.Content{provider.TextBlock{Text: "result"}}}}},
297+ {Role: provider.RoleAssistant, Content: []provider.Content{provider.TextBlock{Text: "after tool"}}},
298+ {Role: provider.RoleUser, Content: []provider.Content{provider.TextBlock{Text: "next"}}},
299+ {Role: provider.RoleAssistant, Content: []provider.Content{provider.TextBlock{Text: "ready"}}},
300+ })
301+ agent.SeedLastTurnUsage(provider.Usage{InputTokens: 90})
302+
303+ if _, err := PromptWithContextRecovery(context.Background(), agent, "continue", nil, nil, ContextRecoveryOptions{ContextWindow: 100}); err != nil {
304+ t.Fatalf("PromptWithContextRecovery: %v", err)
305+ }
306+ client.mu.Lock()
307+ requests := append([]provider.Request(nil), client.requests...)
308+ client.mu.Unlock()
309+ if len(requests) != 2 {
310+ t.Fatalf("provider requests = %d, want summary then continuation", len(requests))
311+ }
312+ for _, message := range requests[1].Messages {
313+ for _, content := range message.Content {
314+ if result, ok := content.(provider.ToolResultBlock); ok && result.CallID == "split-call" {
315+ t.Fatalf("continuation contains orphaned tool result: %#v", requests[1].Messages)
316+ }
317+ }
318+ }
319+}
320+
321+func recoveryRequestContainsText(req provider.Request, want string) bool {
322+ for _, message := range req.Messages {
323+ for _, content := range message.Content {
324+ if text, ok := content.(provider.TextBlock); ok && strings.Contains(text.Text, want) {
325+ return true
326+ }
327+ }
328+ }
329+ return false
330+}
331+
25332 func (c *hostRecoveryClient) Stream(_ context.Context, req provider.Request) (<-chan provider.Event, error) {
26333 c.mu.Lock()
27334 c.calls++
@@ -6,6 +6,8 @@ import (
6 "context"6 "context"
7 "encoding/json"7 "encoding/json"
8 "errors"8 "errors"
9+ "reflect"
10+ "strings"
9 "sync"11 "sync"
10 "testing"12 "testing"
11 13
@@ -22,6 +24,311 @@ type hostRecoveryClient struct {
22 24
23 func (c *hostRecoveryClient) Name() string { return "host-context-recovery-test" }25 func (c *hostRecoveryClient) Name() string { return "host-context-recovery-test" }
24 26
27+type proactiveContextClient struct {
28+ mu sync.Mutex
29+ requests []provider.Request
30+ summaryErr error
31+ onSummary func()
32+ normalOverflow bool
33+}
34+
35+func (c *proactiveContextClient) Name() string { return "proactive-context-test" }
36+
37+func (c *proactiveContextClient) Stream(_ context.Context, req provider.Request) (<-chan provider.Event, error) {
38+ c.mu.Lock()
39+ c.requests = append(c.requests, req)
40+ c.mu.Unlock()
41+
42+ events := make(chan provider.Event, 2)
43+ go func() {
44+ defer close(events)
45+ if strings.Contains(req.System, "context summarization assistant") {
46+ if c.onSummary != nil {
47+ c.onSummary()
48+ }
49+ if c.summaryErr != nil {
50+ events <- provider.EventDone{Stop: provider.StopError, Err: c.summaryErr}
51+ return
52+ }
53+ events <- provider.EventTextDelta{Delta: "condensed history"}
54+ events <- provider.EventDone{Stop: provider.StopEnd}
55+ return
56+ }
57+ if c.normalOverflow {
58+ events <- provider.EventDone{Stop: provider.StopError, Err: errors.New("context window exceeded")}
59+ return
60+ }
61+ events <- provider.EventDone{Stop: provider.StopEnd, Message: provider.Message{
62+ Role: provider.RoleAssistant,
63+ Content: []provider.Content{provider.TextBlock{Text: "completed"}},
64+ }}
65+ }()
66+ return events, nil
67+}
68+
69+func TestPromptWithContextRecoveryProactivelyCompactsCacheOnlyUsage(t *testing.T) {
70+ client := &proactiveContextClient{}
71+ agent := core.NewAgent(client, "model", "system", nil)
72+ history := []provider.Message{
73+ {Role: provider.RoleUser, Content: []provider.Content{provider.TextBlock{Text: "one"}}},
74+ {Role: provider.RoleAssistant, Content: []provider.Content{provider.TextBlock{Text: "two"}}},
75+ {Role: provider.RoleUser, Content: []provider.Content{provider.TextBlock{Text: "three"}}},
76+ {Role: provider.RoleAssistant, Content: []provider.Content{provider.TextBlock{Text: "four"}}},
77+ {Role: provider.RoleUser, Content: []provider.Content{provider.TextBlock{Text: "five"}}},
78+ {Role: provider.RoleAssistant, Content: []provider.Content{provider.TextBlock{Text: "six"}}},
79+ }
80+ agent.SetMessages(history)
81+ agent.SeedLastTurnUsage(provider.Usage{CacheReadTokens: 85})
82+
83+ var persisted [][]provider.Message
84+ result, err := PromptWithContextRecovery(context.Background(), agent, "continue", nil, nil, ContextRecoveryOptions{
85+ ContextWindow: 100,
86+ PersistCompaction: func(messages []provider.Message) error {
87+ persisted = append(persisted, messages)
88+ return nil
89+ },
90+ })
91+ if err != nil {
92+ t.Fatalf("PromptWithContextRecovery: %v", err)
93+ }
94+ if !result.Compacted || len(persisted) != 1 {
95+ t.Fatalf("compacted/persisted = %t/%d, want true/1", result.Compacted, len(persisted))
96+ }
97+ if got := recoveredAssistantText(agent.Messages()[result.OutputStart:]); got != "completed" {
98+ t.Fatalf("output after compacted boundary = %q, want completed", got)
99+ }
100+
101+ client.mu.Lock()
102+ requests := append([]provider.Request(nil), client.requests...)
103+ client.mu.Unlock()
104+ if len(requests) != 2 {
105+ t.Fatalf("provider requests = %d, want compaction then prompt", len(requests))
106+ }
107+ if !recoveryRequestContainsText(requests[0], "<conversation>") || !recoveryRequestContainsText(requests[0], "one") {
108+ t.Fatalf("first request was not compaction over old history: %#v", requests[0])
109+ }
110+ if got := requestUserTextCount(requests[1], "continue"); got != 1 {
111+ t.Fatalf("prompt count after proactive compaction = %d, want 1", got)
112+ }
113+}
114+
115+func TestPromptWithContextRecoveryHonorsDisabledThreshold(t *testing.T) {
116+ client := &proactiveContextClient{}
117+ agent := core.NewAgent(client, "model", "system", nil)
118+ agent.SetMessages([]provider.Message{{Role: provider.RoleUser, Content: []provider.Content{provider.TextBlock{Text: "history"}}}})
119+ agent.SeedLastTurnUsage(provider.Usage{InputTokens: 100, CacheReadTokens: 100})
120+ disabled := 0
121+
122+ result, err := PromptWithContextRecovery(context.Background(), agent, "continue", nil, nil, ContextRecoveryOptions{
123+ ContextWindow: 100,
124+ AutoCompactThreshold: &disabled,
125+ })
126+ if err != nil {
127+ t.Fatalf("PromptWithContextRecovery: %v", err)
128+ }
129+ if result.Compacted {
130+ t.Fatal("disabled threshold unexpectedly compacted")
131+ }
132+ client.mu.Lock()
133+ requests := append([]provider.Request(nil), client.requests...)
134+ client.mu.Unlock()
135+ if len(requests) != 1 || requestUserTextCount(requests[0], "continue") != 1 {
136+ t.Fatalf("requests with disabled threshold = %#v, want one direct prompt", requests)
137+ }
138+}
139+
140+func TestPromptWithContextRecoveryRestoresProactiveCompactionOnPersistenceFailure(t *testing.T) {
141+ client := &proactiveContextClient{}
142+ agent := core.NewAgent(client, "model", "system", nil)
143+ history := []provider.Message{
144+ {Role: provider.RoleUser, Content: []provider.Content{provider.TextBlock{Text: "original"}}},
145+ {Role: provider.RoleAssistant, Content: []provider.Content{provider.TextBlock{Text: "answer"}}},
146+ }
147+ agent.SetMessages(history)
148+ agent.SeedLastTurnUsage(provider.Usage{CacheWriteTokens: 90})
149+
150+ _, err := PromptWithContextRecovery(context.Background(), agent, "must not send", nil, nil, ContextRecoveryOptions{
151+ ContextWindow: 100,
152+ PersistCompaction: func([]provider.Message) error {
153+ return errors.New("disk full")
154+ },
155+ })
156+ if err == nil || !strings.Contains(err.Error(), "persist compacted transcript: disk full") {
157+ t.Fatalf("error = %v, want persistence failure", err)
158+ }
159+ if got := agent.Messages(); !reflect.DeepEqual(got, history) {
160+ t.Fatalf("transcript after persistence failure = %#v, want %#v", got, history)
161+ }
162+ client.mu.Lock()
163+ requestCount := len(client.requests)
164+ client.mu.Unlock()
165+ if requestCount != 1 {
166+ t.Fatalf("provider requests = %d, want compaction only", requestCount)
167+ }
168+}
169+
170+func TestPromptWithContextRecoveryDoesNotProactivelyCompactSoleTask(t *testing.T) {
171+ client := &proactiveContextClient{}
172+ agent := core.NewAgent(client, "model", "system", nil)
173+ task := provider.Message{Role: provider.RoleUser, Content: []provider.Content{provider.TextBlock{Text: "exact sole task"}}}
174+ agent.SetMessages([]provider.Message{task})
175+ agent.SeedLastTurnUsage(provider.Usage{CacheReadTokens: 95})
176+
177+ result, err := PromptWithContextRecovery(context.Background(), agent, "continue", nil, nil, ContextRecoveryOptions{ContextWindow: 100})
178+ if err != nil {
179+ t.Fatalf("PromptWithContextRecovery: %v", err)
180+ }
181+ if result.Compacted {
182+ t.Fatal("sole task was proactively compacted")
183+ }
184+ client.mu.Lock()
185+ requests := append([]provider.Request(nil), client.requests...)
186+ client.mu.Unlock()
187+ if len(requests) != 1 || !recoveryRequestContainsText(requests[0], "exact sole task") {
188+ t.Fatalf("requests = %#v, want one direct request retaining the sole task", requests)
189+ }
190+}
191+
192+func TestPromptWithContextRecoveryFallsBackWhenProactiveSummaryOverflows(t *testing.T) {
193+ client := &proactiveContextClient{summaryErr: errors.New("context window exceeded while summarizing")}
194+ agent := core.NewAgent(client, "model", "system", nil)
195+ history := []provider.Message{
196+ {Role: provider.RoleUser, Content: []provider.Content{provider.TextBlock{Text: "history"}}},
197+ {Role: provider.RoleAssistant, Content: []provider.Content{provider.TextBlock{Text: "answer"}}},
198+ }
199+ agent.SetMessages(history)
200+ agent.SeedLastTurnUsage(provider.Usage{InputTokens: 90})
201+
202+ result, err := PromptWithContextRecovery(context.Background(), agent, "same task", nil, nil, ContextRecoveryOptions{ContextWindow: 100})
203+ if err != nil {
204+ t.Fatalf("PromptWithContextRecovery: %v", err)
205+ }
206+ if result.Compacted {
207+ t.Fatal("overflowing proactive summary reported compaction")
208+ }
209+ client.mu.Lock()
210+ requests := append([]provider.Request(nil), client.requests...)
211+ client.mu.Unlock()
212+ if len(requests) != 2 {
213+ t.Fatalf("provider requests = %d, want one summary and one normal prompt", len(requests))
214+ }
215+ if got := requestUserTextCount(requests[1], "same task"); got != 1 {
216+ t.Fatalf("normal recovery prompt count = %d, want 1", got)
217+ }
218+}
219+
220+func TestPromptWithContextRecoveryStopsBeforeContinuationWhenPersistenceCancels(t *testing.T) {
221+ ctx, cancel := context.WithCancel(context.Background())
222+ client := &proactiveContextClient{}
223+ agent := core.NewAgent(client, "model", "system", nil)
224+ history := []provider.Message{
225+ {Role: provider.RoleUser, Content: []provider.Content{provider.TextBlock{Text: "history"}}},
226+ {Role: provider.RoleAssistant, Content: []provider.Content{provider.TextBlock{Text: "answer"}}},
227+ }
228+ agent.SetMessages(history)
229+ agent.SeedLastTurnUsage(provider.Usage{CacheWriteTokens: 90})
230+ persisted := 0
231+
232+ result, err := PromptWithContextRecovery(ctx, agent, "must not run", nil, nil, ContextRecoveryOptions{
233+ ContextWindow: 100,
234+ PersistCompaction: func([]provider.Message) error {
235+ persisted++
236+ cancel()
237+ return nil
238+ },
239+ })
240+ if !errors.Is(err, context.Canceled) {
241+ t.Fatalf("error = %v, want context cancellation", err)
242+ }
243+ if !result.Compacted || persisted != 1 {
244+ t.Fatalf("compaction result/persistence = %v/%d, want true/1", result.Compacted, persisted)
245+ }
246+ client.mu.Lock()
247+ requestCount := len(client.requests)
248+ client.mu.Unlock()
249+ if requestCount != 1 {
250+ t.Fatalf("provider requests = %d, want summary only", requestCount)
251+ }
252+}
253+
254+func TestPromptWithContextRecoveryRollsBackWhenCanceledAfterSummary(t *testing.T) {
255+ ctx, cancel := context.WithCancel(context.Background())
256+ client := &proactiveContextClient{onSummary: cancel}
257+ agent := core.NewAgent(client, "model", "system", nil)
258+ history := []provider.Message{
259+ {Role: provider.RoleUser, Content: []provider.Content{provider.TextBlock{Text: "history"}}},
260+ {Role: provider.RoleAssistant, Content: []provider.Content{provider.TextBlock{Text: "answer"}}},
261+ }
262+ agent.SetMessages(history)
263+ agent.SeedLastTurnUsage(provider.Usage{CacheWriteTokens: 90})
264+ persisted := 0
265+
266+ _, err := PromptWithContextRecovery(ctx, agent, "must not run", nil, nil, ContextRecoveryOptions{
267+ ContextWindow: 100,
268+ PersistCompaction: func([]provider.Message) error {
269+ persisted++
270+ return nil
271+ },
272+ })
273+ if !errors.Is(err, context.Canceled) {
274+ t.Fatalf("error = %v, want context cancellation", err)
275+ }
276+ if persisted != 0 {
277+ t.Fatalf("persistence calls = %d, want 0", persisted)
278+ }
279+ if got := agent.Messages(); !reflect.DeepEqual(got, history) {
280+ t.Fatalf("transcript after cancellation = %#v, want %#v", got, history)
281+ }
282+ client.mu.Lock()
283+ requestCount := len(client.requests)
284+ client.mu.Unlock()
285+ if requestCount != 1 {
286+ t.Fatalf("provider requests = %d, want summary only", requestCount)
287+ }
288+}
289+
290+func TestPromptWithContextRecoveryRepairsToolPairAtCompactionBoundary(t *testing.T) {
291+ client := &proactiveContextClient{}
292+ agent := core.NewAgent(client, "model", "system", nil)
293+ agent.SetMessages([]provider.Message{
294+ {Role: provider.RoleUser, Content: []provider.Content{provider.TextBlock{Text: "old"}}},
295+ {Role: provider.RoleAssistant, Content: []provider.Content{provider.ToolCallBlock{ID: "split-call", Name: "read"}}},
296+ {Role: provider.RoleTool, Content: []provider.Content{provider.ToolResultBlock{CallID: "split-call", Content: []provider.Content{provider.TextBlock{Text: "result"}}}}},
297+ {Role: provider.RoleAssistant, Content: []provider.Content{provider.TextBlock{Text: "after tool"}}},
298+ {Role: provider.RoleUser, Content: []provider.Content{provider.TextBlock{Text: "next"}}},
299+ {Role: provider.RoleAssistant, Content: []provider.Content{provider.TextBlock{Text: "ready"}}},
300+ })
301+ agent.SeedLastTurnUsage(provider.Usage{InputTokens: 90})
302+
303+ if _, err := PromptWithContextRecovery(context.Background(), agent, "continue", nil, nil, ContextRecoveryOptions{ContextWindow: 100}); err != nil {
304+ t.Fatalf("PromptWithContextRecovery: %v", err)
305+ }
306+ client.mu.Lock()
307+ requests := append([]provider.Request(nil), client.requests...)
308+ client.mu.Unlock()
309+ if len(requests) != 2 {
310+ t.Fatalf("provider requests = %d, want summary then continuation", len(requests))
311+ }
312+ for _, message := range requests[1].Messages {
313+ for _, content := range message.Content {
314+ if result, ok := content.(provider.ToolResultBlock); ok && result.CallID == "split-call" {
315+ t.Fatalf("continuation contains orphaned tool result: %#v", requests[1].Messages)
316+ }
317+ }
318+ }
319+}
320+
321+func recoveryRequestContainsText(req provider.Request, want string) bool {
322+ for _, message := range req.Messages {
323+ for _, content := range message.Content {
324+ if text, ok := content.(provider.TextBlock); ok && strings.Contains(text.Text, want) {
325+ return true
326+ }
327+ }
328+ }
329+ return false
330+}
331+
25 func (c *hostRecoveryClient) Stream(_ context.Context, req provider.Request) (<-chan provider.Event, error) {332 func (c *hostRecoveryClient) Stream(_ context.Context, req provider.Request) (<-chan provider.Event, error) {
26 c.mu.Lock()333 c.mu.Lock()
27 c.calls++334 c.calls++
M packages/agent/modes/interactive.go +12 -8
@@ -4494,7 +4494,7 @@ func (i *Interactive) openSettingsDialog() {
44944494 {value: "90", label: "90%", desc: "retain more history before compacting"},
44954495 {value: "95", label: "95%", desc: "retain most history before compacting"},
44964496 }
4497- autoCompactThreshold := normalizeAutoCompactThreshold(i.cfg.AutoCompactThreshold)
4497+ autoCompactThreshold := NormalizeAutoCompactThreshold(i.cfg.AutoCompactThreshold)
44984498 autoCompactChoice := 0
44994499 for idx, opt := range autoCompactOptions {
45004500 if opt.value == strconv.Itoa(autoCompactThreshold) {
@@ -4762,7 +4762,7 @@ func (i *Interactive) applyAutoCompactThresholdSetting(value string) {
47624762 if err != nil {
47634763 return
47644764 }
4765- threshold = normalizeAutoCompactThreshold(&threshold)
4765+ threshold = NormalizeAutoCompactThreshold(&threshold)
47664766 if store, ok := i.cfg.SettingsStore.(autoCompactThresholdSettingsStore); ok {
47674767 if err := store.SetAutoCompactThreshold(threshold); err != nil {
47684768 i.mu.Lock()
@@ -7930,9 +7930,11 @@ func autoCompactNoteLine(th tui.Theme, msg string) string {
79307930 return " " + th.FGColor(th.Warning, "⚠ "+msg)
79317931 }
79327932
7933-const defaultAutoCompactThreshold = 85
7934-
7935-func normalizeAutoCompactThreshold(threshold *int) int {
7933+// NormalizeAutoCompactThreshold applies the persisted auto-compaction
7934+// setting's supported values and default. It is exported so non-interactive
7935+// hosts can share the interactive mode's exact configuration semantics.
7936+func NormalizeAutoCompactThreshold(threshold *int) int {
7937+ const defaultAutoCompactThreshold = 85
79367938 if threshold == nil {
79377939 return defaultAutoCompactThreshold
79387940 }
@@ -7944,7 +7946,9 @@ func normalizeAutoCompactThreshold(threshold *int) int {
79447946 }
79457947 }
79467948
7947-func shouldAutoCompact(inputTokens, contextWindow, thresholdPercent int) bool {
7949+// ShouldAutoCompact reports whether the latest request context reached the
7950+// configured percentage of the model's advertised context window.
7951+func ShouldAutoCompact(inputTokens, contextWindow, thresholdPercent int) bool {
79487952 if inputTokens <= 0 || contextWindow <= 0 || thresholdPercent <= 0 {
79497953 return false
79507954 }
@@ -8094,8 +8098,8 @@ func (i *Interactive) shouldAutoCompactLocked() bool {
80948098 if err != nil || m.ContextWindow <= 0 {
80958099 return false
80968100 }
8097- threshold := normalizeAutoCompactThreshold(i.cfg.AutoCompactThreshold)
8098- return shouldAutoCompact(i.lastCtxInput, m.ContextWindow, threshold)
8101+ threshold := NormalizeAutoCompactThreshold(i.cfg.AutoCompactThreshold)
8102+ return ShouldAutoCompact(i.lastCtxInput, m.ContextWindow, threshold)
80998103 }
81008104
81018105 func eventAffectsPresentation(ev core.AgentEvent) bool {
@@ -4494,7 +4494,7 @@ func (i *Interactive) openSettingsDialog() {
4494 {value: "90", label: "90%", desc: "retain more history before compacting"},4494 {value: "90", label: "90%", desc: "retain more history before compacting"},
4495 {value: "95", label: "95%", desc: "retain most history before compacting"},4495 {value: "95", label: "95%", desc: "retain most history before compacting"},
4496 }4496 }
4497- autoCompactThreshold := normalizeAutoCompactThreshold(i.cfg.AutoCompactThreshold)4497+ autoCompactThreshold := NormalizeAutoCompactThreshold(i.cfg.AutoCompactThreshold)
4498 autoCompactChoice := 04498 autoCompactChoice := 0
4499 for idx, opt := range autoCompactOptions {4499 for idx, opt := range autoCompactOptions {
4500 if opt.value == strconv.Itoa(autoCompactThreshold) {4500 if opt.value == strconv.Itoa(autoCompactThreshold) {
@@ -4762,7 +4762,7 @@ func (i *Interactive) applyAutoCompactThresholdSetting(value string) {
4762 if err != nil {4762 if err != nil {
4763 return4763 return
4764 }4764 }
4765- threshold = normalizeAutoCompactThreshold(&threshold)4765+ threshold = NormalizeAutoCompactThreshold(&threshold)
4766 if store, ok := i.cfg.SettingsStore.(autoCompactThresholdSettingsStore); ok {4766 if store, ok := i.cfg.SettingsStore.(autoCompactThresholdSettingsStore); ok {
4767 if err := store.SetAutoCompactThreshold(threshold); err != nil {4767 if err := store.SetAutoCompactThreshold(threshold); err != nil {
4768 i.mu.Lock()4768 i.mu.Lock()
@@ -7930,9 +7930,11 @@ func autoCompactNoteLine(th tui.Theme, msg string) string {
7930 return " " + th.FGColor(th.Warning, "⚠ "+msg)7930 return " " + th.FGColor(th.Warning, "⚠ "+msg)
7931 }7931 }
7932 7932
7933-const defaultAutoCompactThreshold = 857933+// NormalizeAutoCompactThreshold applies the persisted auto-compaction
7934-7934+// setting's supported values and default. It is exported so non-interactive
7935-func normalizeAutoCompactThreshold(threshold *int) int {7935+// hosts can share the interactive mode's exact configuration semantics.
7936+func NormalizeAutoCompactThreshold(threshold *int) int {
7937+ const defaultAutoCompactThreshold = 85
7936 if threshold == nil {7938 if threshold == nil {
7937 return defaultAutoCompactThreshold7939 return defaultAutoCompactThreshold
7938 }7940 }
@@ -7944,7 +7946,9 @@ func normalizeAutoCompactThreshold(threshold *int) int {
7944 }7946 }
7945 }7947 }
7946 7948
7947-func shouldAutoCompact(inputTokens, contextWindow, thresholdPercent int) bool {7949+// ShouldAutoCompact reports whether the latest request context reached the
7950+// configured percentage of the model's advertised context window.
7951+func ShouldAutoCompact(inputTokens, contextWindow, thresholdPercent int) bool {
7948 if inputTokens <= 0 || contextWindow <= 0 || thresholdPercent <= 0 {7952 if inputTokens <= 0 || contextWindow <= 0 || thresholdPercent <= 0 {
7949 return false7953 return false
7950 }7954 }
@@ -8094,8 +8098,8 @@ func (i *Interactive) shouldAutoCompactLocked() bool {
8094 if err != nil || m.ContextWindow <= 0 {8098 if err != nil || m.ContextWindow <= 0 {
8095 return false8099 return false
8096 }8100 }
8097- threshold := normalizeAutoCompactThreshold(i.cfg.AutoCompactThreshold)8101+ threshold := NormalizeAutoCompactThreshold(i.cfg.AutoCompactThreshold)
8098- return shouldAutoCompact(i.lastCtxInput, m.ContextWindow, threshold)8102+ return ShouldAutoCompact(i.lastCtxInput, m.ContextWindow, threshold)
8099 }8103 }
8100 8104
8101 func eventAffectsPresentation(ev core.AgentEvent) bool {8105 func eventAffectsPresentation(ev core.AgentEvent) bool {
M packages/agent/subagent_worker.go +87 -11
@@ -54,6 +54,10 @@ func runSubagentWorkerMode(ctx context.Context, args Args, version string) error
5454 args.inheritedAccountID = inherited.AccountID
5555 }
5656
57+ cfg, err := loadSubagentWorkerConfig()
58+ if err != nil {
59+ return err
60+ }
5761 r, err := Resolve(args, true)
5862 if err != nil {
5963 return err
@@ -86,6 +90,11 @@ func runSubagentWorkerMode(ctx context.Context, args Args, version string) error
8690 }
8791 announceSession(extMgr, sess)
8892
93+ // Resolve retains metadata for catalog models and for valid synthesized
94+ // local/routed/open-catalog models. The active agent also survives a session
95+ // resume that rebuilt it for a different provider/model pair.
96+ contextWindow := ag.ContextWindow
97+
8998 // Open the inbox listener BEFORE emitting agent_ready so the
9099 // supervisor can dial through on the very first send. The
91100 // subagent supervisor's Inbox retries dialing for a
@@ -204,14 +213,38 @@ func runSubagentWorkerMode(ctx context.Context, args Args, version string) error
204213 em.emit(ev.Type(), subagentEventData(ev))
205214 }
206215
216+ baseline := workerTranscriptState{
217+ messages: ag.Messages(),
218+ cost: ag.Cost(),
219+ lastTurn: ag.LastTurnUsage(),
220+ }
221+ compacted := false
207222 var persistCompaction func([]provider.Message) error
208223 if sess != nil {
209- persistCompaction = sess.AppendCompaction
224+ // Buffer the checkpoint until the continued turn completes. The final
225+ // transcript and cumulative usage are then persisted in one JSONL row.
226+ persistCompaction = func([]provider.Message) error {
227+ if err := c.Err(); err != nil {
228+ return err
229+ }
230+ compacted = true
231+ return nil
232+ }
210233 }
211- start, err := promptWithContextRecovery(c, ag, prompt, sink, persistCompaction)
212- WriteNewTranscript(ag, sess, start)
213-
234+ start, err := promptWithContextRecovery(c, ag, prompt, sink, modes.ContextRecoveryOptions{
235+ PersistCompaction: persistCompaction,
236+ ContextWindow: contextWindow,
237+ AutoCompactThreshold: cfg.AutoCompactThreshold,
238+ })
214239 output := finalAssistantText(ag.Messages()[start:])
240+ if persistErr := persistWorkerTranscript(c, ag, sess, start, compacted, baseline); persistErr != nil {
241+ if err != nil {
242+ err = errors.Join(err, persistErr)
243+ } else {
244+ err = persistErr
245+ }
246+ }
247+
215248 resultStatus := "succeeded"
216249 if err != nil {
217250 if errors.Is(err, context.Canceled) {
@@ -439,14 +472,57 @@ func canonicalWorkerEvent(typ string) string {
439472
440473 const subagentContextLimitMessage = "request exceeds the model context window; narrow the task or reduce gathered context"
441474
475+func loadSubagentWorkerConfig() (Config, error) {
476+ cfg, err := LoadConfig()
477+ if err != nil {
478+ return Config{}, fmt.Errorf("load subagent worker config: %w", err)
479+ }
480+ return cfg, nil
481+}
482+
483+// workerTranscriptState is restored when an atomic post-compaction checkpoint
484+// cannot be persisted.
485+type workerTranscriptState struct {
486+ messages []provider.Message
487+ cost provider.Usage
488+ lastTurn provider.Usage
489+}
490+
491+func persistWorkerTranscript(ctx context.Context, ag *core.Agent, sess *core.Session, from int, compacted bool, baseline workerTranscriptState) error {
492+ if sess == nil || ag == nil {
493+ return nil
494+ }
495+ if !compacted {
496+ if err := ctx.Err(); err != nil {
497+ return err
498+ }
499+ if err := WriteNewTranscript(ag, sess, from); err != nil {
500+ return fmt.Errorf("persist subagent transcript: %w", err)
501+ }
502+ return nil
503+ }
504+
505+ rollback := func() {
506+ ag.SetMessages(baseline.messages)
507+ ag.SeedCost(baseline.cost)
508+ ag.SeedLastTurnUsage(baseline.lastTurn)
509+ }
510+ if err := ctx.Err(); err != nil {
511+ rollback()
512+ return err
513+ }
514+ if err := sess.AppendCompactionWithUsage(ag.Messages(), ag.Cost()); err != nil {
515+ rollback()
516+ return fmt.Errorf("persist compacted subagent transcript: %w", err)
517+ }
518+ return nil
519+}
520+
442521 // promptWithContextRecovery preserves the worker's existing helper boundary
443-// while delegating its one-shot overflow recovery to modes. The returned index
444-// is the first message to include in the result or append after a persisted
445-// compaction checkpoint.
446-func promptWithContextRecovery(ctx context.Context, ag *core.Agent, prompt string, sink func(core.AgentEvent), persistCompaction func([]provider.Message) error) (outputStart int, err error) {
447- result, err := modes.PromptWithContextRecovery(ctx, ag, prompt, nil, sink, modes.ContextRecoveryOptions{
448- PersistCompaction: persistCompaction,
449- })
522+// while delegating proactive compaction and one-shot overflow recovery to
523+// modes. The returned index is the first message to include in the result.
524+func promptWithContextRecovery(ctx context.Context, ag *core.Agent, prompt string, sink func(core.AgentEvent), options modes.ContextRecoveryOptions) (outputStart int, err error) {
525+ result, err := modes.PromptWithContextRecovery(ctx, ag, prompt, nil, sink, options)
450526 return result.OutputStart, err
451527 }
452528
@@ -54,6 +54,10 @@ func runSubagentWorkerMode(ctx context.Context, args Args, version string) error
54 args.inheritedAccountID = inherited.AccountID54 args.inheritedAccountID = inherited.AccountID
55 }55 }
56 56
57+ cfg, err := loadSubagentWorkerConfig()
58+ if err != nil {
59+ return err
60+ }
57 r, err := Resolve(args, true)61 r, err := Resolve(args, true)
58 if err != nil {62 if err != nil {
59 return err63 return err
@@ -86,6 +90,11 @@ func runSubagentWorkerMode(ctx context.Context, args Args, version string) error
86 }90 }
87 announceSession(extMgr, sess)91 announceSession(extMgr, sess)
88 92
93+ // Resolve retains metadata for catalog models and for valid synthesized
94+ // local/routed/open-catalog models. The active agent also survives a session
95+ // resume that rebuilt it for a different provider/model pair.
96+ contextWindow := ag.ContextWindow
97+
89 // Open the inbox listener BEFORE emitting agent_ready so the98 // Open the inbox listener BEFORE emitting agent_ready so the
90 // supervisor can dial through on the very first send. The99 // supervisor can dial through on the very first send. The
91 // subagent supervisor's Inbox retries dialing for a100 // subagent supervisor's Inbox retries dialing for a
@@ -204,14 +213,38 @@ func runSubagentWorkerMode(ctx context.Context, args Args, version string) error
204 em.emit(ev.Type(), subagentEventData(ev))213 em.emit(ev.Type(), subagentEventData(ev))
205 }214 }
206 215
216+ baseline := workerTranscriptState{
217+ messages: ag.Messages(),
218+ cost: ag.Cost(),
219+ lastTurn: ag.LastTurnUsage(),
220+ }
221+ compacted := false
207 var persistCompaction func([]provider.Message) error222 var persistCompaction func([]provider.Message) error
208 if sess != nil {223 if sess != nil {
209- persistCompaction = sess.AppendCompaction224+ // Buffer the checkpoint until the continued turn completes. The final
225+ // transcript and cumulative usage are then persisted in one JSONL row.
226+ persistCompaction = func([]provider.Message) error {
227+ if err := c.Err(); err != nil {
228+ return err
229+ }
230+ compacted = true
231+ return nil
232+ }
210 }233 }
211- start, err := promptWithContextRecovery(c, ag, prompt, sink, persistCompaction)234+ start, err := promptWithContextRecovery(c, ag, prompt, sink, modes.ContextRecoveryOptions{
212- WriteNewTranscript(ag, sess, start)235+ PersistCompaction: persistCompaction,
213-236+ ContextWindow: contextWindow,
237+ AutoCompactThreshold: cfg.AutoCompactThreshold,
238+ })
214 output := finalAssistantText(ag.Messages()[start:])239 output := finalAssistantText(ag.Messages()[start:])
240+ if persistErr := persistWorkerTranscript(c, ag, sess, start, compacted, baseline); persistErr != nil {
241+ if err != nil {
242+ err = errors.Join(err, persistErr)
243+ } else {
244+ err = persistErr
245+ }
246+ }
247+
215 resultStatus := "succeeded"248 resultStatus := "succeeded"
216 if err != nil {249 if err != nil {
217 if errors.Is(err, context.Canceled) {250 if errors.Is(err, context.Canceled) {
@@ -439,14 +472,57 @@ func canonicalWorkerEvent(typ string) string {
439 472
440 const subagentContextLimitMessage = "request exceeds the model context window; narrow the task or reduce gathered context"473 const subagentContextLimitMessage = "request exceeds the model context window; narrow the task or reduce gathered context"
441 474
475+func loadSubagentWorkerConfig() (Config, error) {
476+ cfg, err := LoadConfig()
477+ if err != nil {
478+ return Config{}, fmt.Errorf("load subagent worker config: %w", err)
479+ }
480+ return cfg, nil
481+}
482+
483+// workerTranscriptState is restored when an atomic post-compaction checkpoint
484+// cannot be persisted.
485+type workerTranscriptState struct {
486+ messages []provider.Message
487+ cost provider.Usage
488+ lastTurn provider.Usage
489+}
490+
491+func persistWorkerTranscript(ctx context.Context, ag *core.Agent, sess *core.Session, from int, compacted bool, baseline workerTranscriptState) error {
492+ if sess == nil || ag == nil {
493+ return nil
494+ }
495+ if !compacted {
496+ if err := ctx.Err(); err != nil {
497+ return err
498+ }
499+ if err := WriteNewTranscript(ag, sess, from); err != nil {
500+ return fmt.Errorf("persist subagent transcript: %w", err)
501+ }
502+ return nil
503+ }
504+
505+ rollback := func() {
506+ ag.SetMessages(baseline.messages)
507+ ag.SeedCost(baseline.cost)
508+ ag.SeedLastTurnUsage(baseline.lastTurn)
509+ }
510+ if err := ctx.Err(); err != nil {
511+ rollback()
512+ return err
513+ }
514+ if err := sess.AppendCompactionWithUsage(ag.Messages(), ag.Cost()); err != nil {
515+ rollback()
516+ return fmt.Errorf("persist compacted subagent transcript: %w", err)
517+ }
518+ return nil
519+}
520+
442 // promptWithContextRecovery preserves the worker's existing helper boundary521 // promptWithContextRecovery preserves the worker's existing helper boundary
443-// while delegating its one-shot overflow recovery to modes. The returned index522+// while delegating proactive compaction and one-shot overflow recovery to
444-// is the first message to include in the result or append after a persisted523+// modes. The returned index is the first message to include in the result.
445-// compaction checkpoint.524+func promptWithContextRecovery(ctx context.Context, ag *core.Agent, prompt string, sink func(core.AgentEvent), options modes.ContextRecoveryOptions) (outputStart int, err error) {
446-func promptWithContextRecovery(ctx context.Context, ag *core.Agent, prompt string, sink func(core.AgentEvent), persistCompaction func([]provider.Message) error) (outputStart int, err error) {525+ result, err := modes.PromptWithContextRecovery(ctx, ag, prompt, nil, sink, options)
447- result, err := modes.PromptWithContextRecovery(ctx, ag, prompt, nil, sink, modes.ContextRecoveryOptions{
448- PersistCompaction: persistCompaction,
449- })
450 return result.OutputStart, err526 return result.OutputStart, err
451 }527 }
452 528
M packages/agent/subagent_worker_test.go +334 -10
@@ -8,10 +8,12 @@ import (
88 "io"
99 "os"
1010 "path/filepath"
11+ "reflect"
1112 "strings"
1213 "sync"
1314 "testing"
1415
16+ "github.com/bnema/zut/packages/agent/modes"
1517 "github.com/bnema/zut/packages/agent/subagents"
1618 "github.com/bnema/zut/packages/core"
1719 "github.com/bnema/zut/packages/provider"
@@ -81,6 +83,33 @@ func (c *subagentContextRecoveryClient) Stream(_ context.Context, req provider.R
8183 return events, nil
8284 }
8385
86+type subagentProactiveClient struct {
87+ mu sync.Mutex
88+ requests []provider.Request
89+}
90+
91+func (c *subagentProactiveClient) Name() string { return "subagent-proactive-test" }
92+
93+func (c *subagentProactiveClient) Stream(_ context.Context, req provider.Request) (<-chan provider.Event, error) {
94+ c.mu.Lock()
95+ c.requests = append(c.requests, req)
96+ c.mu.Unlock()
97+ events := make(chan provider.Event, 2)
98+ go func() {
99+ defer close(events)
100+ if strings.Contains(req.System, "context summarization assistant") {
101+ events <- provider.EventTextDelta{Delta: "worker summary"}
102+ events <- provider.EventDone{Stop: provider.StopEnd}
103+ return
104+ }
105+ events <- provider.EventDone{Stop: provider.StopEnd, Message: provider.Message{
106+ Role: provider.RoleAssistant,
107+ Content: []provider.Content{provider.TextBlock{Text: "continued same task"}},
108+ }}
109+ }()
110+ return events, nil
111+}
112+
84113 func TestPromptWithContextRecoveryCompactsAndContinues(t *testing.T) {
85114 client := &subagentContextRecoveryClient{retried: make(chan provider.Request, 1)}
86115 agent := core.NewAgent(client, "test-model", "", nil)
@@ -91,7 +120,7 @@ func TestPromptWithContextRecoveryCompactsAndContinues(t *testing.T) {
91120 {Role: provider.RoleAssistant, Content: []provider.Content{provider.TextBlock{Text: "four"}}},
92121 })
93122
94- outputStart, err := promptWithContextRecovery(context.Background(), agent, "finish the task", nil, nil)
123+ outputStart, err := promptWithContextRecovery(context.Background(), agent, "finish the task", nil, modes.ContextRecoveryOptions{})
95124 if err != nil {
96125 t.Fatalf("promptWithContextRecovery returned %v", err)
97126 }
@@ -123,7 +152,7 @@ func TestPromptWithContextRecoveryStopsAfterOneRetry(t *testing.T) {
123152 agent := core.NewAgent(client, "test-model", "", nil)
124153 agent.SetMessages([]provider.Message{{Role: provider.RoleUser, Content: []provider.Content{provider.TextBlock{Text: "prior context"}}}})
125154
126- _, err := promptWithContextRecovery(context.Background(), agent, "finish the task", nil, nil)
155+ _, err := promptWithContextRecovery(context.Background(), agent, "finish the task", nil, modes.ContextRecoveryOptions{})
127156 if !provider.IsContextOverflowError(err) {
128157 t.Fatalf("promptWithContextRecovery error = %v, want context overflow", err)
129158 }
@@ -147,7 +176,7 @@ func TestPromptWithContextRecoveryDoesNotCompactAfterCancellation(t *testing.T)
147176 if turnEnd, ok := event.(core.EvTurnEnd); ok && provider.IsContextOverflowError(turnEnd.Err) {
148177 cancel()
149178 }
150- }, nil)
179+ }, modes.ContextRecoveryOptions{})
151180 if !provider.IsContextOverflowError(err) {
152181 t.Fatalf("promptWithContextRecovery error = %v, want context overflow", err)
153182 }
@@ -170,7 +199,7 @@ func TestPromptWithContextRecoveryPropagatesCompactionFailure(t *testing.T) {
170199 agent := core.NewAgent(client, "test-model", "", nil)
171200 agent.SetMessages([]provider.Message{{Role: provider.RoleUser, Content: []provider.Content{provider.TextBlock{Text: "prior context"}}}})
172201
173- _, err := promptWithContextRecovery(context.Background(), agent, "finish the task", nil, nil)
202+ _, err := promptWithContextRecovery(context.Background(), agent, "finish the task", nil, modes.ContextRecoveryOptions{})
174203 if err == nil || !strings.Contains(err.Error(), "compact transcript after initial overflow") {
175204 t.Fatalf("promptWithContextRecovery error = %v, want compaction failure", err)
176205 }
@@ -184,7 +213,7 @@ func TestPromptWithContextRecoveryKeepsSinglePromptIntact(t *testing.T) {
184213 client := &subagentContextRecoveryClient{retried: make(chan provider.Request, 1)}
185214 agent := core.NewAgent(client, "test-model", "", nil)
186215
187- _, err := promptWithContextRecovery(context.Background(), agent, "finish the task", nil, nil)
216+ _, err := promptWithContextRecovery(context.Background(), agent, "finish the task", nil, modes.ContextRecoveryOptions{})
188217 if !provider.IsContextOverflowError(err) {
189218 t.Fatalf("promptWithContextRecovery error = %v, want context overflow", err)
190219 }
@@ -216,8 +245,8 @@ func TestPromptWithContextRecoveryRetainsPromptWhenPersistenceFails(t *testing.T
216245 agent := core.NewAgent(client, "test-model", "", nil)
217246 agent.SetMessages(history)
218247
219- outputStart, err := promptWithContextRecovery(context.Background(), agent, "finish the task", nil, func([]provider.Message) error {
220- return errors.New("persist failed")
248+ outputStart, err := promptWithContextRecovery(context.Background(), agent, "finish the task", nil, modes.ContextRecoveryOptions{
249+ PersistCompaction: func([]provider.Message) error { return errors.New("persist failed") },
221250 })
222251 if err == nil || !strings.Contains(err.Error(), "persist compacted transcript") {
223252 t.Fatalf("promptWithContextRecovery error = %v, want persistence failure", err)
@@ -225,7 +254,9 @@ func TestPromptWithContextRecoveryRetainsPromptWhenPersistenceFails(t *testing.T
225254 if got := workerRequestMessageTextCount(agent.Messages(), "prior context"); got != 1 || workerRequestMessageTextCount(agent.Messages(), "finish the task") != 1 {
226255 t.Fatalf("agent transcript lost the pending task after persistence failure: %#v", agent.Messages())
227256 }
228- WriteNewTranscript(agent, session, outputStart)
257+ if err := WriteNewTranscript(agent, session, outputStart); err != nil {
258+ t.Fatalf("persist transcript: %v", err)
259+ }
229260 if err := session.Close(); err != nil {
230261 t.Fatalf("close session: %v", err)
231262 }
@@ -269,11 +300,15 @@ func TestPromptWithContextRecoveryPersistsCompaction(t *testing.T) {
269300
270301 agent := core.NewAgent(client, "test-model", "", nil)
271302 agent.SetMessages(history)
272- outputStart, err := promptWithContextRecovery(context.Background(), agent, "finish the task", nil, session.AppendCompaction)
303+ outputStart, err := promptWithContextRecovery(context.Background(), agent, "finish the task", nil, modes.ContextRecoveryOptions{
304+ PersistCompaction: session.AppendCompaction,
305+ })
273306 if err != nil {
274307 t.Fatalf("promptWithContextRecovery returned %v", err)
275308 }
276- WriteNewTranscript(agent, session, outputStart)
309+ if err := WriteNewTranscript(agent, session, outputStart); err != nil {
310+ t.Fatalf("persist transcript: %v", err)
311+ }
277312 if err := session.Close(); err != nil {
278313 t.Fatalf("close session: %v", err)
279314 }
@@ -315,6 +350,295 @@ func TestPromptWithContextRecoveryPersistsCompaction(t *testing.T) {
315350 }
316351 }
317352
353+func TestWorkerUsesConfiguredAutoCompactionThreshold(t *testing.T) {
354+ t.Setenv("ZUT_HOME", t.TempDir())
355+ threshold := 70
356+ if err := SaveConfig(Config{AutoCompactThreshold: &threshold}); err != nil {
357+ t.Fatalf("save config: %v", err)
358+ }
359+ cfg, err := loadSubagentWorkerConfig()
360+ if err != nil {
361+ t.Fatalf("load worker config: %v", err)
362+ }
363+
364+ client := &subagentProactiveClient{}
365+ agent := core.NewAgent(client, "model", "", nil)
366+ agent.SetMessages([]provider.Message{
367+ {Role: provider.RoleUser, Content: []provider.Content{provider.TextBlock{Text: "history"}}},
368+ {Role: provider.RoleAssistant, Content: []provider.Content{provider.TextBlock{Text: "answer"}}},
369+ })
370+ agent.SeedLastTurnUsage(provider.Usage{CacheReadTokens: 75})
371+ start, err := promptWithContextRecovery(context.Background(), agent, "finish exact task", nil, modes.ContextRecoveryOptions{
372+ ContextWindow: 100,
373+ AutoCompactThreshold: cfg.AutoCompactThreshold,
374+ })
375+ if err != nil {
376+ t.Fatalf("promptWithContextRecovery: %v", err)
377+ }
378+ if got := finalAssistantText(agent.Messages()[start:]); got != "continued same task" {
379+ t.Fatalf("continued output = %q", got)
380+ }
381+ client.mu.Lock()
382+ requests := append([]provider.Request(nil), client.requests...)
383+ client.mu.Unlock()
384+ if len(requests) != 2 || workerRequestUserTextCount(requests[1], "finish exact task") != 1 {
385+ t.Fatalf("requests = %#v, want compaction then the same task once", requests)
386+ }
387+}
388+
389+func TestWorkerConfigLoadRejectsMalformedConfig(t *testing.T) {
390+ t.Setenv("ZUT_HOME", t.TempDir())
391+ if err := os.MkdirAll(filepath.Dir(ConfigPath()), 0o755); err != nil {
392+ t.Fatal(err)
393+ }
394+ if err := os.WriteFile(ConfigPath(), []byte("{malformed"), 0o600); err != nil {
395+ t.Fatal(err)
396+ }
397+ if _, err := loadSubagentWorkerConfig(); err == nil || !strings.Contains(err.Error(), "load subagent worker config") {
398+ t.Fatalf("loadSubagentWorkerConfig error = %v, want parse failure", err)
399+ }
400+}
401+
402+type subagentSummaryUsageClient struct{}
403+
404+func (c *subagentSummaryUsageClient) Name() string { return "subagent-summary-usage-test" }
405+
406+func (c *subagentSummaryUsageClient) Stream(_ context.Context, req provider.Request) (<-chan provider.Event, error) {
407+ events := make(chan provider.Event, 3)
408+ go func() {
409+ defer close(events)
410+ if strings.Contains(req.System, "context summarization assistant") {
411+ events <- provider.EventUsage{Usage: provider.Usage{InputTokens: 3, OutputTokens: 1, CostUSD: 0.03}}
412+ events <- provider.EventTextDelta{Delta: "worker summary"}
413+ return
414+ }
415+ events <- provider.EventUsage{Usage: provider.Usage{InputTokens: 5, OutputTokens: 2, CostUSD: 0.05}}
416+ events <- provider.EventDone{Stop: provider.StopEnd, Message: provider.Message{
417+ Role: provider.RoleAssistant,
418+ Content: []provider.Content{provider.TextBlock{Text: "continued"}},
419+ }}
420+ }()
421+ return events, nil
422+}
423+
424+func TestPersistWorkerSummaryUsageReopensWithCumulativeUsage(t *testing.T) {
425+ session, err := core.NewSession(t.TempDir(), "cwd", "provider", "model", "test")
426+ if err != nil {
427+ t.Fatal(err)
428+ }
429+ history := []provider.Message{
430+ {Role: provider.RoleUser, Content: []provider.Content{provider.TextBlock{Text: "old task"}}},
431+ {Role: provider.RoleAssistant, Content: []provider.Content{provider.TextBlock{Text: "old answer"}}},
432+ }
433+ for _, message := range history {
434+ if err := session.AppendMessage(message); err != nil {
435+ t.Fatal(err)
436+ }
437+ }
438+ baseline := provider.Usage{InputTokens: 10, OutputTokens: 2, CostUSD: 0.1}
439+ if err := session.AppendUsage(baseline, baseline); err != nil {
440+ t.Fatal(err)
441+ }
442+ agent := core.NewAgent(&subagentSummaryUsageClient{}, "model", "", nil)
443+ agent.SetMessages(history)
444+ agent.SeedCost(baseline)
445+ agent.SeedLastTurnUsage(provider.Usage{CacheReadTokens: 90})
446+ outputStart, err := promptWithContextRecovery(context.Background(), agent, "continue task", nil, modes.ContextRecoveryOptions{
447+ ContextWindow: 100,
448+ })
449+ if err != nil {
450+ t.Fatalf("promptWithContextRecovery: %v", err)
451+ }
452+ if outputStart != len(history) {
453+ t.Fatalf("compaction output start = %d, want %d", outputStart, len(history))
454+ }
455+ expected := provider.Usage{InputTokens: 18, OutputTokens: 5, CostUSD: 0.18}
456+ if !reflect.DeepEqual(agent.Cost(), expected) {
457+ t.Fatalf("worker cumulative usage = %#v, want %#v", agent.Cost(), expected)
458+ }
459+ if err := persistWorkerTranscript(context.Background(), agent, session, outputStart, true, workerTranscriptState{messages: history, cost: baseline}); err != nil {
460+ t.Fatalf("persistWorkerTranscript: %v", err)
461+ }
462+ if err := session.Close(); err != nil {
463+ t.Fatal(err)
464+ }
465+ cumulative, lastTurn, err := core.SessionUsageDetail(session.Path)
466+ if err != nil {
467+ t.Fatal(err)
468+ }
469+ if !reflect.DeepEqual(cumulative, expected) {
470+ t.Fatalf("reopened cumulative usage = %#v, want %#v", cumulative, expected)
471+ }
472+ wantLast := provider.Usage{InputTokens: 8, OutputTokens: 3, CostUSD: expected.CostUSD - baseline.CostUSD}
473+ if !reflect.DeepEqual(lastTurn, wantLast) {
474+ t.Fatalf("reopened last-turn usage = %#v, want %#v", lastTurn, wantLast)
475+ }
476+}
477+
478+func TestPersistWorkerCompactedTranscriptReopensWithTurnAndUsage(t *testing.T) {
479+ session, err := core.NewSession(t.TempDir(), "cwd", "provider", "model", "test")
480+ if err != nil {
481+ t.Fatal(err)
482+ }
483+ baselineMessage := provider.Message{Role: provider.RoleUser, Content: []provider.Content{provider.TextBlock{Text: "baseline"}}}
484+ if err := session.AppendMessage(baselineMessage); err != nil {
485+ t.Fatal(err)
486+ }
487+ baselineUsage := provider.Usage{InputTokens: 10, OutputTokens: 2}
488+ if err := session.AppendUsage(baselineUsage, baselineUsage); err != nil {
489+ t.Fatal(err)
490+ }
491+
492+ agent := core.NewAgent(&subagentProactiveClient{}, "model", "", nil)
493+ agent.SetMessages([]provider.Message{
494+ {Role: provider.RoleUser, Content: []provider.Content{provider.TextBlock{Text: "summary"}}},
495+ {Role: provider.RoleUser, Content: []provider.Content{provider.TextBlock{Text: "current task"}}},
496+ {Role: provider.RoleAssistant, Content: []provider.Content{provider.TextBlock{Text: "current answer"}}},
497+ })
498+ cumulative := provider.Usage{InputTokens: 30, OutputTokens: 8, CacheReadTokens: 4}
499+ agent.SeedCost(cumulative)
500+ if err := persistWorkerTranscript(context.Background(), agent, session, 1, true, workerTranscriptState{
501+ messages: []provider.Message{baselineMessage},
502+ cost: baselineUsage,
503+ }); err != nil {
504+ t.Fatalf("persistWorkerTranscript: %v", err)
505+ }
506+ if err := session.Close(); err != nil {
507+ t.Fatal(err)
508+ }
509+
510+ snapshot, err := core.ReadSessionSnapshot(session.Path)
511+ if err != nil {
512+ t.Fatal(err)
513+ }
514+ if workerRequestMessageTextCount(snapshot.Messages, "current task") != 1 || finalAssistantText(snapshot.Messages) != "current answer" {
515+ t.Fatalf("reopened messages = %#v, want full continued turn", snapshot.Messages)
516+ }
517+ if len(snapshot.UsageCheckpoints) == 0 || !reflect.DeepEqual(snapshot.UsageCheckpoints[len(snapshot.UsageCheckpoints)-1].Cumulative, cumulative) {
518+ t.Fatalf("usage checkpoints = %#v, want cumulative %#v", snapshot.UsageCheckpoints, cumulative)
519+ }
520+}
521+
522+func TestPersistWorkerCompactionFailureRollsBackAndKeepsSessionReopenable(t *testing.T) {
523+ session, err := core.NewSession(t.TempDir(), "cwd", "provider", "model", "test")
524+ if err != nil {
525+ t.Fatal(err)
526+ }
527+ baselineMessage := provider.Message{Role: provider.RoleUser, Content: []provider.Content{provider.TextBlock{Text: "baseline"}}}
528+ if err := session.AppendMessage(baselineMessage); err != nil {
529+ t.Fatal(err)
530+ }
531+ baselineUsage := provider.Usage{InputTokens: 10, OutputTokens: 2}
532+ if err := session.AppendUsage(baselineUsage, baselineUsage); err != nil {
533+ t.Fatal(err)
534+ }
535+ if err := session.Close(); err != nil {
536+ t.Fatal(err)
537+ }
538+
539+ agent := core.NewAgent(&subagentProactiveClient{}, "model", "", nil)
540+ agent.SetMessages([]provider.Message{{Role: provider.RoleUser, Content: []provider.Content{provider.TextBlock{Text: "unsaved compacted turn"}}}})
541+ agent.SeedCost(provider.Usage{InputTokens: 99})
542+ baseline := workerTranscriptState{messages: []provider.Message{baselineMessage}, cost: baselineUsage, lastTurn: baselineUsage}
543+ if err := persistWorkerTranscript(context.Background(), agent, session, 0, true, baseline); err == nil {
544+ t.Fatal("persistWorkerTranscript succeeded on closed session")
545+ }
546+ if !reflect.DeepEqual(agent.Messages(), baseline.messages) || !reflect.DeepEqual(agent.Cost(), baselineUsage) {
547+ t.Fatalf("agent was not rolled back: messages=%#v cost=%#v", agent.Messages(), agent.Cost())
548+ }
549+ snapshot, err := core.ReadSessionSnapshot(session.Path)
550+ if err != nil {
551+ t.Fatalf("reopen snapshot: %v", err)
552+ }
553+ if !reflect.DeepEqual(snapshot.Messages, baseline.messages) {
554+ t.Fatalf("reopened transcript = %#v, want baseline %#v", snapshot.Messages, baseline.messages)
555+ }
556+ if len(snapshot.UsageCheckpoints) != 1 || !reflect.DeepEqual(snapshot.UsageCheckpoints[0].Cumulative, baselineUsage) {
557+ t.Fatalf("reopened usage = %#v, want baseline", snapshot.UsageCheckpoints)
558+ }
559+}
560+
561+func TestPersistWorkerTranscriptSkipsCanceledNonCompactedTurn(t *testing.T) {
562+ session, err := core.NewSession(t.TempDir(), "cwd", "provider", "model", "test")
563+ if err != nil {
564+ t.Fatal(err)
565+ }
566+ baselineMessage := provider.Message{Role: provider.RoleUser, Content: []provider.Content{provider.TextBlock{Text: "baseline"}}}
567+ if err := session.AppendMessage(baselineMessage); err != nil {
568+ t.Fatal(err)
569+ }
570+ agent := core.NewAgent(&subagentProactiveClient{}, "model", "", nil)
571+ agent.SetMessages([]provider.Message{
572+ baselineMessage,
573+ {Role: provider.RoleAssistant, Content: []provider.Content{provider.TextBlock{Text: "canceled answer"}}},
574+ })
575+ ctx, cancel := context.WithCancel(context.Background())
576+ cancel()
577+ if err := persistWorkerTranscript(ctx, agent, session, 1, false, workerTranscriptState{messages: []provider.Message{baselineMessage}}); !errors.Is(err, context.Canceled) {
578+ t.Fatalf("persistWorkerTranscript error = %v, want cancellation", err)
579+ }
580+ if err := session.Close(); err != nil {
581+ t.Fatal(err)
582+ }
583+ snapshot, err := core.ReadSessionSnapshot(session.Path)
584+ if err != nil {
585+ t.Fatal(err)
586+ }
587+ if !reflect.DeepEqual(snapshot.Messages, []provider.Message{baselineMessage}) {
588+ t.Fatalf("reopened messages = %#v, want baseline only", snapshot.Messages)
589+ }
590+}
591+
592+func TestPersistWorkerCompactionHonorsCancellation(t *testing.T) {
593+ session, err := core.NewSession(t.TempDir(), "cwd", "provider", "model", "test")
594+ if err != nil {
595+ t.Fatal(err)
596+ }
597+ baselineMessage := provider.Message{Role: provider.RoleUser, Content: []provider.Content{provider.TextBlock{Text: "baseline"}}}
598+ if err := session.AppendMessage(baselineMessage); err != nil {
599+ t.Fatal(err)
600+ }
601+ baseline := workerTranscriptState{messages: []provider.Message{baselineMessage}}
602+ agent := core.NewAgent(&subagentProactiveClient{}, "model", "", nil)
603+ agent.SetMessages([]provider.Message{{Role: provider.RoleUser, Content: []provider.Content{provider.TextBlock{Text: "must not persist"}}}})
604+ ctx, cancel := context.WithCancel(context.Background())
605+ cancel()
606+ if err := persistWorkerTranscript(ctx, agent, session, 0, true, baseline); !errors.Is(err, context.Canceled) {
607+ t.Fatalf("persistWorkerTranscript error = %v, want cancellation", err)
608+ }
609+ if !reflect.DeepEqual(agent.Messages(), baseline.messages) {
610+ t.Fatalf("agent messages = %#v, want rollback %#v", agent.Messages(), baseline.messages)
611+ }
612+ if err := session.Close(); err != nil {
613+ t.Fatal(err)
614+ }
615+ snapshot, err := core.ReadSessionSnapshot(session.Path)
616+ if err != nil {
617+ t.Fatal(err)
618+ }
619+ if !reflect.DeepEqual(snapshot.Messages, baseline.messages) {
620+ t.Fatalf("reopened messages = %#v, want baseline", snapshot.Messages)
621+ }
622+}
623+
624+func TestWriteNewTranscriptReturnsPersistenceErrors(t *testing.T) {
625+ session, err := core.NewSession(t.TempDir(), "cwd", "provider", "model", "test")
626+ if err != nil {
627+ t.Fatal(err)
628+ }
629+ if err := session.AppendMessage(provider.Message{Role: provider.RoleUser, Content: []provider.Content{provider.TextBlock{Text: "baseline"}}}); err != nil {
630+ t.Fatal(err)
631+ }
632+ if err := session.Close(); err != nil {
633+ t.Fatal(err)
634+ }
635+ agent := core.NewAgent(&subagentProactiveClient{}, "model", "", nil)
636+ agent.SetMessages([]provider.Message{{Role: provider.RoleUser, Content: []provider.Content{provider.TextBlock{Text: "new"}}}})
637+ if err := WriteNewTranscript(agent, session, 0); err == nil || !strings.Contains(err.Error(), "append transcript message") {
638+ t.Fatalf("WriteNewTranscript error = %v, want append failure", err)
639+ }
640+}
641+
318642 func workerRequestMessageTextCount(messages []provider.Message, want string) int {
319643 count := 0
320644 for _, message := range messages {
@@ -8,10 +8,12 @@ import (
8 "io"8 "io"
9 "os"9 "os"
10 "path/filepath"10 "path/filepath"
11+ "reflect"
11 "strings"12 "strings"
12 "sync"13 "sync"
13 "testing"14 "testing"
14 15
16+ "github.com/bnema/zut/packages/agent/modes"
15 "github.com/bnema/zut/packages/agent/subagents"17 "github.com/bnema/zut/packages/agent/subagents"
16 "github.com/bnema/zut/packages/core"18 "github.com/bnema/zut/packages/core"
17 "github.com/bnema/zut/packages/provider"19 "github.com/bnema/zut/packages/provider"
@@ -81,6 +83,33 @@ func (c *subagentContextRecoveryClient) Stream(_ context.Context, req provider.R
81 return events, nil83 return events, nil
82 }84 }
83 85
86+type subagentProactiveClient struct {
87+ mu sync.Mutex
88+ requests []provider.Request
89+}
90+
91+func (c *subagentProactiveClient) Name() string { return "subagent-proactive-test" }
92+
93+func (c *subagentProactiveClient) Stream(_ context.Context, req provider.Request) (<-chan provider.Event, error) {
94+ c.mu.Lock()
95+ c.requests = append(c.requests, req)
96+ c.mu.Unlock()
97+ events := make(chan provider.Event, 2)
98+ go func() {
99+ defer close(events)
100+ if strings.Contains(req.System, "context summarization assistant") {
101+ events <- provider.EventTextDelta{Delta: "worker summary"}
102+ events <- provider.EventDone{Stop: provider.StopEnd}
103+ return
104+ }
105+ events <- provider.EventDone{Stop: provider.StopEnd, Message: provider.Message{
106+ Role: provider.RoleAssistant,
107+ Content: []provider.Content{provider.TextBlock{Text: "continued same task"}},
108+ }}
109+ }()
110+ return events, nil
111+}
112+
84 func TestPromptWithContextRecoveryCompactsAndContinues(t *testing.T) {113 func TestPromptWithContextRecoveryCompactsAndContinues(t *testing.T) {
85 client := &subagentContextRecoveryClient{retried: make(chan provider.Request, 1)}114 client := &subagentContextRecoveryClient{retried: make(chan provider.Request, 1)}
86 agent := core.NewAgent(client, "test-model", "", nil)115 agent := core.NewAgent(client, "test-model", "", nil)
@@ -91,7 +120,7 @@ func TestPromptWithContextRecoveryCompactsAndContinues(t *testing.T) {
91 {Role: provider.RoleAssistant, Content: []provider.Content{provider.TextBlock{Text: "four"}}},120 {Role: provider.RoleAssistant, Content: []provider.Content{provider.TextBlock{Text: "four"}}},
92 })121 })
93 122
94- outputStart, err := promptWithContextRecovery(context.Background(), agent, "finish the task", nil, nil)123+ outputStart, err := promptWithContextRecovery(context.Background(), agent, "finish the task", nil, modes.ContextRecoveryOptions{})
95 if err != nil {124 if err != nil {
96 t.Fatalf("promptWithContextRecovery returned %v", err)125 t.Fatalf("promptWithContextRecovery returned %v", err)
97 }126 }
@@ -123,7 +152,7 @@ func TestPromptWithContextRecoveryStopsAfterOneRetry(t *testing.T) {
123 agent := core.NewAgent(client, "test-model", "", nil)152 agent := core.NewAgent(client, "test-model", "", nil)
124 agent.SetMessages([]provider.Message{{Role: provider.RoleUser, Content: []provider.Content{provider.TextBlock{Text: "prior context"}}}})153 agent.SetMessages([]provider.Message{{Role: provider.RoleUser, Content: []provider.Content{provider.TextBlock{Text: "prior context"}}}})
125 154
126- _, err := promptWithContextRecovery(context.Background(), agent, "finish the task", nil, nil)155+ _, err := promptWithContextRecovery(context.Background(), agent, "finish the task", nil, modes.ContextRecoveryOptions{})
127 if !provider.IsContextOverflowError(err) {156 if !provider.IsContextOverflowError(err) {
128 t.Fatalf("promptWithContextRecovery error = %v, want context overflow", err)157 t.Fatalf("promptWithContextRecovery error = %v, want context overflow", err)
129 }158 }
@@ -147,7 +176,7 @@ func TestPromptWithContextRecoveryDoesNotCompactAfterCancellation(t *testing.T)
147 if turnEnd, ok := event.(core.EvTurnEnd); ok && provider.IsContextOverflowError(turnEnd.Err) {176 if turnEnd, ok := event.(core.EvTurnEnd); ok && provider.IsContextOverflowError(turnEnd.Err) {
148 cancel()177 cancel()
149 }178 }
150- }, nil)179+ }, modes.ContextRecoveryOptions{})
151 if !provider.IsContextOverflowError(err) {180 if !provider.IsContextOverflowError(err) {
152 t.Fatalf("promptWithContextRecovery error = %v, want context overflow", err)181 t.Fatalf("promptWithContextRecovery error = %v, want context overflow", err)
153 }182 }
@@ -170,7 +199,7 @@ func TestPromptWithContextRecoveryPropagatesCompactionFailure(t *testing.T) {
170 agent := core.NewAgent(client, "test-model", "", nil)199 agent := core.NewAgent(client, "test-model", "", nil)
171 agent.SetMessages([]provider.Message{{Role: provider.RoleUser, Content: []provider.Content{provider.TextBlock{Text: "prior context"}}}})200 agent.SetMessages([]provider.Message{{Role: provider.RoleUser, Content: []provider.Content{provider.TextBlock{Text: "prior context"}}}})
172 201
173- _, err := promptWithContextRecovery(context.Background(), agent, "finish the task", nil, nil)202+ _, err := promptWithContextRecovery(context.Background(), agent, "finish the task", nil, modes.ContextRecoveryOptions{})
174 if err == nil || !strings.Contains(err.Error(), "compact transcript after initial overflow") {203 if err == nil || !strings.Contains(err.Error(), "compact transcript after initial overflow") {
175 t.Fatalf("promptWithContextRecovery error = %v, want compaction failure", err)204 t.Fatalf("promptWithContextRecovery error = %v, want compaction failure", err)
176 }205 }
@@ -184,7 +213,7 @@ func TestPromptWithContextRecoveryKeepsSinglePromptIntact(t *testing.T) {
184 client := &subagentContextRecoveryClient{retried: make(chan provider.Request, 1)}213 client := &subagentContextRecoveryClient{retried: make(chan provider.Request, 1)}
185 agent := core.NewAgent(client, "test-model", "", nil)214 agent := core.NewAgent(client, "test-model", "", nil)
186 215
187- _, err := promptWithContextRecovery(context.Background(), agent, "finish the task", nil, nil)216+ _, err := promptWithContextRecovery(context.Background(), agent, "finish the task", nil, modes.ContextRecoveryOptions{})
188 if !provider.IsContextOverflowError(err) {217 if !provider.IsContextOverflowError(err) {
189 t.Fatalf("promptWithContextRecovery error = %v, want context overflow", err)218 t.Fatalf("promptWithContextRecovery error = %v, want context overflow", err)
190 }219 }
@@ -216,8 +245,8 @@ func TestPromptWithContextRecoveryRetainsPromptWhenPersistenceFails(t *testing.T
216 agent := core.NewAgent(client, "test-model", "", nil)245 agent := core.NewAgent(client, "test-model", "", nil)
217 agent.SetMessages(history)246 agent.SetMessages(history)
218 247
219- outputStart, err := promptWithContextRecovery(context.Background(), agent, "finish the task", nil, func([]provider.Message) error {248+ outputStart, err := promptWithContextRecovery(context.Background(), agent, "finish the task", nil, modes.ContextRecoveryOptions{
220- return errors.New("persist failed")249+ PersistCompaction: func([]provider.Message) error { return errors.New("persist failed") },
221 })250 })
222 if err == nil || !strings.Contains(err.Error(), "persist compacted transcript") {251 if err == nil || !strings.Contains(err.Error(), "persist compacted transcript") {
223 t.Fatalf("promptWithContextRecovery error = %v, want persistence failure", err)252 t.Fatalf("promptWithContextRecovery error = %v, want persistence failure", err)
@@ -225,7 +254,9 @@ func TestPromptWithContextRecoveryRetainsPromptWhenPersistenceFails(t *testing.T
225 if got := workerRequestMessageTextCount(agent.Messages(), "prior context"); got != 1 || workerRequestMessageTextCount(agent.Messages(), "finish the task") != 1 {254 if got := workerRequestMessageTextCount(agent.Messages(), "prior context"); got != 1 || workerRequestMessageTextCount(agent.Messages(), "finish the task") != 1 {
226 t.Fatalf("agent transcript lost the pending task after persistence failure: %#v", agent.Messages())255 t.Fatalf("agent transcript lost the pending task after persistence failure: %#v", agent.Messages())
227 }256 }
228- WriteNewTranscript(agent, session, outputStart)257+ if err := WriteNewTranscript(agent, session, outputStart); err != nil {
258+ t.Fatalf("persist transcript: %v", err)
259+ }
229 if err := session.Close(); err != nil {260 if err := session.Close(); err != nil {
230 t.Fatalf("close session: %v", err)261 t.Fatalf("close session: %v", err)
231 }262 }
@@ -269,11 +300,15 @@ func TestPromptWithContextRecoveryPersistsCompaction(t *testing.T) {
269 300
270 agent := core.NewAgent(client, "test-model", "", nil)301 agent := core.NewAgent(client, "test-model", "", nil)
271 agent.SetMessages(history)302 agent.SetMessages(history)
272- outputStart, err := promptWithContextRecovery(context.Background(), agent, "finish the task", nil, session.AppendCompaction)303+ outputStart, err := promptWithContextRecovery(context.Background(), agent, "finish the task", nil, modes.ContextRecoveryOptions{
304+ PersistCompaction: session.AppendCompaction,
305+ })
273 if err != nil {306 if err != nil {
274 t.Fatalf("promptWithContextRecovery returned %v", err)307 t.Fatalf("promptWithContextRecovery returned %v", err)
275 }308 }
276- WriteNewTranscript(agent, session, outputStart)309+ if err := WriteNewTranscript(agent, session, outputStart); err != nil {
310+ t.Fatalf("persist transcript: %v", err)
311+ }
277 if err := session.Close(); err != nil {312 if err := session.Close(); err != nil {
278 t.Fatalf("close session: %v", err)313 t.Fatalf("close session: %v", err)
279 }314 }
@@ -315,6 +350,295 @@ func TestPromptWithContextRecoveryPersistsCompaction(t *testing.T) {
315 }350 }
316 }351 }
317 352
353+func TestWorkerUsesConfiguredAutoCompactionThreshold(t *testing.T) {
354+ t.Setenv("ZUT_HOME", t.TempDir())
355+ threshold := 70
356+ if err := SaveConfig(Config{AutoCompactThreshold: &threshold}); err != nil {
357+ t.Fatalf("save config: %v", err)
358+ }
359+ cfg, err := loadSubagentWorkerConfig()
360+ if err != nil {
361+ t.Fatalf("load worker config: %v", err)
362+ }
363+
364+ client := &subagentProactiveClient{}
365+ agent := core.NewAgent(client, "model", "", nil)
366+ agent.SetMessages([]provider.Message{
367+ {Role: provider.RoleUser, Content: []provider.Content{provider.TextBlock{Text: "history"}}},
368+ {Role: provider.RoleAssistant, Content: []provider.Content{provider.TextBlock{Text: "answer"}}},
369+ })
370+ agent.SeedLastTurnUsage(provider.Usage{CacheReadTokens: 75})
371+ start, err := promptWithContextRecovery(context.Background(), agent, "finish exact task", nil, modes.ContextRecoveryOptions{
372+ ContextWindow: 100,
373+ AutoCompactThreshold: cfg.AutoCompactThreshold,
374+ })
375+ if err != nil {
376+ t.Fatalf("promptWithContextRecovery: %v", err)
377+ }
378+ if got := finalAssistantText(agent.Messages()[start:]); got != "continued same task" {
379+ t.Fatalf("continued output = %q", got)
380+ }
381+ client.mu.Lock()
382+ requests := append([]provider.Request(nil), client.requests...)
383+ client.mu.Unlock()
384+ if len(requests) != 2 || workerRequestUserTextCount(requests[1], "finish exact task") != 1 {
385+ t.Fatalf("requests = %#v, want compaction then the same task once", requests)
386+ }
387+}
388+
389+func TestWorkerConfigLoadRejectsMalformedConfig(t *testing.T) {
390+ t.Setenv("ZUT_HOME", t.TempDir())
391+ if err := os.MkdirAll(filepath.Dir(ConfigPath()), 0o755); err != nil {
392+ t.Fatal(err)
393+ }
394+ if err := os.WriteFile(ConfigPath(), []byte("{malformed"), 0o600); err != nil {
395+ t.Fatal(err)
396+ }
397+ if _, err := loadSubagentWorkerConfig(); err == nil || !strings.Contains(err.Error(), "load subagent worker config") {
398+ t.Fatalf("loadSubagentWorkerConfig error = %v, want parse failure", err)
399+ }
400+}
401+
402+type subagentSummaryUsageClient struct{}
403+
404+func (c *subagentSummaryUsageClient) Name() string { return "subagent-summary-usage-test" }
405+
406+func (c *subagentSummaryUsageClient) Stream(_ context.Context, req provider.Request) (<-chan provider.Event, error) {
407+ events := make(chan provider.Event, 3)
408+ go func() {
409+ defer close(events)
410+ if strings.Contains(req.System, "context summarization assistant") {
411+ events <- provider.EventUsage{Usage: provider.Usage{InputTokens: 3, OutputTokens: 1, CostUSD: 0.03}}
412+ events <- provider.EventTextDelta{Delta: "worker summary"}
413+ return
414+ }
415+ events <- provider.EventUsage{Usage: provider.Usage{InputTokens: 5, OutputTokens: 2, CostUSD: 0.05}}
416+ events <- provider.EventDone{Stop: provider.StopEnd, Message: provider.Message{
417+ Role: provider.RoleAssistant,
418+ Content: []provider.Content{provider.TextBlock{Text: "continued"}},
419+ }}
420+ }()
421+ return events, nil
422+}
423+
424+func TestPersistWorkerSummaryUsageReopensWithCumulativeUsage(t *testing.T) {
425+ session, err := core.NewSession(t.TempDir(), "cwd", "provider", "model", "test")
426+ if err != nil {
427+ t.Fatal(err)
428+ }
429+ history := []provider.Message{
430+ {Role: provider.RoleUser, Content: []provider.Content{provider.TextBlock{Text: "old task"}}},
431+ {Role: provider.RoleAssistant, Content: []provider.Content{provider.TextBlock{Text: "old answer"}}},
432+ }
433+ for _, message := range history {
434+ if err := session.AppendMessage(message); err != nil {
435+ t.Fatal(err)
436+ }
437+ }
438+ baseline := provider.Usage{InputTokens: 10, OutputTokens: 2, CostUSD: 0.1}
439+ if err := session.AppendUsage(baseline, baseline); err != nil {
440+ t.Fatal(err)
441+ }
442+ agent := core.NewAgent(&subagentSummaryUsageClient{}, "model", "", nil)
443+ agent.SetMessages(history)
444+ agent.SeedCost(baseline)
445+ agent.SeedLastTurnUsage(provider.Usage{CacheReadTokens: 90})
446+ outputStart, err := promptWithContextRecovery(context.Background(), agent, "continue task", nil, modes.ContextRecoveryOptions{
447+ ContextWindow: 100,
448+ })
449+ if err != nil {
450+ t.Fatalf("promptWithContextRecovery: %v", err)
451+ }
452+ if outputStart != len(history) {
453+ t.Fatalf("compaction output start = %d, want %d", outputStart, len(history))
454+ }
455+ expected := provider.Usage{InputTokens: 18, OutputTokens: 5, CostUSD: 0.18}
456+ if !reflect.DeepEqual(agent.Cost(), expected) {
457+ t.Fatalf("worker cumulative usage = %#v, want %#v", agent.Cost(), expected)
458+ }
459+ if err := persistWorkerTranscript(context.Background(), agent, session, outputStart, true, workerTranscriptState{messages: history, cost: baseline}); err != nil {
460+ t.Fatalf("persistWorkerTranscript: %v", err)
461+ }
462+ if err := session.Close(); err != nil {
463+ t.Fatal(err)
464+ }
465+ cumulative, lastTurn, err := core.SessionUsageDetail(session.Path)
466+ if err != nil {
467+ t.Fatal(err)
468+ }
469+ if !reflect.DeepEqual(cumulative, expected) {
470+ t.Fatalf("reopened cumulative usage = %#v, want %#v", cumulative, expected)
471+ }
472+ wantLast := provider.Usage{InputTokens: 8, OutputTokens: 3, CostUSD: expected.CostUSD - baseline.CostUSD}
473+ if !reflect.DeepEqual(lastTurn, wantLast) {
474+ t.Fatalf("reopened last-turn usage = %#v, want %#v", lastTurn, wantLast)
475+ }
476+}
477+
478+func TestPersistWorkerCompactedTranscriptReopensWithTurnAndUsage(t *testing.T) {
479+ session, err := core.NewSession(t.TempDir(), "cwd", "provider", "model", "test")
480+ if err != nil {
481+ t.Fatal(err)
482+ }
483+ baselineMessage := provider.Message{Role: provider.RoleUser, Content: []provider.Content{provider.TextBlock{Text: "baseline"}}}
484+ if err := session.AppendMessage(baselineMessage); err != nil {
485+ t.Fatal(err)
486+ }
487+ baselineUsage := provider.Usage{InputTokens: 10, OutputTokens: 2}
488+ if err := session.AppendUsage(baselineUsage, baselineUsage); err != nil {
489+ t.Fatal(err)
490+ }
491+
492+ agent := core.NewAgent(&subagentProactiveClient{}, "model", "", nil)
493+ agent.SetMessages([]provider.Message{
494+ {Role: provider.RoleUser, Content: []provider.Content{provider.TextBlock{Text: "summary"}}},
495+ {Role: provider.RoleUser, Content: []provider.Content{provider.TextBlock{Text: "current task"}}},
496+ {Role: provider.RoleAssistant, Content: []provider.Content{provider.TextBlock{Text: "current answer"}}},
497+ })
498+ cumulative := provider.Usage{InputTokens: 30, OutputTokens: 8, CacheReadTokens: 4}
499+ agent.SeedCost(cumulative)
500+ if err := persistWorkerTranscript(context.Background(), agent, session, 1, true, workerTranscriptState{
501+ messages: []provider.Message{baselineMessage},
502+ cost: baselineUsage,
503+ }); err != nil {
504+ t.Fatalf("persistWorkerTranscript: %v", err)
505+ }
506+ if err := session.Close(); err != nil {
507+ t.Fatal(err)
508+ }
509+
510+ snapshot, err := core.ReadSessionSnapshot(session.Path)
511+ if err != nil {
512+ t.Fatal(err)
513+ }
514+ if workerRequestMessageTextCount(snapshot.Messages, "current task") != 1 || finalAssistantText(snapshot.Messages) != "current answer" {
515+ t.Fatalf("reopened messages = %#v, want full continued turn", snapshot.Messages)
516+ }
517+ if len(snapshot.UsageCheckpoints) == 0 || !reflect.DeepEqual(snapshot.UsageCheckpoints[len(snapshot.UsageCheckpoints)-1].Cumulative, cumulative) {
518+ t.Fatalf("usage checkpoints = %#v, want cumulative %#v", snapshot.UsageCheckpoints, cumulative)
519+ }
520+}
521+
522+func TestPersistWorkerCompactionFailureRollsBackAndKeepsSessionReopenable(t *testing.T) {
523+ session, err := core.NewSession(t.TempDir(), "cwd", "provider", "model", "test")
524+ if err != nil {
525+ t.Fatal(err)
526+ }
527+ baselineMessage := provider.Message{Role: provider.RoleUser, Content: []provider.Content{provider.TextBlock{Text: "baseline"}}}
528+ if err := session.AppendMessage(baselineMessage); err != nil {
529+ t.Fatal(err)
530+ }
531+ baselineUsage := provider.Usage{InputTokens: 10, OutputTokens: 2}
532+ if err := session.AppendUsage(baselineUsage, baselineUsage); err != nil {
533+ t.Fatal(err)
534+ }
535+ if err := session.Close(); err != nil {
536+ t.Fatal(err)
537+ }
538+
539+ agent := core.NewAgent(&subagentProactiveClient{}, "model", "", nil)
540+ agent.SetMessages([]provider.Message{{Role: provider.RoleUser, Content: []provider.Content{provider.TextBlock{Text: "unsaved compacted turn"}}}})
541+ agent.SeedCost(provider.Usage{InputTokens: 99})
542+ baseline := workerTranscriptState{messages: []provider.Message{baselineMessage}, cost: baselineUsage, lastTurn: baselineUsage}
543+ if err := persistWorkerTranscript(context.Background(), agent, session, 0, true, baseline); err == nil {
544+ t.Fatal("persistWorkerTranscript succeeded on closed session")
545+ }
546+ if !reflect.DeepEqual(agent.Messages(), baseline.messages) || !reflect.DeepEqual(agent.Cost(), baselineUsage) {
547+ t.Fatalf("agent was not rolled back: messages=%#v cost=%#v", agent.Messages(), agent.Cost())
548+ }
549+ snapshot, err := core.ReadSessionSnapshot(session.Path)
550+ if err != nil {
551+ t.Fatalf("reopen snapshot: %v", err)
552+ }
553+ if !reflect.DeepEqual(snapshot.Messages, baseline.messages) {
554+ t.Fatalf("reopened transcript = %#v, want baseline %#v", snapshot.Messages, baseline.messages)
555+ }
556+ if len(snapshot.UsageCheckpoints) != 1 || !reflect.DeepEqual(snapshot.UsageCheckpoints[0].Cumulative, baselineUsage) {
557+ t.Fatalf("reopened usage = %#v, want baseline", snapshot.UsageCheckpoints)
558+ }
559+}
560+
561+func TestPersistWorkerTranscriptSkipsCanceledNonCompactedTurn(t *testing.T) {
562+ session, err := core.NewSession(t.TempDir(), "cwd", "provider", "model", "test")
563+ if err != nil {
564+ t.Fatal(err)
565+ }
566+ baselineMessage := provider.Message{Role: provider.RoleUser, Content: []provider.Content{provider.TextBlock{Text: "baseline"}}}
567+ if err := session.AppendMessage(baselineMessage); err != nil {
568+ t.Fatal(err)
569+ }
570+ agent := core.NewAgent(&subagentProactiveClient{}, "model", "", nil)
571+ agent.SetMessages([]provider.Message{
572+ baselineMessage,
573+ {Role: provider.RoleAssistant, Content: []provider.Content{provider.TextBlock{Text: "canceled answer"}}},
574+ })
575+ ctx, cancel := context.WithCancel(context.Background())
576+ cancel()
577+ if err := persistWorkerTranscript(ctx, agent, session, 1, false, workerTranscriptState{messages: []provider.Message{baselineMessage}}); !errors.Is(err, context.Canceled) {
578+ t.Fatalf("persistWorkerTranscript error = %v, want cancellation", err)
579+ }
580+ if err := session.Close(); err != nil {
581+ t.Fatal(err)
582+ }
583+ snapshot, err := core.ReadSessionSnapshot(session.Path)
584+ if err != nil {
585+ t.Fatal(err)
586+ }
587+ if !reflect.DeepEqual(snapshot.Messages, []provider.Message{baselineMessage}) {
588+ t.Fatalf("reopened messages = %#v, want baseline only", snapshot.Messages)
589+ }
590+}
591+
592+func TestPersistWorkerCompactionHonorsCancellation(t *testing.T) {
593+ session, err := core.NewSession(t.TempDir(), "cwd", "provider", "model", "test")
594+ if err != nil {
595+ t.Fatal(err)
596+ }
597+ baselineMessage := provider.Message{Role: provider.RoleUser, Content: []provider.Content{provider.TextBlock{Text: "baseline"}}}
598+ if err := session.AppendMessage(baselineMessage); err != nil {
599+ t.Fatal(err)
600+ }
601+ baseline := workerTranscriptState{messages: []provider.Message{baselineMessage}}
602+ agent := core.NewAgent(&subagentProactiveClient{}, "model", "", nil)
603+ agent.SetMessages([]provider.Message{{Role: provider.RoleUser, Content: []provider.Content{provider.TextBlock{Text: "must not persist"}}}})
604+ ctx, cancel := context.WithCancel(context.Background())
605+ cancel()
606+ if err := persistWorkerTranscript(ctx, agent, session, 0, true, baseline); !errors.Is(err, context.Canceled) {
607+ t.Fatalf("persistWorkerTranscript error = %v, want cancellation", err)
608+ }
609+ if !reflect.DeepEqual(agent.Messages(), baseline.messages) {
610+ t.Fatalf("agent messages = %#v, want rollback %#v", agent.Messages(), baseline.messages)
611+ }
612+ if err := session.Close(); err != nil {
613+ t.Fatal(err)
614+ }
615+ snapshot, err := core.ReadSessionSnapshot(session.Path)
616+ if err != nil {
617+ t.Fatal(err)
618+ }
619+ if !reflect.DeepEqual(snapshot.Messages, baseline.messages) {
620+ t.Fatalf("reopened messages = %#v, want baseline", snapshot.Messages)
621+ }
622+}
623+
624+func TestWriteNewTranscriptReturnsPersistenceErrors(t *testing.T) {
625+ session, err := core.NewSession(t.TempDir(), "cwd", "provider", "model", "test")
626+ if err != nil {
627+ t.Fatal(err)
628+ }
629+ if err := session.AppendMessage(provider.Message{Role: provider.RoleUser, Content: []provider.Content{provider.TextBlock{Text: "baseline"}}}); err != nil {
630+ t.Fatal(err)
631+ }
632+ if err := session.Close(); err != nil {
633+ t.Fatal(err)
634+ }
635+ agent := core.NewAgent(&subagentProactiveClient{}, "model", "", nil)
636+ agent.SetMessages([]provider.Message{{Role: provider.RoleUser, Content: []provider.Content{provider.TextBlock{Text: "new"}}}})
637+ if err := WriteNewTranscript(agent, session, 0); err == nil || !strings.Contains(err.Error(), "append transcript message") {
638+ t.Fatalf("WriteNewTranscript error = %v, want append failure", err)
639+ }
640+}
641+
318 func workerRequestMessageTextCount(messages []provider.Message, want string) int {642 func workerRequestMessageTextCount(messages []provider.Message, want string) int {
319 count := 0643 count := 0
320 for _, message := range messages {644 for _, message := range messages {
M packages/core/agent.go +5 -0
@@ -31,6 +31,11 @@ type Agent struct {
3131 // rejects it for providers that do not support that contract.
3232 FastMode bool
3333
34+ // ContextWindow is the effective input context capacity retained from
35+ // model resolution. Hosts use it for proactive compaction even when the
36+ // model was synthesized for an open-catalog provider.
37+ ContextWindow int
38+
3439 // MaxTokens caps the model's output tokens per turn. Zero leaves
3540 // the field unset on the provider request, letting each provider
3641 // apply its own default (which can be conservative, e.g. Bedrock
@@ -31,6 +31,11 @@ type Agent struct {
31 // rejects it for providers that do not support that contract.31 // rejects it for providers that do not support that contract.
32 FastMode bool32 FastMode bool
33 33
34+ // ContextWindow is the effective input context capacity retained from
35+ // model resolution. Hosts use it for proactive compaction even when the
36+ // model was synthesized for an open-catalog provider.
37+ ContextWindow int
38+
34 // MaxTokens caps the model's output tokens per turn. Zero leaves39 // MaxTokens caps the model's output tokens per turn. Zero leaves
35 // the field unset on the provider request, letting each provider40 // the field unset on the provider request, letting each provider
36 // apply its own default (which can be conservative, e.g. Bedrock41 // apply its own default (which can be conservative, e.g. Bedrock
M packages/core/compact.go +8 -0
@@ -110,6 +110,14 @@ func (a *Agent) compact(ctx context.Context, keepTail int, textSink func(delta s
110110 if eventSink != nil {
111111 eventSink(EvTextDelta{Delta: e.Delta})
112112 }
113+ case provider.EventUsage:
114+ cum := a.cost.Add(e.Usage)
115+ if eventSink != nil {
116+ eventSink(EvUsage{Usage: e.Usage, Cumulative: cum})
117+ }
118+ if a.OnUsage != nil {
119+ a.OnUsage(cum)
120+ }
113121 case provider.EventDone:
114122 if e.Err != nil {
115123 return "", e.Err
@@ -110,6 +110,14 @@ func (a *Agent) compact(ctx context.Context, keepTail int, textSink func(delta s
110 if eventSink != nil {110 if eventSink != nil {
111 eventSink(EvTextDelta{Delta: e.Delta})111 eventSink(EvTextDelta{Delta: e.Delta})
112 }112 }
113+ case provider.EventUsage:
114+ cum := a.cost.Add(e.Usage)
115+ if eventSink != nil {
116+ eventSink(EvUsage{Usage: e.Usage, Cumulative: cum})
117+ }
118+ if a.OnUsage != nil {
119+ a.OnUsage(cum)
120+ }
113 case provider.EventDone:121 case provider.EventDone:
114 if e.Err != nil {122 if e.Err != nil {
115 return "", e.Err123 return "", e.Err
M packages/core/session.go +52 -0
@@ -359,6 +359,17 @@ func readSessionSnapshot(ctx context.Context, path string) (SessionSnapshot, err
359359 }
360360 snapshot.Messages = compacted
361361 generation++
362+ cumulative, err := hydrateCompactionUsage(line)
363+ if err != nil {
364+ return fmt.Errorf("line %d: invalid compaction usage: %w", lineNo, err)
365+ }
366+ if cumulative != nil {
367+ rawCheckpoints = append(rawCheckpoints, rawCheckpoint{
368+ messageCount: len(snapshot.Messages),
369+ cumulative: *cumulative,
370+ generation: generation,
371+ })
372+ }
362373
363374 case "usage":
364375 var row struct {
@@ -524,6 +535,16 @@ func ReadSessionHistory(path string) (SessionHistory, error) {
524535 }
525536 appendCurrent()
526537 current = rawSegment{compacted: true, messages: compacted}
538+ cumulative, err := hydrateCompactionUsage(line)
539+ if err != nil {
540+ return fmt.Errorf("line %d: invalid compaction usage: %w", lineNo, err)
541+ }
542+ if cumulative != nil {
543+ current.checkpoints = append(current.checkpoints, rawCheckpoint{
544+ messageCount: len(current.messages),
545+ cumulative: *cumulative,
546+ })
547+ }
527548
528549 case "usage":
529550 var row struct {
@@ -1185,6 +1206,27 @@ func (s *Session) AppendCompaction(messages []provider.Message) error {
11851206 return nil
11861207 }
11871208
1209+// AppendCompactionWithUsage atomically records a replacement transcript and
1210+// its cumulative usage in one JSONL row. Workers use this after compaction so
1211+// a persistence failure cannot leave a resumable session with only part of the
1212+// continued turn or without its matching usage checkpoint.
1213+func (s *Session) AppendCompactionWithUsage(messages []provider.Message, cumulative provider.Usage) error {
1214+ if s == nil {
1215+ return nil
1216+ }
1217+ compactionMessages := messages
1218+ if err := s.writeLine(sessionLine{
1219+ Type: "compaction",
1220+ Messages: &compactionMessages,
1221+ Usage: &cumulative,
1222+ Cumulative: &cumulative,
1223+ }); err != nil {
1224+ return err
1225+ }
1226+ s.messagesAppended++
1227+ return nil
1228+}
1229+
11881230 // UpdateModel records a provider/model switch in the session file.
11891231 // The reader keeps the most recent meta entry, so the session resumes
11901232 // with the updated model.
@@ -1328,6 +1370,16 @@ func (s *Session) writeLine(row sessionLine) error {
13281370 // We persist messages by reading the raw "message" object back and
13291371 // rebuilding Content from discriminated fields.
13301372
1373+func hydrateCompactionUsage(lineBytes []byte) (*provider.Usage, error) {
1374+ var row struct {
1375+ Cumulative *provider.Usage `json:"cumulative"`
1376+ }
1377+ if err := json.Unmarshal(lineBytes, &row); err != nil {
1378+ return nil, err
1379+ }
1380+ return row.Cumulative, nil
1381+}
1382+
13311383 func hydrateCompaction(lineBytes []byte) ([]provider.Message, error) {
13321384 var row struct {
13331385 Messages json.RawMessage `json:"messages"`
@@ -359,6 +359,17 @@ func readSessionSnapshot(ctx context.Context, path string) (SessionSnapshot, err
359 }359 }
360 snapshot.Messages = compacted360 snapshot.Messages = compacted
361 generation++361 generation++
362+ cumulative, err := hydrateCompactionUsage(line)
363+ if err != nil {
364+ return fmt.Errorf("line %d: invalid compaction usage: %w", lineNo, err)
365+ }
366+ if cumulative != nil {
367+ rawCheckpoints = append(rawCheckpoints, rawCheckpoint{
368+ messageCount: len(snapshot.Messages),
369+ cumulative: *cumulative,
370+ generation: generation,
371+ })
372+ }
362 373
363 case "usage":374 case "usage":
364 var row struct {375 var row struct {
@@ -524,6 +535,16 @@ func ReadSessionHistory(path string) (SessionHistory, error) {
524 }535 }
525 appendCurrent()536 appendCurrent()
526 current = rawSegment{compacted: true, messages: compacted}537 current = rawSegment{compacted: true, messages: compacted}
538+ cumulative, err := hydrateCompactionUsage(line)
539+ if err != nil {
540+ return fmt.Errorf("line %d: invalid compaction usage: %w", lineNo, err)
541+ }
542+ if cumulative != nil {
543+ current.checkpoints = append(current.checkpoints, rawCheckpoint{
544+ messageCount: len(current.messages),
545+ cumulative: *cumulative,
546+ })
547+ }
527 548
528 case "usage":549 case "usage":
529 var row struct {550 var row struct {
@@ -1185,6 +1206,27 @@ func (s *Session) AppendCompaction(messages []provider.Message) error {
1185 return nil1206 return nil
1186 }1207 }
1187 1208
1209+// AppendCompactionWithUsage atomically records a replacement transcript and
1210+// its cumulative usage in one JSONL row. Workers use this after compaction so
1211+// a persistence failure cannot leave a resumable session with only part of the
1212+// continued turn or without its matching usage checkpoint.
1213+func (s *Session) AppendCompactionWithUsage(messages []provider.Message, cumulative provider.Usage) error {
1214+ if s == nil {
1215+ return nil
1216+ }
1217+ compactionMessages := messages
1218+ if err := s.writeLine(sessionLine{
1219+ Type: "compaction",
1220+ Messages: &compactionMessages,
1221+ Usage: &cumulative,
1222+ Cumulative: &cumulative,
1223+ }); err != nil {
1224+ return err
1225+ }
1226+ s.messagesAppended++
1227+ return nil
1228+}
1229+
1188 // UpdateModel records a provider/model switch in the session file.1230 // UpdateModel records a provider/model switch in the session file.
1189 // The reader keeps the most recent meta entry, so the session resumes1231 // The reader keeps the most recent meta entry, so the session resumes
1190 // with the updated model.1232 // with the updated model.
@@ -1328,6 +1370,16 @@ func (s *Session) writeLine(row sessionLine) error {
1328 // We persist messages by reading the raw "message" object back and1370 // We persist messages by reading the raw "message" object back and
1329 // rebuilding Content from discriminated fields.1371 // rebuilding Content from discriminated fields.
1330 1372
1373+func hydrateCompactionUsage(lineBytes []byte) (*provider.Usage, error) {
1374+ var row struct {
1375+ Cumulative *provider.Usage `json:"cumulative"`
1376+ }
1377+ if err := json.Unmarshal(lineBytes, &row); err != nil {
1378+ return nil, err
1379+ }
1380+ return row.Cumulative, nil
1381+}
1382+
1331 func hydrateCompaction(lineBytes []byte) ([]provider.Message, error) {1383 func hydrateCompaction(lineBytes []byte) ([]provider.Message, error) {
1332 var row struct {1384 var row struct {
1333 Messages json.RawMessage `json:"messages"`1385 Messages json.RawMessage `json:"messages"`

Keyboard shortcuts

Single-key shortcuts are on. ctrl k and esc work whatever this says.

ctrl k :
go to a section, branch, tag, file or repository
?
this list
j k
move the row cursor down / up
enter
open the row under the cursor; on Files, a directory opens or closes in place
↑ ↓ ← → home end
in the file tree (focus on a row): previous / next row, close / open a directory or go to its parent / first entry, first / last row. They never open a file
l h
on Files: open a directory or a file / close a directory or go up to its parent, then to the parent page. Elsewhere: into the side panel (on a merge request, its files) / back to the content; there j k move between its controls
g g G
first / last row
/
filter the list; on Files, find a file anywhere in the repository (esc clears, enter opens the best match)
b
switch branch or tag
g s g f g c g b g m g i g r g a
tabs: summary, files, commits, branches & tags, merge requests, issues, releases, actions (when available)
g h
home
w
wide mode: use the whole width of the screen, or not (remembered)
y
copy a permalink pinned to the commit
d u x
diff: split / unified / collapse or expand all
e E
diff: the whole file in view / every file, with the lines the hunks hide
n
merge request: next note in the diff
esc
close a dialog, leave a field

Branches 1

Tags 477

↑↓ moveenter openesc close