lkn
/
ollie
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

refactor: eliminate TurnCtx, make loop functions Agent methods 

- Delete TurnCtx struct (was progressively populated with optional nil fields)
- Convert run(), streamResponse(), execToolCalls(), execBatch(), execOne(),
  trackErrors(), retryCountdown() to Agent methods
- Extract autoCompact() and popInject() as Agent methods
- turn.go now installs a turn-scoped output handler wrapper instead of
  building TurnCtx closures
- context.Context is now passed as plain context.Context, not smuggled
  with unrelated state

Add history compaction tests:
- TestBuildCompactedHistory_OrphanedToolWalkback: verifies walk-back logic
  prevents orphaned tool messages at hot zone boundary
- TestBuildCompactedHistory_NoOrphanAtBoundary: verifies tool messages
  always have preceding assistant with tool_calls
This commit is contained in:
Levi Neely 2026-08-12 18:06:19 +02:00
parent fe24736b19
commit 23f3a6092b
3 changed files with 226 additions and 118 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

126
cmd/olliesrv/internal/agent/history_test.go Normal file
View File

@ -0,0 +1,126 @@
package agent
import (
"testing"
"ollie/cmd/olliesrv/internal/backend"
)
func TestBuildCompactedHistory_OrphanedToolWalkback(t *testing.T) {
// Scenario: hotStart would land on a tool message. The walk-back logic
// should include the preceding assistant message with tool_calls to avoid
// violating the API constraint that tool messages must follow the assistant
// message containing the corresponding tool_calls.
// Build a history where the naive hotStart (total - hotTailSize) lands on a tool message.
// hotTailSize = 6, so we need enough messages that slicing at -6 hits a tool message.
// We want total - hotTailSize to land on a tool message.
// hotTailSize = 6, so with 12 messages, hotStart = 6.
// If messages[6] is a tool message, the walk-back should kick in.
messages := []backend.Message{
{Role: "user", Content: "old message 1"},
{Role: "assistant", Content: "old response 1"},
{Role: "user", Content: "old message 2"},
{Role: "assistant", Content: "old response 2"},
{Role: "user", Content: "request tool use"},
{Role: "assistant", Content: "", ToolCalls: []backend.ToolCall{{ID: "tc1", Name: "shell"}}}, // index 5
{Role: "tool", ToolCallID: "tc1", Content: "command output"}, // index 6 - naive hotStart lands here!
{Role: "user", Content: "next question"},
{Role: "assistant", Content: "answer"},
{Role: "user", Content: "another question"},
{Role: "assistant", Content: "another answer"},
{Role: "user", Content: "final question"},
}
// Total = 12, hotTailSize = 6, so naive hotStart = 6
// messages[6] is a tool message → should walk back to 5.
result := buildCompactedHistory("test summary", messages)
// Result structure:
// [0] cold zone (compaction summary)
// [1] warm zone (if any)
// [2+] hot zone
// The hot zone should start with the assistant message at index 5, not the tool at 6.
// Find the first non-summary message.
var hotStart int
for i, m := range result {
if m.Role == "assistant" && len(m.ToolCalls) > 0 {
hotStart = i
break
}
if m.Role == "tool" {
t.Errorf("hot zone starts with orphaned tool message at index %d", i)
return
}
}
// Verify the assistant with tool_calls is included before the tool message
if hotStart == 0 {
// The assistant message was not found, which is fine if the walk-back
// moved the boundary correctly. Let's just verify no orphaned tool.
for i, m := range result {
if m.Role == "tool" {
// Check that the previous message is assistant with matching tool_calls
if i == 0 {
t.Error("tool message at start of result with no preceding assistant")
return
}
prev := result[i-1]
if prev.Role != "assistant" {
t.Errorf("tool message at %d not preceded by assistant (got %s)", i, prev.Role)
}
}
}
}
}
func TestBuildCompactedHistory_NoOrphanAtBoundary(t *testing.T) {
// Simpler test: verify that any tool message in the result has
// a preceding assistant message with tool_calls.
messages := []backend.Message{
{Role: "user", Content: "m1"},
{Role: "assistant", Content: "r1"},
{Role: "user", Content: "m2"},
{Role: "assistant", Content: "", ToolCalls: []backend.ToolCall{{ID: "t1", Name: "shell"}, {ID: "t2", Name: "file_read"}}},
{Role: "tool", ToolCallID: "t1", Content: "output1"},
{Role: "tool", ToolCallID: "t2", Content: "output2"},
{Role: "user", Content: "m3"},
{Role: "assistant", Content: "r3"},
{Role: "user", Content: "m4"},
{Role: "assistant", Content: "r4"},
{Role: "user", Content: "m5"},
{Role: "assistant", Content: "r5"},
}
result := buildCompactedHistory("summary", messages)
// Scan for tool messages and verify they have proper context
for i, m := range result {
if m.Role == "tool" {
if i == 0 {
t.Fatal("tool message at index 0 with no preceding message")
}
// Walk backward to find assistant with tool_calls
found := false
for j := i - 1; j >= 0; j-- {
if result[j].Role == "assistant" && len(result[j].ToolCalls) > 0 {
found = true
break
}
if result[j].Role == "user" {
// Hit a user message without finding assistant+tool_calls
break
}
}
if !found {
t.Errorf("tool message at index %d has no preceding assistant with tool_calls", i)
}
}
}
}

