diff --git a/cmd/olliesrv/internal/agent/history_test.go b/cmd/olliesrv/internal/agent/history_test.go
new file mode 100644
index 0000000..26a0333
--- /dev/null
+++ b/cmd/olliesrv/internal/agent/history_test.go
@@ -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)
+ }
+ }
+ }
+}
diff --git a/cmd/olliesrv/internal/agent/loop.go b/cmd/olliesrv/internal/agent/loop.go
index 262a4b3..f128daa 100644
--- a/cmd/olliesrv/internal/agent/loop.go
+++ b/cmd/olliesrv/internal/agent/loop.go
@@ -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("\nYou have used %d/%d research steps without taking action. Consider making progress — write code, edit files, or run commands. Action tools reset this counter.\n", step, rt.MaxSteps),
}, nil)
} else if step >= rt.MaxSteps-1 {
- 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("\nStep budget exhausted (%d/%d steps used). Stop calling tools. Summarize what you have done and what remains, then stop.\n", step+1, rt.MaxSteps),
@@ -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: "\n"})
+ ag.emit(Event{Role: "reasoning", Content: "\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\n"})
+ ag.emit(Event{Role: "reasoning", Content: "\n\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\n"})
+ ag.emit(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{
+ 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\n"})
+ ag.emit(Event{Role: "reasoning", Content: "\n\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\n"})
+ ag.emit(Event{Role: "reasoning", Content: "\n\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("\nid=%d cmd=%q\n", 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\n" + injected + "\n"
- }
+ if injected := ag.popInject(); injected != "" {
+ result += "\n\n\n" + injected + "\n"
}
- 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\n" + injected + "\n"
- }
+ if injected := ag.popInject(); injected != "" {
+ suffix += "\n\n\n" + injected + "\n"
}
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()
diff --git a/cmd/olliesrv/internal/agent/turn.go b/cmd/olliesrv/internal/agent/turn.go
index 59cbfed..826dc19 100644
--- a/cmd/olliesrv/internal/agent/turn.go
+++ b/cmd/olliesrv/internal/agent/turn.go
@@ -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"}
}
-