diff --git a/agent/loop.go b/agent/loop.go
index db4d2d1..b55a53b 100644
--- a/agent/loop.go
+++ b/agent/loop.go
@@ -60,7 +60,6 @@ func extractFilePath(args json.RawMessage) string {
// cacheValid checks if a cached result is still valid by stat-ing the file.
func cacheValid(entry cachedResult, args json.RawMessage) bool {
if entry.Size < 0 {
- // No file was associated — always valid (non-file tool).
return true
}
path := extractFilePath(args)
@@ -69,7 +68,6 @@ func cacheValid(entry cachedResult, args json.RawMessage) bool {
}
info, err := os.Stat(path)
if err != nil {
- // File gone or unreadable — cache is stale.
return false
}
return info.ModTime().UnixNano() == entry.ModTime && info.Size() == entry.Size
@@ -113,22 +111,23 @@ func toolMemoryTier(rt *Runtime, name string) MemoryTier {
type TurnCtx struct {
context.Context
Output EventHandler
- PopInject func() string // returns and clears pending inject, or ""
- AutoCompact func(ctx context.Context) // called after each tool round; may compact in-place
- Save func() // called after each state.update(); persists mid-turn progress
+ PopInject func() string
+ AutoCompact func(ctx context.Context)
+ Save func()
IncrToolCallCount func() int64
ResultCache *sync.Map
}
+// errorState tracks consecutive and repeated tool errors across turns.
+type errorState struct {
+ consecutive int
+ lastSig string
+ repeatCount int
+}
+
func run(rt *Runtime, ctx TurnCtx, h *History) error {
var step int
- var consecutiveErrors int // rounds where every tool call returned an error
- var lastErrorSig string // signature of previous round's error(s) for repeat detection
- var repeatErrorCount int // consecutive rounds with identical error signature
- // resultCache stores outputs of read-safe tool calls keyed by name+args.
- // Only tools classified as parallel-safe (immutable reads) are cached.
- // sync.Map is required because execOne may run in concurrent goroutines.
- // Uses the cross-turn cache from TurnCtx if available.
+ var errs errorState
resultCache := ctx.ResultCache
if resultCache == nil {
resultCache = &sync.Map{}
@@ -136,7 +135,6 @@ func run(rt *Runtime, ctx TurnCtx, h *History) error {
for {
emit(ctx, Event{Role: "state", Content: "thinking"})
- // Proactive context gate: strip cold material before calling the backend.
if budget := contextBudget(ctx, rt.Backend); budget > 0 && h.estimateTokens() > budget {
h.stripCold(ctx, rt.Backend)
}
@@ -146,375 +144,40 @@ func run(rt *Runtime, ctx TurnCtx, h *History) error {
history = append([]backend.Message{{Role: "system", Content: preamble}}, history...)
}
- // Stream the assistant's response, retrying on rate limits, transient
- // backend errors (5xx, network), and mid-stream drops. One stable ID is
- // shared by all chunks and the completed assistant message.
responseID := paths.NewUUID()
- var content strings.Builder
- var reasoning strings.Builder
- var toolCalls []backend.ToolCall
- var stopReason string
-
- for attempt := range maxTransientRetries + 1 {
- content.Reset()
- reasoning.Reset()
- toolCalls = nil
-
- ch, err := rt.Backend.ChatStream(ctx, history, rt.Tools, rt.GenParams)
- if err != nil {
- if ctx.Err() != nil {
- return ctx.Err()
- }
- wait, retryable := transientWait(err, attempt)
- if !retryable || attempt >= maxTransientRetries {
- return fmt.Errorf("step %d: %w", step, err)
- }
- var rlErr *backend.RateLimitError
- if errors.As(err, &rlErr) {
- emit(ctx, Event{Role: "limitretry"})
- }
- if err := retryCountdown(ctx, wait); err != nil {
- return fmt.Errorf("step %d: %w", step, err)
- }
- continue
- }
-
- var done bool
- var hadReasoning bool
- for ev := range ch {
- if ev.Reasoning != "" {
- if !hadReasoning {
- emit(ctx, Event{Role: "reasoning", Content: "\n"})
- hadReasoning = true
- }
- reasoning.WriteString(ev.Reasoning)
- emit(ctx, Event{Role: "reasoning", Content: ev.Reasoning})
- }
- if ev.Content != "" {
- if hadReasoning {
- emit(ctx, Event{Role: "reasoning", Content: "\n\n"})
- hadReasoning = false
- }
- content.WriteString(ev.Content)
- emit(ctx, Event{Role: "assistant", Content: ev.Content, ResponseID: responseID})
- }
- toolCalls = append(toolCalls, ev.ToolCalls...)
- if ev.Done {
- if hadReasoning {
- emit(ctx, Event{Role: "reasoning", Content: "\n\n"})
- hadReasoning = false
- }
- stopReason = ev.StopReason
- done = true
- if ev.Usage.InputTokens > 0 || ev.Usage.OutputTokens > 0 {
- emit(ctx, 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 {
- // Backend didn't report usage; estimate from content.
- inChars := 0
- for _, m := range history {
- inChars += len(m.Content)
- for _, tc := range m.ToolCalls {
- inChars += len(tc.Name) + len(tc.Arguments)
- }
- }
- emit(ctx, Event{
- Role: "usage",
- Content: fmt.Sprintf("%d %d 1 0", inChars/4, content.Len()/4),
- })
- }
- break
- }
- }
-
- if done {
- break
- }
-
- if ctx.Err() != nil {
- // User pause: record partial state and return.
- if hadReasoning {
- emit(ctx, Event{Role: "reasoning", Content: "\n\n"})
- }
- msg := backend.Message{ID: responseID, Role: "assistant", Content: content.String(), Reasoning: reasoning.String(), ToolCalls: toolCalls}
- var results []toolResult
+ content, reasoning, toolCalls, stopReason, err := streamResponse(rt, ctx, history, responseID, step)
+ if err != nil {
+ // On context cancellation mid-stream, persist partial state.
+ if ctx.Err() != nil && (content != "" || len(toolCalls) > 0) {
+ msg := backend.Message{ID: responseID, Role: "assistant", Content: content, Reasoning: reasoning, ToolCalls: toolCalls}
+ var partialResults []toolResult
for _, tc := range toolCalls {
- results = append(results, toolResult{
- ToolCallID: tc.ID,
- Name: tc.Name,
- Content: `{"status":"cancelled","error":"stream interrupted"}`,
- IsError: true,
+ partialResults = append(partialResults, toolResult{
+ ToolCallID: tc.ID, Name: tc.Name,
+ Content: `{"status":"cancelled","error":"stream interrupted"}`, IsError: true,
})
}
- h.update(msg, results)
+ h.update(msg, partialResults)
ctx.Save()
- return ctx.Err()
- }
-
- // Pure stream drop — retry if attempts remain.
- if attempt >= maxTransientRetries {
- return fmt.Errorf("step %d: stream dropped (no more retries)", step)
- }
- if hadReasoning {
- emit(ctx, Event{Role: "reasoning", Content: "\n\n"})
- }
- wait := streamDropBaseDelay << attempt
- if err := retryCountdown(ctx, wait); err != nil {
- return fmt.Errorf("step %d: %w", step, err)
}
+ return err
}
switch stopReason {
case "stop", "tool_calls", "length", "error", "":
- // normal
default:
return fmt.Errorf("step %d: %s", step, stopReason)
}
- // Parse text-based tool calls: models that don't support the function
- // calling API emit tool invocations as plain text (e.g.
- // "file_read: args=[...]" (text-based tool call syntax)). Parse these into proper ToolCall structs.
+ // Parse text-based tool calls for models without function calling.
if len(toolCalls) == 0 && len(rt.Tools) > 0 {
- if parsed := parseTextToolCalls(content.String(), rt.Tools); len(parsed) > 0 {
+ if parsed := parseTextToolCalls(content, rt.Tools); len(parsed) > 0 {
toolCalls = parsed
}
}
- // Execute tool calls, running consecutive parallel-read-safe tools concurrently.
- msg := backend.Message{ID: responseID, Role: "assistant", Content: content.String(), Reasoning: reasoning.String(), ToolCalls: toolCalls}
- results := make([]toolResult, 0, len(toolCalls))
- interrupted := false
-
- cancelledResult := func(call backend.ToolCall) toolResult {
- return toolResult{
- ToolCallID: call.ID,
- Name: call.Name,
- Content: `{"status":"cancelled","error":"interrupted"}`,
- IsError: true,
- }
- }
-
- // execOne runs a single tool call end-to-end and reports whether the
- // context was cancelled during execution.
- execOne := func(call backend.ToolCall) (toolResult, bool) {
- if call.Name == "" {
- return toolResult{ToolCallID: call.ID, Name: call.Name, Content: "error: empty tool name", IsError: true}, false
- }
- if ctx.Err() != nil {
- cr := cancelledResult(call)
- emit(ctx, Event{Role: "tool", Name: call.Name, Content: cr.Content, OutputFormat: toolOutputFormat(rt, call.Name)})
- return cr, true
- }
- emit(ctx, Event{Role: "call", Name: call.Name, Content: string(call.Arguments)})
- readSafe := rt.ToolMeta[call.Name].ReadOnly
- if readSafe {
- key := call.Name + "\x00" + string(call.Arguments)
- if v, ok := resultCache.Load(key); ok {
- entry := v.(cachedResult)
- if cacheValid(entry, call.Arguments) {
- emit(ctx, 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
- }
- // Stale — remove and re-execute.
- resultCache.Delete(key)
- }
- }
- var result string
- var resultBlocks []backend.ContentBlock
- var isErr bool
- streamed := false
- if rt.Exec != nil {
- streamBytes := 0
- streamCtx := toolsrv.WithOutputStream(ctx, func(data string) {
- if streamBytes >= defaultToolResultMaxBytes {
- return // already at ceiling, drop further chunks
- }
- if !streamed {
- emit(ctx, Event{Role: "tool", Name: call.Name, OutputFormat: toolOutputFormat(rt, call.Name)})
- streamed = true
- }
- streamBytes += len(data)
- if streamBytes > defaultToolResultMaxBytes {
- // Emit only the portion within the ceiling.
- excess := streamBytes - defaultToolResultMaxBytes
- data = data[:len(data)-excess]
- }
- emit(ctx, Event{Role: "tool", Name: call.Name, Content: data, OutputFormat: toolOutputFormat(rt, call.Name)})
- })
- out, blocks, err := rt.Exec(streamCtx, call.Name, call.Arguments)
- if err != nil {
- isErr = true
- if ctx.Err() != nil {
- result = "error: tool execution interrupted by user"
- if ctx.PopInject != nil {
- if injected := ctx.PopInject(); injected != "" {
- result += "\n\n\n" + injected + "\n"
- }
- }
- emit(ctx, 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
- }
- // Accumulate suffix text (user-interruptions, truncation hints)
- // that must be emitted after streaming completes.
- var suffix string
- if ctx.PopInject != nil {
- if injected := ctx.PopInject(); injected != "" {
- suffix += "\n\n\n" + injected + "\n"
- }
- }
- result += suffix
- // Safety ceiling: unconditionally cap all tool results at 128KB.
- 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,
- })
- }
- if !streamed {
- emit(ctx, Event{Role: "tool", Name: call.Name, Content: result, OutputFormat: toolOutputFormat(rt, call.Name)})
- } else {
- // Emit suffixes that were appended after streaming completed
- // so they appear at the end of the chat output.
- if suffix != "" {
- emit(ctx, Event{Role: "tool", Name: call.Name, Content: suffix, OutputFormat: toolOutputFormat(rt, call.Name)})
- }
- // Signal the log handler to close the source fence and emit
- // [[[end]]]. Without this, streamingRole stays set and the
- // fence is only closed on the next role transition — or never,
- // if the agent turn ends or is interrupted.
- emit(ctx, Event{Role: "tool", Name: call.Name, 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
- }
-
- isParallelSafe := func(name string) bool {
- return name != "" && rt.ToolMeta[name].ReadOnly
- }
-
- for i := 0; i < len(toolCalls) && !interrupted; {
- if ctx.Err() != nil {
- for _, remaining := range toolCalls[i:] {
- cr := cancelledResult(remaining)
- emit(ctx, Event{Role: "tool", Name: remaining.Name, Content: cr.Content, OutputFormat: toolOutputFormat(rt, remaining.Name)})
- results = append(results, cr)
- }
- interrupted = true
- break
- }
-
- // Collect a run of consecutive parallel-safe calls.
- j := i + 1
- if isParallelSafe(toolCalls[i].Name) {
- for j < len(toolCalls) && isParallelSafe(toolCalls[j].Name) {
- j++
- }
- }
- batch := toolCalls[i:j]
-
- fillCancelled := func(from int) {
- for _, remaining := range toolCalls[from:] {
- cr := cancelledResult(remaining)
- emit(ctx, Event{Role: "tool", Name: remaining.Name, Content: cr.Content, OutputFormat: toolOutputFormat(rt, remaining.Name)})
- results = append(results, cr)
- }
- }
-
- if len(batch) == 1 {
- tr, wasInt := execOne(batch[0])
- results = append(results, tr)
- if wasInt {
- fillCancelled(j)
- interrupted = true
- }
- } else {
- // Fan out the batch concurrently; preserve submission order in results.
- // Deduplicate identical calls so only one executes per unique key.
- type inflightResult struct {
- tr toolResult
- wasInt bool
- }
- inflight := make(map[string]int) // key -> index of first occurrence
- batchResults := make([]toolResult, len(batch))
- batchInt := make([]bool, len(batch))
- var wg sync.WaitGroup
- uniqueResults := make([]inflightResult, len(batch))
- for k, call := range batch {
- key := call.Name + "\x00" + string(call.Arguments)
- if first, dup := inflight[key]; dup {
- // Will copy result from first occurrence after wg.Wait.
- batchResults[k] = toolResult{} // placeholder
- _ = first // used below
- continue
- }
- inflight[key] = k
- wg.Add(1)
- go func(k int, call backend.ToolCall) {
- defer wg.Done()
- uniqueResults[k].tr, uniqueResults[k].wasInt = execOne(call)
- }(k, call)
- }
- wg.Wait()
- // Fill results: unique calls get their own result, duplicates copy from first.
- for k, call := range batch {
- key := call.Name + "\x00" + string(call.Arguments)
- first := inflight[key]
- if k == first {
- batchResults[k] = uniqueResults[k].tr
- batchInt[k] = uniqueResults[k].wasInt
- } else {
- batchResults[k] = toolResult{
- ToolCallID: call.ID,
- Name: call.Name,
- Content: uniqueResults[first].tr.Content,
- IsError: uniqueResults[first].tr.IsError,
- }
- batchInt[k] = uniqueResults[first].wasInt
- emit(ctx, Event{Role: "call", Name: call.Name, Content: string(call.Arguments)})
- emit(ctx, Event{Role: "tool", Name: call.Name, Content: batchResults[k].Content, OutputFormat: toolOutputFormat(rt, call.Name)})
- }
- }
- // Add all batch results — model needs a result for every tool call.
- results = append(results, batchResults...)
- for _, wasInt := range batchInt {
- if wasInt {
- fillCancelled(j)
- interrupted = true
- break
- }
- }
- }
- i = j
- }
+ msg := backend.Message{ID: responseID, Role: "assistant", Content: content, Reasoning: reasoning, ToolCalls: toolCalls}
+ results, interrupted := execToolCalls(rt, ctx, toolCalls, resultCache)
h.update(msg, results)
ctx.Save()
@@ -522,77 +185,15 @@ func run(rt *Runtime, ctx TurnCtx, h *History) error {
if interrupted {
return ctx.Err()
}
-
if len(toolCalls) == 0 {
break
}
- // Track consecutive rounds where every tool call errored.
- // Rate-limited errors don't count — they're a temporary block,
- // not a real failure. Reset consecutiveErrors so the hard abort
- // doesn't fire.
- 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 {
- consecutiveErrors = 0
- } else if allErrors {
- consecutiveErrors++
- } else {
- consecutiveErrors = 0
- }
- if consecutiveErrors >= consecutiveErrorHardLimit {
- emit(ctx, Event{Role: "error", Content: fmt.Sprintf("%d consecutive tool errors — aborting", consecutiveErrors)})
- return fmt.Errorf("step %d: %d consecutive tool errors", step, consecutiveErrors)
- }
- if consecutiveErrors == consecutiveErrorSoftLimit {
- // Nudge the model to try a different approach and keep the plan current.
- 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)
+ if abort := trackErrors(h, ctx, results, &errs, step); abort != nil {
+ return abort
}
- // Repeated identical error detection: if the same error occurs 3 times
- // in a row, the model is stuck on a format/argument issue. Inject the
- // error content as a corrective nudge immediately rather than waiting
- // for the generic soft limit.
- 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 == lastErrorSig {
- repeatErrorCount++
- } else {
- repeatErrorCount = 1
- lastErrorSig = sig
- }
- if repeatErrorCount == 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 {
- repeatErrorCount = 0
- lastErrorSig = ""
- }
-
- // Reset step counter when an action-oriented tool is called successfully.
+ // 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 {
@@ -602,7 +203,7 @@ func run(rt *Runtime, ctx TurnCtx, h *History) error {
}
}
- // Step budget warnings: nudge the agent at 50% and when exhausted.
+ // Step budget warnings.
if rt.MaxSteps > 0 {
halfBudget := rt.MaxSteps / 2
if step == halfBudget {
@@ -630,6 +231,415 @@ func run(rt *Runtime, ctx TurnCtx, h *History) error {
return 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 streamResponse(rt *Runtime, ctx TurnCtx, history []backend.Message, responseID string, step int) (content string, reasoning string, toolCalls []backend.ToolCall, stopReason string, err error) {
+ 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) {
+ emit(ctx, Event{Role: "limitretry"})
+ }
+ if err := 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 {
+ emit(ctx, Event{Role: "reasoning", Content: "\n"})
+ hadReasoning = true
+ }
+ reasoningBuf.WriteString(ev.Reasoning)
+ emit(ctx, Event{Role: "reasoning", Content: ev.Reasoning})
+ }
+ if ev.Content != "" {
+ if hadReasoning {
+ emit(ctx, Event{Role: "reasoning", Content: "\n\n"})
+ hadReasoning = false
+ }
+ contentBuf.WriteString(ev.Content)
+ emit(ctx, Event{Role: "assistant", Content: ev.Content, ResponseID: responseID})
+ }
+ toolCalls = append(toolCalls, ev.ToolCalls...)
+ if ev.Done {
+ if hadReasoning {
+ emit(ctx, Event{Role: "reasoning", Content: "\n\n"})
+ hadReasoning = false
+ }
+ stopReason = ev.StopReason
+ done = true
+ if ev.Usage.InputTokens > 0 || ev.Usage.OutputTokens > 0 {
+ emit(ctx, 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)
+ }
+ }
+ emit(ctx, 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 {
+ emit(ctx, 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 {
+ emit(ctx, Event{Role: "reasoning", Content: "\n\n"})
+ }
+ wait := streamDropBaseDelay << attempt
+ if err := 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, running consecutive read-safe
+// calls in parallel. Returns the results and whether the loop was interrupted.
+func execToolCalls(rt *Runtime, ctx TurnCtx, toolCalls []backend.ToolCall, resultCache *sync.Map) ([]toolResult, bool) {
+ 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)
+ emit(ctx, Event{Role: "tool", Name: c.Name, Content: cr.Content, OutputFormat: toolOutputFormat(rt, c.Name)})
+ results = append(results, cr)
+ }
+ }
+
+ isParallelSafe := func(name string) bool {
+ return name != "" && rt.ToolMeta[name].ReadOnly
+ }
+
+ for i := 0; i < len(toolCalls); {
+ if ctx.Err() != nil {
+ fillCancelled(toolCalls[i:])
+ return results, true
+ }
+
+ // Collect a run of consecutive parallel-safe calls.
+ j := i + 1
+ if isParallelSafe(toolCalls[i].Name) {
+ for j < len(toolCalls) && isParallelSafe(toolCalls[j].Name) {
+ j++
+ }
+ }
+ batch := toolCalls[i:j]
+
+ if len(batch) == 1 {
+ tr, wasInt := execOne(rt, ctx, batch[0], resultCache)
+ results = append(results, tr)
+ if wasInt {
+ fillCancelled(toolCalls[j:])
+ return results, true
+ }
+ } else {
+ batchResults, wasInt := execBatch(rt, ctx, batch, resultCache)
+ results = append(results, batchResults...)
+ if wasInt {
+ fillCancelled(toolCalls[j:])
+ return results, true
+ }
+ }
+ i = j
+ }
+
+ return results, false
+}
+
+// execBatch fans out a batch of parallel-safe calls, deduplicating identical ones.
+func execBatch(rt *Runtime, ctx TurnCtx, batch []backend.ToolCall, resultCache *sync.Map) ([]toolResult, bool) {
+ 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 = execOne(rt, ctx, call, resultCache)
+ }(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,
+ }
+ emit(ctx, Event{Role: "call", Name: call.Name, Content: string(call.Arguments)})
+ emit(ctx, 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 execOne(rt *Runtime, ctx TurnCtx, call backend.ToolCall, resultCache *sync.Map) (toolResult, bool) {
+ 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}
+ emit(ctx, Event{Role: "tool", Name: call.Name, Content: cr.Content, OutputFormat: toolOutputFormat(rt, call.Name)})
+ return cr, true
+ }
+ emit(ctx, Event{Role: "call", Name: call.Name, Content: string(call.Arguments)})
+
+ readSafe := rt.ToolMeta[call.Name].ReadOnly
+ if readSafe {
+ key := call.Name + "\x00" + string(call.Arguments)
+ if v, ok := resultCache.Load(key); ok {
+ entry := v.(cachedResult)
+ if cacheValid(entry, call.Arguments) {
+ emit(ctx, 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
+ streamed := false
+
+ if rt.Exec != nil {
+ streamBytes := 0
+ streamCtx := toolsrv.WithOutputStream(ctx, func(data string) {
+ if streamBytes >= defaultToolResultMaxBytes {
+ return
+ }
+ if !streamed {
+ emit(ctx, Event{Role: "tool", Name: call.Name, OutputFormat: toolOutputFormat(rt, call.Name)})
+ streamed = true
+ }
+ streamBytes += len(data)
+ if streamBytes > defaultToolResultMaxBytes {
+ excess := streamBytes - defaultToolResultMaxBytes
+ data = data[:len(data)-excess]
+ }
+ emit(ctx, Event{Role: "tool", Name: call.Name, Content: data, OutputFormat: toolOutputFormat(rt, call.Name)})
+ })
+ out, blocks, err := rt.Exec(streamCtx, call.Name, call.Arguments)
+ if err != nil {
+ isErr = true
+ if ctx.Err() != nil {
+ result = "error: tool execution interrupted by user"
+ if ctx.PopInject != nil {
+ if injected := ctx.PopInject(); injected != "" {
+ result += "\n\n\n" + injected + "\n"
+ }
+ }
+ emit(ctx, 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 ctx.PopInject != nil {
+ if injected := ctx.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,
+ })
+ }
+
+ if !streamed {
+ emit(ctx, Event{Role: "tool", Name: call.Name, Content: result, OutputFormat: toolOutputFormat(rt, call.Name)})
+ } else {
+ if suffix != "" {
+ emit(ctx, Event{Role: "tool", Name: call.Name, Content: suffix, OutputFormat: toolOutputFormat(rt, call.Name)})
+ }
+ emit(ctx, Event{Role: "tool", Name: call.Name, 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
+}
+
+// ── trackErrors ─────────────────────────────────────────────────────────────
+
+// trackErrors updates consecutive/repeated error tracking and injects nudge
+// messages when thresholds are hit. Returns non-nil to abort the loop.
+func trackErrors(h *History, ctx TurnCtx, results []toolResult, es *errorState, step int) error {
+ 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 {
+ emit(ctx, 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 ─────────────────────────────────────────────────────────────────
+
func emit(ctx TurnCtx, msg Event) {
if ctx.Output != nil {
ctx.Output(msg)
@@ -652,8 +662,6 @@ func contextBudget(ctx context.Context, b backend.Backend) int {
// defaultToolResultMaxBytes caps tool result content sent back to the model.
const defaultToolResultMaxBytes = 131072
-
-
// MemoryTier classifies how long a tool result stays in the hot message list.
type MemoryTier int
@@ -665,14 +673,15 @@ const (
// toolResult holds the output of a single tool call.
type toolResult struct {
- ToolCallID string // may be empty (Ollama does not always set this)
+ ToolCallID string
Name string
Content string
- ContentBlocks []backend.ContentBlock // image blocks extracted from tool output
+ ContentBlocks []backend.ContentBlock
IsError bool
- Tier MemoryTier // default MemoryHot; tools may self-classify
+ Tier MemoryTier
}
-// retryable. Rate limits use longer waits; transient/network errors use shorter.
+
+// 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) {