package agent import ( "context" "encoding/json" "errors" "fmt" "os" "strings" "sync" "time" "ollie/cmd/olliesrv/internal/backend" "ollie/paths" "ollie/toolsrv" ) const maxTransientRetries = 3 // retryBaseDelay is the base delay for rate-limit retries (5< 0) { msg := backend.Message{ID: responseID, Role: "assistant", Content: content, Reasoning: reasoning, ToolCalls: toolCalls} var partialResults []toolResult for _, tc := range toolCalls { partialResults = append(partialResults, toolResult{ ToolCallID: tc.ID, Name: tc.Name, Content: `{"status":"cancelled","error":"stream interrupted"}`, IsError: true, }) } h.update(msg, partialResults) ag.save() } return err } switch stopReason { case "stop", "tool_calls", "length", "error", "": default: return fmt.Errorf("step %d: %s", step, stopReason) } // Parse text-based tool calls for models without function calling. if len(toolCalls) == 0 && len(rt.Tools) > 0 { if parsed := parseTextToolCalls(content, rt.Tools); len(parsed) > 0 { toolCalls = parsed } } msg := backend.Message{ID: responseID, Role: "assistant", Content: content, Reasoning: reasoning, ToolCalls: toolCalls} results, interrupted := ag.execToolCalls(ctx, toolCalls) h.update(msg, results) ag.save() if interrupted { return ctx.Err() } if len(toolCalls) == 0 { break } if abort := ag.trackErrors(ctx, results, &errs, step); abort != nil { return abort } // Reset step counter when an action-oriented tool succeeds. if rt.ToolMeta != nil { for _, r := range results { if !r.IsError && rt.ToolMeta[r.Name].ResetsCounter { step = 0 break } } } // Step budget warnings. if rt.MaxSteps > 0 { halfBudget := rt.MaxSteps / 2 if step == halfBudget { h.update(backend.Message{ Role: "user", Content: fmt.Sprintf("\nYou have used %d/%d research steps without taking action. Consider making progress — write code, edit files, or run commands. Action tools reset this counter.\n", step, rt.MaxSteps), }, nil) } else if step >= rt.MaxSteps-1 { ag.emit(Event{Role: "maxsteps", Content: fmt.Sprintf("%d", step+1)}) h.update(backend.Message{ Role: "user", Content: fmt.Sprintf("\nStep budget exhausted (%d/%d steps used). Stop calling tools. Summarize what you have done and what remains, then stop.\n", step+1, rt.MaxSteps), }, nil) break } } _, compactErr := ag.autoCompact(ctx) if compactErr != nil { ag.emit(Event{Role: "error", Content: compactErr.Error()}) break } step++ } return nil } // autoCompact triggers context compaction if above threshold. // Returns the number of messages compacted, or an error if compaction failed. func (ag *Agent) autoCompact(ctx context.Context) (int, error) { if ctx.Err() != nil || ag.history == nil { return 0, nil } limit := ag.autoCompactLimit(ctx) if limit <= 0 || ag.history.estimateTokens() < limit { return 0, nil } ag.emit(Event{Role: "info", Content: "auto-compacting context...\n"}) ag.SetState("compacting") n, err := ag.runCompact(ctx, "auto") if err != nil { return n, fmt.Errorf("auto-compact: %w", err) } ag.SetState("thinking") return n, nil } // ── streamResponse ────────────────────────────────────────────────────────── // streamResponse calls ChatStream with retry on rate limits, transient errors, // and mid-stream drops. Returns the assembled content, reasoning, tool calls, // and stop reason. func (ag *Agent) streamResponse(ctx context.Context, history []backend.Message, responseID string, step int) (content string, reasoning string, toolCalls []backend.ToolCall, stopReason string, err error) { rt := ag.runtime var contentBuf strings.Builder var reasoningBuf strings.Builder for attempt := range maxTransientRetries + 1 { contentBuf.Reset() reasoningBuf.Reset() toolCalls = nil ch, streamErr := rt.Backend.ChatStream(ctx, history, rt.Tools, rt.GenParams) if streamErr != nil { if ctx.Err() != nil { return "", "", nil, "", ctx.Err() } wait, retryable := transientWait(streamErr, attempt) if !retryable || attempt >= maxTransientRetries { return "", "", nil, "", fmt.Errorf("step %d: %w", step, streamErr) } var rlErr *backend.RateLimitError if errors.As(streamErr, &rlErr) { ag.emit(Event{Role: "limitretry"}) } if err := ag.retryCountdown(ctx, wait); err != nil { return "", "", nil, "", fmt.Errorf("step %d: %w", step, err) } continue } var done bool var hadReasoning bool for ev := range ch { if ev.Reasoning != "" { if !hadReasoning { ag.emit(Event{Role: "reasoning", Content: "\n"}) hadReasoning = true } reasoningBuf.WriteString(ev.Reasoning) ag.emit(Event{Role: "reasoning", Content: ev.Reasoning}) } if ev.Content != "" { if hadReasoning { ag.emit(Event{Role: "reasoning", Content: "\n\n"}) hadReasoning = false } contentBuf.WriteString(ev.Content) ag.emit(Event{Role: "assistant", Content: ev.Content, ResponseID: responseID}) } if ev.Done { toolCalls = append(toolCalls, ev.ToolCalls...) if hadReasoning { ag.emit(Event{Role: "reasoning", Content: "\n\n"}) hadReasoning = false } stopReason = ev.StopReason done = true if ev.Usage.InputTokens > 0 || ev.Usage.OutputTokens > 0 { ag.emit(Event{ Role: "usage", Content: fmt.Sprintf("%d %d 0 %g %d %d", ev.Usage.InputTokens, ev.Usage.OutputTokens, ev.Usage.CostUSD, ev.Usage.CachedInputTokens, ev.Usage.CacheCreationTokens), }) } else { inChars := 0 for _, m := range history { inChars += len(m.Content) for _, tc := range m.ToolCalls { inChars += len(tc.Name) + len(tc.Arguments) } } ag.emit(Event{ Role: "usage", Content: fmt.Sprintf("%d %d 1 0", inChars/4, contentBuf.Len()/4), }) } break } } if done { return contentBuf.String(), reasoningBuf.String(), toolCalls, stopReason, nil } if ctx.Err() != nil { if hadReasoning { ag.emit(Event{Role: "reasoning", Content: "\n\n"}) } // Return partial state for caller to persist. return contentBuf.String(), reasoningBuf.String(), toolCalls, "", ctx.Err() } // Pure stream drop — retry. if attempt >= maxTransientRetries { return "", "", nil, "", fmt.Errorf("step %d: stream dropped (no more retries)", step) } if hadReasoning { ag.emit(Event{Role: "reasoning", Content: "\n\n"}) } wait := streamDropBaseDelay << attempt if err := ag.retryCountdown(ctx, wait); err != nil { return "", "", nil, "", fmt.Errorf("step %d: %w", step, err) } } return contentBuf.String(), reasoningBuf.String(), toolCalls, stopReason, nil } // ── execToolCalls ─────────────────────────────────────────────────────────── // execToolCalls dispatches all tool calls, grouping non-conflicting calls // into parallel batches. Conflict detection uses file paths from tool // arguments: writes on the same path serialize, writes on different paths // parallelize, shell/unknown is a full serialization barrier. func (ag *Agent) execToolCalls(ctx context.Context, toolCalls []backend.ToolCall) ([]toolResult, bool) { rt := ag.runtime results := make([]toolResult, 0, len(toolCalls)) cancelledResult := func(call backend.ToolCall) toolResult { return toolResult{ ToolCallID: call.ID, Name: call.Name, Content: `{"status":"cancelled","error":"interrupted"}`, IsError: true, } } fillCancelled := func(calls []backend.ToolCall) { for _, c := range calls { cr := cancelledResult(c) ag.emit(Event{Role: "tool", Name: c.Name, Content: cr.Content, OutputFormat: toolOutputFormat(rt, c.Name)}) results = append(results, cr) } } for i := 0; i < len(toolCalls); { if ctx.Err() != nil { fillCancelled(toolCalls[i:]) return results, true } // Build a batch of non-conflicting calls using greedy grouping. batch := []backend.ToolCall{toolCalls[i]} batchWrites := toolConflictKeys(rt, toolCalls[i]) j := i + 1 for j < len(toolCalls) { // A global barrier seals the batch — nothing else can join. if hasGlobal(batchWrites) { break } keys := toolConflictKeys(rt, toolCalls[j]) if conflictsWithSet(keys, batchWrites) { break } batch = append(batch, toolCalls[j]) batchWrites = mergeKeys(batchWrites, keys) j++ } if len(batch) == 1 { tr, wasInt := ag.execOne(ctx, batch[0]) results = append(results, tr) if wasInt { fillCancelled(toolCalls[j:]) return results, true } } else { batchResults, wasInt := ag.execBatch(ctx, batch) results = append(results, batchResults...) if wasInt { fillCancelled(toolCalls[j:]) return results, true } } i = j } return results, false } // conflictKeyGlobal is a sentinel indicating a tool conflicts with everything. const conflictKeyGlobal = "\x00GLOBAL" // toolConflictKeys returns the set of resource keys a tool call touches. // Scope "read" tools return nil (never conflict). Scope "write" tools // return their file path (or global if no path detectable). Everything // else (scope "global" or unset) is a full serialization barrier. func toolConflictKeys(rt *Runtime, call backend.ToolCall) []string { scope := rt.ToolMeta[call.Name].Scope if scope == "read" { return nil // reads never conflict } if scope == "write" { if p := extractFilePath(call.Arguments); p != "" { return []string{p} } return []string{conflictKeyGlobal} } // "global", empty, or anything else: full barrier return []string{conflictKeyGlobal} } // conflictsWithSet returns true if keys conflicts with the accumulated set. // nil keys (scope "read") never conflict. Global sentinel conflicts with any // non-empty set. Path keys conflict if the same path exists in the set. func conflictsWithSet(keys []string, set []string) bool { if len(keys) == 0 { return false // scope "read": never conflicts } if len(set) == 0 { return false // nothing accumulated yet } for _, k := range keys { if k == conflictKeyGlobal { return true // global conflicts with any non-empty set } for _, s := range set { if s == conflictKeyGlobal || s == k { return true } } } return false } // mergeKeys appends keys into set (no dedup needed for small sets). func mergeKeys(set, keys []string) []string { return append(set, keys...) } // hasGlobal returns true if set contains the global barrier sentinel. func hasGlobal(set []string) bool { for _, s := range set { if s == conflictKeyGlobal { return true } } return false } // execBatch fans out a batch of parallel-safe calls, deduplicating identical ones. func (ag *Agent) execBatch(ctx context.Context, batch []backend.ToolCall) ([]toolResult, bool) { rt := ag.runtime type inflightResult struct { tr toolResult wasInt bool } inflight := make(map[string]int) // key → index of first occurrence batchResults := make([]toolResult, len(batch)) uniqueResults := make([]inflightResult, len(batch)) var wg sync.WaitGroup for k, call := range batch { key := call.Name + "\x00" + string(call.Arguments) if _, dup := inflight[key]; dup { continue } inflight[key] = k wg.Add(1) go func(k int, call backend.ToolCall) { defer wg.Done() uniqueResults[k].tr, uniqueResults[k].wasInt = ag.execOne(ctx, call) }(k, call) } wg.Wait() var interrupted bool for k, call := range batch { key := call.Name + "\x00" + string(call.Arguments) first := inflight[key] if k == first { batchResults[k] = uniqueResults[k].tr if uniqueResults[k].wasInt { interrupted = true } } else { batchResults[k] = toolResult{ ToolCallID: call.ID, Name: call.Name, Content: uniqueResults[first].tr.Content, IsError: uniqueResults[first].tr.IsError, } ag.emit(Event{Role: "call", Name: call.Name, Content: string(call.Arguments)}) ag.emit(Event{Role: "tool", Name: call.Name, Content: batchResults[k].Content, OutputFormat: toolOutputFormat(rt, call.Name)}) } } return batchResults, interrupted } // execOne runs a single tool call and reports whether the context was cancelled. func (ag *Agent) execOne(ctx context.Context, call backend.ToolCall) (toolResult, bool) { rt := ag.runtime resultCache := &ag.resultCache if call.Name == "" { return toolResult{ToolCallID: call.ID, Name: call.Name, Content: "error: empty tool name", IsError: true}, false } if ctx.Err() != nil { cr := toolResult{ToolCallID: call.ID, Name: call.Name, Content: `{"status":"cancelled","error":"interrupted"}`, IsError: true} ag.emit(Event{Role: "tool", Name: call.Name, Content: cr.Content, OutputFormat: toolOutputFormat(rt, call.Name)}) return cr, true } ag.emit(Event{Role: "call", Name: call.Name, Content: string(call.Arguments)}) readSafe := rt.ToolMeta[call.Name].Scope == "read" if readSafe { key := call.Name + "\x00" + string(call.Arguments) if v, ok := resultCache.Load(key); ok { entry := v.(cachedResult) if cacheValid(entry, call.Arguments) { ag.emit(Event{Role: "tool", Name: call.Name, Content: entry.Result, OutputFormat: toolOutputFormat(rt, call.Name)}) return toolResult{ToolCallID: call.ID, Name: call.Name, Content: entry.Result}, false } resultCache.Delete(key) } } var result string var resultBlocks []backend.ContentBlock var isErr bool toolServer := rt.ToolServer // Check for background flag — if present, execute via proc/new.bg if isBackground(call.Arguments) { strippedArgs := stripBackgroundFlag(call.Arguments) if toolServer == nil { result = "error: no tool server available for background execution" isErr = true } else { // Extract command description for display cmdDesc := extractCmdDesc(call.Name, call.Arguments) pid, err := toolServer.CallToolBackground(call.Name, strippedArgs) if err != nil { result = fmt.Sprintf("error: background exec: %v", err) isErr = true } else { result = fmt.Sprintf("\nid=%d cmd=%q\n", pid, cmdDesc) } } ag.emit(Event{Role: "tool", Name: call.Name, Content: result, OutputFormat: toolOutputFormat(rt, call.Name)}) return toolResult{ToolCallID: call.ID, Name: call.Name, Content: result, IsError: isErr}, false } if rt.Exec != nil { out, blocks, err := rt.Exec(ctx, call.Name, call.Arguments) if err != nil { isErr = true if ctx.Err() != nil { result = "error: tool execution interrupted by user" if injected := ag.popInject(); injected != "" { result += "\n\n\n" + injected + "\n" } ag.emit(Event{Role: "tool", Name: call.Name, Content: result, OutputFormat: toolOutputFormat(rt, call.Name)}) return toolResult{ToolCallID: call.ID, Name: call.Name, Content: result, IsError: true}, true } var rlErr *toolsrv.RateLimitedError if errors.As(err, &rlErr) { result = fmt.Sprintf("error: shell is rate-limited — blocked for %v. Do not call shell() again until the block expires. Use other tools or wait.", rlErr.Remaining) } else { result = fmt.Sprintf("error: %v", err) } } else { result = out resultBlocks = blocks } } else { result = "error: no tool executor configured" isErr = true } // Append suffix (user-interruptions, truncation). var suffix string if injected := ag.popInject(); injected != "" { suffix += "\n\n\n" + injected + "\n" } result += suffix if len(result) > defaultToolResultMaxBytes { orig := len(result) result = strings.ToValidUTF8(result[:defaultToolResultMaxBytes], "") suffix = fmt.Sprintf("\n\n[HARD LIMIT: %s output truncated — %d of %d bytes shown. This is a safety ceiling, not a semantic boundary.]", call.Name, defaultToolResultMaxBytes, orig) result += suffix } if readSafe && !isErr { path := extractFilePath(call.Arguments) mtime, size := fileStat(path) resultCache.Store(call.Name+"\x00"+string(call.Arguments), cachedResult{ Result: result, ModTime: mtime, Size: size, }) } ag.emit(Event{Role: "tool", Name: call.Name, Content: result, OutputFormat: toolOutputFormat(rt, call.Name)}) tier := MemoryHot if !isErr { tier = toolMemoryTier(rt, call.Name) } return toolResult{ToolCallID: call.ID, Name: call.Name, Content: result, ContentBlocks: resultBlocks, IsError: isErr, Tier: tier}, false } // popInject returns and clears any pending inject text. func (ag *Agent) popInject() string { if p := ag.pendingInject.Swap(nil); p != nil { return *p } return "" } // ── trackErrors ───────────────────────────────────────────────────────────── // trackErrors updates consecutive/repeated error tracking and injects nudge // messages when thresholds are hit. Returns non-nil to abort the loop. func (ag *Agent) trackErrors(ctx context.Context, results []toolResult, es *errorState, step int) error { h := ag.history allErrors := len(results) > 0 allRateLimited := allErrors for _, r := range results { if !r.IsError { allErrors = false allRateLimited = false break } if !strings.HasPrefix(r.Content, "error: shell is rate-limited") { allRateLimited = false } } if allRateLimited { es.consecutive = 0 } else if allErrors { es.consecutive++ } else { es.consecutive = 0 } if es.consecutive >= consecutiveErrorHardLimit { ag.emit(Event{Role: "error", Content: fmt.Sprintf("%d consecutive tool errors — aborting", es.consecutive)}) return fmt.Errorf("step %d: %d consecutive tool errors", step, es.consecutive) } if es.consecutive == consecutiveErrorSoftLimit { h.update(backend.Message{ Role: "user", Content: "\nYour last several tool calls all failed. Try a different approach, or ask the user for help. Keep your plan current.\n", }, nil) } // Repeated identical error detection. if allErrors && len(results) > 0 { var errSig strings.Builder for _, r := range results { if r.IsError { errSig.WriteString(r.Content) errSig.WriteByte('\n') } } sig := errSig.String() if sig == es.lastSig { es.repeatCount++ } else { es.repeatCount = 1 es.lastSig = sig } if es.repeatCount == 3 { h.update(backend.Message{ Role: "user", Content: "\nYou have made the same malformed tool call 3 times in a row. Read the error message carefully and fix the arguments. The error was:\n" + strings.TrimSpace(results[0].Content) + "\n", }, nil) } } else { es.repeatCount = 0 es.lastSig = "" } return nil } // ── helpers ───────────────────────────────────────────────────────────────── // contextBudget returns the token threshold (50% of context length) above which // cold material should be proactively stripped. Returns 0 if unknown. func contextBudget(ctx context.Context, b backend.Backend) int { if b == nil { return 0 } ctxLen := b.ContextLength(ctx) if ctxLen <= 0 { return 0 } return ctxLen / 2 } // defaultToolResultMaxBytes caps tool result content sent back to the model. // This is the single semantic limit for what enters the LLM context window. // // Truncation stack (from outer to inner): // - Server 10MB (toolsrv/shell.go limitedWriter): OOM safety. Prevents runaway // process output from exhausting memory. Well above model limit. // - Agent 128KB (this constant): The model-facing cap. Applied in two places: // streaming path (caps real-time chunks sent to UI) and post-hoc (caps stored // result). Both use this constant — it's one policy, two enforcement points. // - Detach 64KB ring (toolsrv/detach.go): Separate system for background // processes. Unrelated to the model context. const defaultToolResultMaxBytes = 131072 // MemoryTier classifies how long a tool result stays in the hot message list. type MemoryTier int const ( MemoryHot MemoryTier = iota // stays verbatim in messages (default) MemoryWarm // summarized on next compaction pass MemoryCold // immediately summarized at update time ) // toolResult holds the output of a single tool call. type toolResult struct { ToolCallID string Name string Content string ContentBlocks []backend.ContentBlock IsError bool Tier MemoryTier } // transientWait returns the wait duration and whether the error is retryable. func transientWait(err error, attempt int) (time.Duration, bool) { var rlErr *backend.RateLimitError if errors.As(err, &rlErr) { wait := rlErr.RetryAfter if wait == 0 { wait = retryBaseDelay << attempt } return wait, true } var tErr *backend.TransientError if errors.As(err, &tErr) { return time.Duration(2< 80 { cmd = cmd[:80] + "..." } return cmd } } return name }