176
cmd/olliesrv/internal/agent/loop.go
View File

@ -104,38 +104,23 @@ func toolMemoryTier(rt *Runtime, name string) MemoryTier {
return MemoryHot
}
// TurnCtx holds per-turn closures and state that vary between turns within
// the same agent. Everything else comes from *Runtime. Embeds context.Context
// so the turn's cancellation is carried with the turn state rather than as a
// separate parameter.
type TurnCtx struct {
context.Context
Output EventHandler
PopInject func() string
AutoCompact func(ctx context.Context)
Save func()
IncrToolCallCount func() int64
ResultCache *sync.Map
ToolServer *toolsrv.Conn
}
// errorState tracks consecutive and repeated tool errors across turns.
type errorState struct {
consecutive int
lastSig string
repeatCount int
consecutive int
lastSig string
repeatCount int
}
func run(rt *Runtime, ctx TurnCtx, h *History) error {
// run is the main agent loop. It streams responses from the LLM, executes tool
// calls, and updates history until the model stops or context is cancelled.
func (ag *Agent) run(ctx context.Context) error {
var step int
var errs errorState
resultCache := ctx.ResultCache
if resultCache == nil {
resultCache = &sync.Map{}
}
rt := ag.runtime
h := ag.history
for {
emit(ctx, Event{Role: "state", Content: "thinking"})
ag.emit(Event{Role: "state", Content: "thinking"})
if budget := contextBudget(ctx, rt.Backend); budget > 0 && h.estimateTokens() > budget {
h.stripCold(ctx, rt.Backend)
}
@ -146,7 +131,7 @@ func run(rt *Runtime, ctx TurnCtx, h *History) error {
}
responseID := paths.NewUUID()
content, reasoning, toolCalls, stopReason, err := streamResponse(rt, ctx, history, responseID, step)
content, reasoning, toolCalls, stopReason, err := ag.streamResponse(ctx, history, responseID, step)
if err != nil {
// On context cancellation mid-stream, persist partial state.
if ctx.Err() != nil && (content != "" || len(toolCalls) > 0) {
@ -159,7 +144,7 @@ func run(rt *Runtime, ctx TurnCtx, h *History) error {
})
}
h.update(msg, partialResults)
ctx.Save()
ag.save()
}
return err
}
@ -178,10 +163,10 @@ func run(rt *Runtime, ctx TurnCtx, h *History) error {
}
msg := backend.Message{ID: responseID, Role: "assistant", Content: content, Reasoning: reasoning, ToolCalls: toolCalls}
results, interrupted := execToolCalls(rt, ctx, toolCalls, resultCache)
results, interrupted := ag.execToolCalls(ctx, toolCalls)
h.update(msg, results)
ctx.Save()
ag.save()
if interrupted {
return ctx.Err()
@ -190,7 +175,7 @@ func run(rt *Runtime, ctx TurnCtx, h *History) error {
break
}
if abort := trackErrors(h, ctx, results, &errs, step); abort != nil {
if abort := ag.trackErrors(ctx, results, &errs, step); abort != nil {
return abort
}
@ -213,7 +198,7 @@ func run(rt *Runtime, ctx TurnCtx, h *History) error {
Content: fmt.Sprintf("<system-step-budget-warning>\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</system-step-budget-warning>", step, rt.MaxSteps),
}, nil)
} else if step >= rt.MaxSteps-1 {
emit(ctx, Event{Role: "maxsteps", Content: fmt.Sprintf("%d", step+1)})
ag.emit(Event{Role: "maxsteps", Content: fmt.Sprintf("%d", step+1)})
h.update(backend.Message{
Role: "user",
Content: fmt.Sprintf("<system-step-budget-stop>\nStep budget exhausted (%d/%d steps used). Stop calling tools. Summarize what you have done and what remains, then stop.\n</system-step-budget-stop>", step+1, rt.MaxSteps),
@ -222,9 +207,7 @@ func run(rt *Runtime, ctx TurnCtx, h *History) error {
}
}
if ctx.AutoCompact != nil {
ctx.AutoCompact(ctx)
}
ag.autoCompact(ctx)
step++
}
@ -232,12 +215,30 @@ func run(rt *Runtime, ctx TurnCtx, h *History) error {
return nil
}
// autoCompact triggers context compaction if above threshold.
func (ag *Agent) autoCompact(ctx context.Context) {
if ctx.Err() != nil || ag.history == nil {
return
}
limit := ag.autoCompactLimit(ctx)
if limit <= 0 || ag.history.estimateTokens() < limit {
return
}
ag.emit(Event{Role: "info", Content: "auto-compacting context...\n"})
ag.SetState("compacting")
if _, err := ag.runCompact(ctx, "auto"); err != nil {
panic(fmt.Sprintf("mid-turn auto-compact: %v", err))
}
ag.SetState("thinking")
}
// ── 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) {
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
@ -257,9 +258,9 @@ func streamResponse(rt *Runtime, ctx TurnCtx, history []backend.Message, respons
}
var rlErr *backend.RateLimitError
if errors.As(streamErr, &rlErr) {
emit(ctx, Event{Role: "limitretry"})
ag.emit(Event{Role: "limitretry"})
}
if err := retryCountdown(ctx, wait); err != nil {
if err := ag.retryCountdown(ctx, wait); err != nil {
return "", "", nil, "", fmt.Errorf("step %d: %w", step, err)
}
continue
@ -270,30 +271,30 @@ func streamResponse(rt *Runtime, ctx TurnCtx, history []backend.Message, respons
for ev := range ch {
if ev.Reasoning != "" {
if !hadReasoning {
emit(ctx, Event{Role: "reasoning", Content: "<think>\n"})
ag.emit(Event{Role: "reasoning", Content: "<think>\n"})
hadReasoning = true
}
reasoningBuf.WriteString(ev.Reasoning)
emit(ctx, Event{Role: "reasoning", Content: ev.Reasoning})
ag.emit(Event{Role: "reasoning", Content: ev.Reasoning})
}
if ev.Content != "" {
if hadReasoning {
emit(ctx, Event{Role: "reasoning", Content: "\n</think>\n"})
ag.emit(Event{Role: "reasoning", Content: "\n</think>\n"})
hadReasoning = false
}
contentBuf.WriteString(ev.Content)
emit(ctx, Event{Role: "assistant", Content: ev.Content, ResponseID: responseID})
ag.emit(Event{Role: "assistant", Content: ev.Content, ResponseID: responseID})
}
if ev.Done {
toolCalls = append(toolCalls, ev.ToolCalls...)
if hadReasoning {
emit(ctx, Event{Role: "reasoning", Content: "\n</think>\n"})
ag.emit(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{
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),
})
@ -305,7 +306,7 @@ func streamResponse(rt *Runtime, ctx TurnCtx, history []backend.Message, respons
inChars += len(tc.Name) + len(tc.Arguments)
}
}
emit(ctx, Event{
ag.emit(Event{
Role: "usage",
Content: fmt.Sprintf("%d %d 1 0", inChars/4, contentBuf.Len()/4),
})
@ -320,7 +321,7 @@ func streamResponse(rt *Runtime, ctx TurnCtx, history []backend.Message, respons
if ctx.Err() != nil {
if hadReasoning {
emit(ctx, Event{Role: "reasoning", Content: "\n</think>\n"})
ag.emit(Event{Role: "reasoning", Content: "\n</think>\n"})
}
// Return partial state for caller to persist.
return contentBuf.String(), reasoningBuf.String(), toolCalls, "", ctx.Err()
@ -331,10 +332,10 @@ func streamResponse(rt *Runtime, ctx TurnCtx, history []backend.Message, respons
return "", "", nil, "", fmt.Errorf("step %d: stream dropped (no more retries)", step)
}
if hadReasoning {
emit(ctx, Event{Role: "reasoning", Content: "\n</think>\n"})
ag.emit(Event{Role: "reasoning", Content: "\n</think>\n"})
}
wait := streamDropBaseDelay << attempt
if err := retryCountdown(ctx, wait); err != nil {
if err := ag.retryCountdown(ctx, wait); err != nil {
return "", "", nil, "", fmt.Errorf("step %d: %w", step, err)
}
}
@ -348,7 +349,8 @@ func streamResponse(rt *Runtime, ctx TurnCtx, history []backend.Message, respons
// 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 execToolCalls(rt *Runtime, ctx TurnCtx, toolCalls []backend.ToolCall, resultCache *sync.Map) ([]toolResult, bool) {
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 {
@ -363,7 +365,7 @@ func execToolCalls(rt *Runtime, ctx TurnCtx, toolCalls []backend.ToolCall, resul
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)})
ag.emit(Event{Role: "tool", Name: c.Name, Content: cr.Content, OutputFormat: toolOutputFormat(rt, c.Name)})
results = append(results, cr)
}
}
@ -394,14 +396,14 @@ func execToolCalls(rt *Runtime, ctx TurnCtx, toolCalls []backend.ToolCall, resul
}
if len(batch) == 1 {
tr, wasInt := execOne(rt, ctx, batch[0], resultCache)
tr, wasInt := ag.execOne(ctx, batch[0])
results = append(results, tr)
if wasInt {
fillCancelled(toolCalls[j:])
return results, true
}
} else {
batchResults, wasInt := execBatch(rt, ctx, batch, resultCache)
batchResults, wasInt := ag.execBatch(ctx, batch)
results = append(results, batchResults...)
if wasInt {
fillCancelled(toolCalls[j:])
@ -475,7 +477,8 @@ func hasGlobal(set []string) bool {
}
// 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) {
func (ag *Agent) execBatch(ctx context.Context, batch []backend.ToolCall) ([]toolResult, bool) {
rt := ag.runtime
type inflightResult struct {
tr toolResult
wasInt bool
@ -494,7 +497,7 @@ func execBatch(rt *Runtime, ctx TurnCtx, batch []backend.ToolCall, resultCache *
wg.Add(1)
go func(k int, call backend.ToolCall) {
defer wg.Done()
uniqueResults[k].tr, uniqueResults[k].wasInt = execOne(rt, ctx, call, resultCache)
uniqueResults[k].tr, uniqueResults[k].wasInt = ag.execOne(ctx, call)
}(k, call)
}
wg.Wait()
@ -515,24 +518,27 @@ func execBatch(rt *Runtime, ctx TurnCtx, batch []backend.ToolCall, resultCache *
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)})
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 execOne(rt *Runtime, ctx TurnCtx, call backend.ToolCall, resultCache *sync.Map) (toolResult, bool) {
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}
emit(ctx, Event{Role: "tool", Name: call.Name, Content: cr.Content, OutputFormat: toolOutputFormat(rt, call.Name)})
ag.emit(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)})
ag.emit(Event{Role: "call", Name: call.Name, Content: string(call.Arguments)})
readSafe := rt.ToolMeta[call.Name].Scope == "read"
if readSafe {
@ -540,7 +546,7 @@ func execOne(rt *Runtime, ctx TurnCtx, call backend.ToolCall, resultCache *sync.
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)})
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)
@ -551,16 +557,18 @@ func execOne(rt *Runtime, ctx TurnCtx, call backend.ToolCall, resultCache *sync.
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 ctx.ToolServer == nil {
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 := ctx.ToolServer.CallToolBackground(call.Name, strippedArgs)
pid, err := toolServer.CallToolBackground(call.Name, strippedArgs)
if err != nil {
result = fmt.Sprintf("error: background exec: %v", err)
isErr = true
@ -568,7 +576,7 @@ func execOne(rt *Runtime, ctx TurnCtx, call backend.ToolCall, resultCache *sync.
result = fmt.Sprintf("<system-proc-background>\nid=%d cmd=%q\n</system-proc-background>", pid, cmdDesc)
}
}
emit(ctx, Event{Role: "tool", Name: call.Name, Content: result, OutputFormat: toolOutputFormat(rt, call.Name)})
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
}
@ -584,16 +592,16 @@ func execOne(rt *Runtime, ctx TurnCtx, call backend.ToolCall, resultCache *sync.
} else if args.Name == "" {
result = "error: name is required"
isErr = true
} else if rt.ToolServer == nil {
} else if toolServer == nil {
result = "error: no tool server"
isErr = true
} else if err := rt.ToolServer.LoadTool(args.Name); err != nil {
} else if err := toolServer.LoadTool(args.Name); err != nil {
result = "error: " + err.Error()
isErr = true
} else {
result = "loaded: " + args.Name
}
emit(ctx, Event{Role: "tool", Name: call.Name, Content: result, OutputFormat: toolOutputFormat(rt, call.Name)})
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
}
@ -603,12 +611,10 @@ func execOne(rt *Runtime, ctx TurnCtx, call backend.ToolCall, resultCache *sync.
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>"
}
if injected := ag.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)})
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
@ -628,10 +634,8 @@ func execOne(rt *Runtime, ctx TurnCtx, call backend.ToolCall, resultCache *sync.
// 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>"
}
if injected := ag.popInject(); injected != "" {
suffix += "\n\n<system-user-interruption>\n" + injected + "\n</system-user-interruption>"
}
result += suffix
@ -653,7 +657,7 @@ func execOne(rt *Runtime, ctx TurnCtx, call backend.ToolCall, resultCache *sync.
})
}
emit(ctx, Event{Role: "tool", Name: call.Name, Content: result, OutputFormat: toolOutputFormat(rt, call.Name)})
ag.emit(Event{Role: "tool", Name: call.Name, Content: result, OutputFormat: toolOutputFormat(rt, call.Name)})
tier := MemoryHot
if !isErr {
@ -662,11 +666,21 @@ func execOne(rt *Runtime, ctx TurnCtx, call backend.ToolCall, resultCache *sync.
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 trackErrors(h *History, ctx TurnCtx, results []toolResult, es *errorState, step int) error {
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 {
@ -689,7 +703,7 @@ func trackErrors(h *History, ctx TurnCtx, results []toolResult, es *errorState,
}
if es.consecutive >= consecutiveErrorHardLimit {
emit(ctx, Event{Role: "error", Content: fmt.Sprintf("%d consecutive tool errors — aborting", es.consecutive)})
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 {
@ -731,12 +745,6 @@ func trackErrors(h *History, ctx TurnCtx, results []toolResult, es *errorState,
// ── helpers ─────────────────────────────────────────────────────────────────
func emit(ctx TurnCtx, msg Event) {
if ctx.Output != nil {
ctx.Output(msg)
}
}
// 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 {
@ -799,7 +807,7 @@ func transientWait(err error, attempt int) (time.Duration, bool) {
return 0, false
}
func retryCountdown(ctx TurnCtx, wait time.Duration) error {
func (ag *Agent) retryCountdown(ctx context.Context, wait time.Duration) error {
deadline := time.Now().Add(wait)
for {
remaining := time.Until(deadline)
@ -810,7 +818,7 @@ func retryCountdown(ctx TurnCtx, wait time.Duration) error {
h := totalSecs / 3600
m := (totalSecs % 3600) / 60
s := totalSecs % 60
emit(ctx, Event{Role: "retry", Content: fmt.Sprintf("%02d:%02d:%02d", h, m, s)})
ag.emit(Event{Role: "retry", Content: fmt.Sprintf("%02d:%02d:%02d", h, m, s)})
select {
case <-ctx.Done():
return ctx.Err()

42
cmd/olliesrv/internal/agent/turn.go
View File

@ -117,12 +117,11 @@ func (ag *Agent) executeTurn(ctx context.Context, input string) string {
ag.runtime.Preamble.Set(SectionTools, renderTools(ti))
}
turnCtx := TurnCtx{
ToolServer: ag.runtime.ToolServer,
}
// Install turn-scoped output handler to intercept events for reply
// accumulation, state updates, and usage tracking.
var replyBuf strings.Builder
turnCtx.Output = func(ev Event) {
origOutput := ag.output
ag.output = func(ev Event) {
switch ev.Role {
case "assistant":
replyBuf.WriteString(ev.Content)
@ -151,34 +150,11 @@ func (ag *Agent) executeTurn(ctx context.Context, input string) string {
}, est != 0)
ag.notifyChange()
}
ag.emit(ev)
}
turnCtx.PopInject = func() string {
if p := ag.pendingInject.Swap(nil); p != nil {
return *p
if origOutput != nil {
origOutput(ev)
}
return ""
}
turnCtx.IncrToolCallCount = func() int64 {
return ag.toolCallCount.Add(1)
}
turnCtx.Save = ag.save
turnCtx.ResultCache = &ag.resultCache
turnCtx.AutoCompact = func(ctx context.Context) {
if ctx.Err() != nil || ag.history == nil {
return
}
limit := ag.autoCompactLimit(ctx)
if limit <= 0 || ag.history.estimateTokens() < limit {
return
}
ag.emit(Event{Role: "info", Content: "auto-compacting context...\n"})
ag.SetState("compacting")
if _, err := ag.runCompact(ctx, "auto"); err != nil {
panic(fmt.Sprintf("mid-turn auto-compact: %v", err))
}
ag.SetState("thinking")
}
defer func() { ag.output = origOutput }()
// Warn once when context usage crosses 60%; compact at 75%.
if ag.history != nil {
@ -211,8 +187,7 @@ func (ag *Agent) executeTurn(ctx context.Context, input string) string {
err error
)
for {
turnCtx.Context = actCtx
err = run(ag.runtime, turnCtx, ag.history)
err = ag.run(actCtx)
actCancel(nil)
ag.currentAction.CompareAndSwap(handle, nil)
@ -358,4 +333,3 @@ const defaultContextLength = 128000
func infoEvent(text string) Event {
return Event{Role: "info", Content: text + "\n"}
}