refactor: extract sub-functions from agent/loop.go run()
run() was ~370 lines doing streaming, tool execution, error tracking, and budget management. Now ~70 lines delegating to: - streamResponse(): retry loop, event processing, usage tracking - execToolCalls()/execBatch()/execOne(): dispatch, caching, fan-out, dedup - trackErrors(): consecutive/repeated error tracking, nudge injection Added errorState struct for clean state threading.
This commit is contained in:
Ollie Agent
parent
74635b8e00
commit
61dbf760ac
879
agent/loop.go
879
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: "<think>\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</think>\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</think>\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</think>\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</think>\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<system-user-interruption>\n" + injected + "\n</system-user-interruption>"
|
||||
}
|
||||
}
|
||||
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<system-user-interruption>\n" + injected + "\n</system-user-interruption>"
|
||||
}
|
||||
}
|
||||
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: "<system-consecutive-errors>\nYour last several tool calls all failed. Try a different approach, or ask the user for help. Keep your plan current.\n</system-consecutive-errors>",
|
||||
}, 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: "<system-repeated-error>\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</system-repeated-error>",
|
||||
}, 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: "<think>\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</think>\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</think>\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</think>\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</think>\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<system-user-interruption>\n" + injected + "\n</system-user-interruption>"
|
||||
}
|
||||
}
|
||||
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<system-user-interruption>\n" + injected + "\n</system-user-interruption>"
|
||||
}
|
||||
}
|
||||
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: "<system-consecutive-errors>\nYour last several tool calls all failed. Try a different approach, or ask the user for help. Keep your plan current.\n</system-consecutive-errors>",
|
||||
}, 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: "<system-repeated-error>\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</system-repeated-error>",
|
||||
}, 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) {
|
||||
|
|
|
|||
Loading…
Reference in New Issue