1package core
2
3import (
4 "context"
5 "encoding/json"
6 "errors"
7 "fmt"
8 "io"
9 "strings"
10 "sync"
11 "time"
12 "unicode/utf8"
13
14 "github.com/bnema/zut/packages/provider"
15 "github.com/google/uuid"
16)
17
18var (
19 ErrStreamIdleTimeout = errors.New("provider stream idle timeout")
20 ErrMaxSteps = errors.New("agent model step limit exceeded")
21)
22
23const (
24 DefaultStreamIdleTimeout = 5 * time.Minute
25 DefaultStreamMaxRetries = 5
26)
27
28// QueuedMessage is a user prompt waiting for the next safe model-call boundary.
29type QueuedMessage struct {
30 Text string
31 Images []provider.ImageBlock
32 // HostEvent distinguishes host evidence from ordinary input so hosts can
33 // retain it after a failed turn. Explicit cancellation may still discard it.
34 HostEvent bool
35}
36
37type queuedMessage struct {
38 message QueuedMessage
39 accepted time.Time
40}
41
42// requestOpenError marks a failure returned before a provider stream exists.
43// Provider clients own retries for this phase; the agent retry budget is only
44// for reconnecting a stream that opened and then failed.
45type requestOpenError struct{ err error }
46
47func (e *requestOpenError) Error() string { return e.err.Error() }
48func (e *requestOpenError) Unwrap() error { return e.err }
49
50// Agent is a stateful conversation bound to a provider client, a model,
51// and a set of tools.
52type Agent struct {
53 Client provider.Client
54 Model string
55 // System is mutex-protected; use SetSystemPrompt to write it and
56 // PromptConfig to read it.
57 System string
58 // Tools is mutex-protected; use SetTools to write it and PromptConfig
59 // or ToolsSnapshot to read it.
60 Tools Registry
61 MaxSteps int
62 Reasoning string
63 Temperature *float32
64 // FastMode requests the OpenAI fast service tier. The provider boundary
65 // rejects it for providers that do not support that contract.
66 FastMode bool
67
68 // ContextWindow is the effective input context capacity retained from
69 // model resolution. Hosts use it for proactive compaction even when the
70 // model was synthesized for an open-catalog provider.
71 ContextWindow int
72
73 // MaxTokens caps the model's output tokens per turn. Zero leaves
74 // the field unset on the provider request, letting each provider
75 // apply its own default (which can be conservative, e.g. Bedrock
76 // defaults to 4096, truncating long writes/edits). Hosts populate
77 // this from the resolved model's MaxOutput so large single-turn
78 // responses aren't silently cut off with stopReason=length.
79 MaxTokens int
80
81 // BeforeToolExecute, if set, is called immediately before each
82 // tool runs. Returning (allowed=false, reason) short-circuits
83 // the call with an error result containing reason. Optionally,
84 // returning a non-nil modifiedArgs replaces the JSON args the
85 // tool will see, which lets guards redact / augment / patch the
86 // model's request without rewriting the transcript. Empty or
87 // malformed modifiedArgs is ignored.
88 BeforeToolExecute func(call provider.ToolCallBlock) (allowed bool, reason string, modifiedArgs json.RawMessage)
89
90 // BeforeTurn, if set, is called before each turn's model call.
91 // Returning (allowed=false, reason) aborts the turn; reason is
92 // surfaced as an assistant-like status line. Used for rate-
93 // limiting, business-hour gates, and deny-by-default setups.
94 BeforeTurn func(step int) (allowed bool, reason string)
95
96 // BeforeTurnContext is the richer turn-start hook. It receives the
97 // current turn context so blocking interception can honor cancellation.
98 // It may return a bounded, non-transcript context string that is appended
99 // to the provider request's system prompt for this turn only. When set, it
100 // replaces BeforeTurn so a host can combine approval and context
101 // collection into one lifecycle round trip.
102 BeforeTurnContext func(ctx context.Context, step int) (allowed bool, reason, context string)
103
104 // CommitToolResult, if set, commits extension/tool metadata immediately
105 // after each tool call completes. Returning an error converts the result to
106 // an error before it is appended to the transcript or sent to the provider.
107 CommitToolResult func(id string, result ToolResult) error
108 // OnToolResult observes the final result after CommitToolResult succeeds or
109 // converts it to an error.
110 OnToolResult func(id string, result ToolResult)
111
112 // BeforeAssistantMessage, if set, is called after the model's
113 // final assistant message is assembled but before it's appended
114 // to the transcript. Returning (allowed=false) suppresses both
115 // the transcript append and the UI event. A non-empty
116 // replacement rewrites the visible text for the user while
117 // leaving the model's original text in the transcript (so the
118 // model can still see what it said in subsequent turns).
119 BeforeAssistantMessage func(text string) (allowed bool, reason, replacement string)
120
121 // MaxRetries controls reconnection attempts for transient failures after the
122 // provider stream opens. Zero disables this retry layer.
123 // RetryBaseDelay is doubled for each attempt; zero uses 2s.
124 MaxRetries int
125 RetryBaseDelay time.Duration
126 // StreamIdleTimeout cancels an individual provider stream when it emits no
127 // events for this duration. Zero leaves the stream unbounded.
128 StreamIdleTimeout time.Duration
129 streamIdleTimer func(time.Duration) (<-chan time.Time, func(), func())
130
131 // OnEvent, if set, mirrors every AgentEvent the loop emits to
132 // this callback in addition to the per-Prompt sink. Used by the
133 // extension manager to fan events out to subscribed extensions
134 // without each caller having to compose sinks manually.
135 OnEvent func(AgentEvent)
136
137 // OnMessageAppended, if set, fires every time a message is
138 // appended to the in-memory transcript by the agent loop — the
139 // initial user prompt, each finalised assistant message, and
140 // each tool-results message (plus the synthetic OpenAI image
141 // mirror, if any). Hosts wire this to the on-disk session so
142 // that turns are durable as soon as they happen, instead of
143 // only being flushed on a clean exit.
144 OnMessageAppended func(provider.Message)
145
146 // OnUsage, if set, fires after every turn's usage row arrives,
147 // carrying the cumulative usage for the session. Hosts wire
148 // this to the on-disk session so the persisted total stays
149 // current and a crash recovers the right cost figure.
150 OnUsage func(cumulative provider.Usage)
151
152 // OnRetryLifecycle receives sanitized retry audit records. It never
153 // receives provider error text or response bodies.
154 OnRetryLifecycle func(RetryLifecycleRecord)
155
156 // OnTranscriptCompacted, if set, fires after Compact replaces the
157 // in-memory transcript with the synthetic summary plus kept tail.
158 // Hosts wire this to append an explicit compaction checkpoint to
159 // the session log; per-message append hooks do not fire for this
160 // wholesale transcript replacement.
161 OnTranscriptCompacted func(messages []provider.Message)
162
163 mu sync.Mutex
164 messages []provider.Message
165 // rev increments whenever the transcript slice is replaced or a
166 // message is appended. The TUI uses it as a cheap redraw cache key
167 // so editor-only typing doesn't copy/rebuild a long transcript on
168 // every keypress.
169 rev uint64
170 cost CostTracker
171
172 // queued holds user messages submitted while the agent is busy.
173 // The loop appends them as normal user messages at safe
174 // boundaries: before the next model call after a tool batch, or
175 // after a text-only assistant turn finishes. It never interrupts
176 // a running tool or cancels an in-flight provider request.
177 queued []queuedMessage
178 // deniedToolCall tracks whether any tool invocation was refused at the
179 // execution boundary (guard or confirmation denial) during the current
180 // top-level runLoop invocation. A denied invocation must not gain a
181 // recovery prompt urging more actions.
182 deniedToolCall bool
183 timeContext agentTimeContext
184 hasSessionTimeContext bool
185 // sessionID identifies this conversation thread. cacheSessionID identifies
186 // the root prompt-cache affinity and may be shared by resident child
187 // threads. Hosts bind both before the first provider request; unbound
188 // agents allocate one shared ephemeral value for the process lifetime.
189 sessionID string
190 cacheSessionID string
191 providerRequestStarted bool
192}
193
194// NewAgent returns an Agent with sensible defaults.
195func NewAgent(client provider.Client, model, system string, tools Registry) *Agent {
196 return &Agent{
197 Client: client,
198 Model: model,
199 System: system,
200 Tools: tools,
201 MaxSteps: 0, // 0 = unlimited
202 MaxRetries: DefaultStreamMaxRetries,
203 RetryBaseDelay: 2 * time.Second,
204 StreamIdleTimeout: DefaultStreamIdleTimeout,
205 timeContext: newAgentTimeContext(time.Now()),
206 }
207}
208
209// BindSessionID binds a root conversation's durable session as both its
210// thread identity and prompt-cache affinity. Resident children use
211// BindRequestIdentity to retain their own thread while sharing their root's
212// cache affinity.
213func (a *Agent) BindSessionID(id string) error {
214 return a.BindRequestIdentity(id, id)
215}
216
217// BindRequestIdentity assigns the stable root cache and thread identities
218// before the first provider request.
219func (a *Agent) BindRequestIdentity(cacheSessionID, threadID string) error {
220 cacheSessionID = strings.TrimSpace(cacheSessionID)
221 threadID = strings.TrimSpace(threadID)
222 if cacheSessionID == "" || threadID == "" {
223 return errors.New("cache session ID and thread ID are required")
224 }
225 a.mu.Lock()
226 defer a.mu.Unlock()
227 if a.providerRequestStarted {
228 return errors.New("cannot change request identity after the first provider request")
229 }
230 a.cacheSessionID = cacheSessionID
231 a.sessionID = threadID
232 return nil
233}
234
235// SessionID returns the stable conversation-thread identity, allocating an
236// ephemeral identity when persistence is disabled and the host did not bind
237// one during runtime assembly.
238func (a *Agent) SessionID() string {
239 a.mu.Lock()
240 defer a.mu.Unlock()
241 a.ensureRequestIdentityLocked()
242 return a.sessionID
243}
244
245// CacheSessionID returns the stable root cache-affinity identity.
246func (a *Agent) CacheSessionID() string {
247 a.mu.Lock()
248 defer a.mu.Unlock()
249 a.ensureRequestIdentityLocked()
250 return a.cacheSessionID
251}
252
253func (a *Agent) ensureRequestIdentityLocked() {
254 if a.sessionID == "" {
255 a.sessionID = uuid.NewString()
256 }
257 if a.cacheSessionID == "" {
258 a.cacheSessionID = a.sessionID
259 }
260}
261
262func (a *Agent) beginTurn() provider.RequestContext {
263 a.mu.Lock()
264 a.ensureRequestIdentityLocked()
265 context := provider.RequestContext{CacheSessionID: a.cacheSessionID, ThreadID: a.sessionID, TurnID: uuid.NewString()}
266 a.mu.Unlock()
267 return context
268}
269
270func (a *Agent) beginProviderRequest() {
271 a.mu.Lock()
272 a.providerRequestStarted = true
273 a.mu.Unlock()
274}
275
276// SetFastMode changes the service-tier preference for the next model call.
277// It is safe to use while a turn is in flight; the active turn keeps the
278// value it already captured.
279func (a *Agent) SetFastMode(enabled bool) {
280 a.mu.Lock()
281 a.FastMode = enabled
282 a.mu.Unlock()
283}
284
285// FastModeEnabled returns the service-tier preference for a model call.
286func (a *Agent) FastModeEnabled() bool {
287 a.mu.Lock()
288 defer a.mu.Unlock()
289 return a.FastMode
290}
291
292// QueueMessage queues a text and image prompt to be injected as a user message
293// at the next safe boundary of the active agent loop. It is non-blocking in the
294// sense that it never waits for model/tool work; it only takes the transcript
295// mutex briefly. Prompts without text or images are ignored.
296func (a *Agent) QueueMessage(text string, images []provider.ImageBlock) bool {
297 return a.QueuePrompt(QueuedMessage{Text: text, Images: images})
298}
299
300// QueuePrompt queues a prompt with its host provenance at the next safe boundary.
301func (a *Agent) QueuePrompt(prompt QueuedMessage) bool {
302 prompt.Text = strings.TrimSpace(prompt.Text)
303 if prompt.Text == "" && len(prompt.Images) == 0 {
304 return false
305 }
306 message := cloneQueuedMessage(prompt)
307 a.mu.Lock()
308 a.queued = append(a.queued, queuedMessage{message: message, accepted: time.Now()})
309 a.mu.Unlock()
310 return true
311}
312
313// PendingQueuedMessages returns a snapshot of user messages waiting
314// to be injected. Used by hosts to render the visible "sliding in"
315// chips without consuming them.
316func (a *Agent) PendingQueuedMessages() []QueuedMessage {
317 a.mu.Lock()
318 defer a.mu.Unlock()
319 out := make([]QueuedMessage, len(a.queued))
320 for i, queued := range a.queued {
321 out[i] = cloneQueuedMessage(queued.message)
322 }
323 return out
324}
325
326// QueuedMessageCount returns the number of messages waiting to be
327// injected at the next safe boundary.
328func (a *Agent) QueuedMessageCount() int {
329 a.mu.Lock()
330 defer a.mu.Unlock()
331 return len(a.queued)
332}
333
334// PopQueuedMessage removes and returns the most recently queued
335// message. Hosts use this for the slide-back keybinding.
336func (a *Agent) PopQueuedMessage() (QueuedMessage, bool) {
337 a.mu.Lock()
338 defer a.mu.Unlock()
339 n := len(a.queued)
340 if n == 0 {
341 return QueuedMessage{}, false
342 }
343 message := cloneQueuedMessage(a.queued[n-1].message)
344 a.queued = a.queued[:n-1]
345 return message, true
346}
347
348// DrainQueuedMessages discards and returns every queued message.
349// Hosts use this on explicit cancel/clear so stale follow-ups do
350// not run after the user aborted the turn.
351func (a *Agent) DrainQueuedMessages() []QueuedMessage {
352 queued := a.drainQueuedMessages()
353 out := make([]QueuedMessage, len(queued))
354 for i, message := range queued {
355 out[i] = cloneQueuedMessage(message.message)
356 }
357 return out
358}
359
360func cloneQueuedMessage(message QueuedMessage) QueuedMessage {
361 clone := QueuedMessage{Text: message.Text, HostEvent: message.HostEvent}
362 if len(message.Images) == 0 {
363 return clone
364 }
365 clone.Images = make([]provider.ImageBlock, len(message.Images))
366 for i, image := range message.Images {
367 clone.Images[i] = image
368 clone.Images[i].Data = append([]byte(nil), image.Data...)
369 }
370 return clone
371}
372
373func (a *Agent) drainQueuedMessages() []queuedMessage {
374 a.mu.Lock()
375 defer a.mu.Unlock()
376 out := append([]queuedMessage(nil), a.queued...)
377 a.queued = nil
378 return out
379}
380
381func (a *Agent) appendQueuedAsUser(messages []queuedMessage, sink func(AgentEvent)) {
382 for _, queued := range messages {
383 accepted := queued.accepted
384 if accepted.IsZero() {
385 accepted = time.Now()
386 }
387 content := make([]provider.Content, 0, 1+len(queued.message.Images))
388 if queued.message.Text != "" {
389 content = append(content, provider.TextBlock{Text: queued.message.Text})
390 }
391 for _, image := range queued.message.Images {
392 content = append(content, image)
393 }
394 msg := provider.Message{
395 Role: provider.RoleUser,
396 Content: content,
397 Time: accepted,
398 }
399 a.mu.Lock()
400 a.messages = append(a.messages, msg)
401 a.rev++
402 a.mu.Unlock()
403 a.fireMessageAppended(msg)
404 if sink != nil {
405 sink(EvUserMessage{Message: msg})
406 }
407 }
408}
409
410// Messages returns a copy of the current transcript.
411func (a *Agent) Messages() []provider.Message {
412 a.mu.Lock()
413 defer a.mu.Unlock()
414 out := make([]provider.Message, len(a.messages))
415 copy(out, a.messages)
416 return out
417}
418
419// MessageCount returns the current transcript length without copying it.
420func (a *Agent) MessageCount() int {
421 a.mu.Lock()
422 defer a.mu.Unlock()
423 return len(a.messages)
424}
425
426// MessagesFrom returns a copy of the transcript suffix beginning at from and
427// the total transcript length observed under the same lock.
428func (a *Agent) MessagesFrom(from int) ([]provider.Message, int) {
429 a.mu.Lock()
430 defer a.mu.Unlock()
431 if from < 0 {
432 from = 0
433 }
434 if from > len(a.messages) {
435 from = len(a.messages)
436 }
437 out := make([]provider.Message, len(a.messages)-from)
438 copy(out, a.messages[from:])
439 return out, len(a.messages)
440}
441
442// Revision returns a monotonically increasing transcript version.
443// It is cheap to query and changes whenever Messages() would return
444// different transcript content because of append/set operations.
445func (a *Agent) Revision() uint64 {
446 a.mu.Lock()
447 defer a.mu.Unlock()
448 return a.rev
449}
450
451// SetTools swaps the tool registry. Used by /reload-ext to hand
452// the agent a fresh registry after extension subprocesses have been
453// respawned (and their freshly-registered tools merged in).
454func (a *Agent) SetTools(reg Registry) {
455 a.mu.Lock()
456 a.Tools = reg
457 a.mu.Unlock()
458}
459
460// SetSystemPrompt replaces the system prompt used for subsequent turns.
461func (a *Agent) SetSystemPrompt(system string) {
462 a.mu.Lock()
463 a.System = system
464 a.mu.Unlock()
465}
466
467// SetPromptConfig atomically replaces the system prompt and tool registry.
468// It returns the previous registry so the caller can close resources that are
469// no longer owned by the agent.
470func (a *Agent) SetPromptConfig(system string, tools Registry) Registry {
471 a.mu.Lock()
472 defer a.mu.Unlock()
473 oldTools := a.Tools
474 a.System = system
475 a.Tools = tools
476 return oldTools
477}
478
479// PromptConfig returns a consistent snapshot of the system prompt and tools
480// used to construct a provider request.
481func (a *Agent) PromptConfig() (string, Registry) {
482 a.mu.Lock()
483 defer a.mu.Unlock()
484 tools := make(Registry, len(a.Tools))
485 for name, tool := range a.Tools {
486 tools[name] = tool
487 }
488 return a.System, tools
489}
490
491// ToolsSnapshot returns a copy of the current tool registry.
492func (a *Agent) ToolsSnapshot() Registry {
493 a.mu.Lock()
494 defer a.mu.Unlock()
495 tools := make(Registry, len(a.Tools))
496 for name, tool := range a.Tools {
497 tools[name] = tool
498 }
499 return tools
500}
501
502// SetMessages replaces the transcript (used when resuming a session).
503func (a *Agent) SetMessages(msgs []provider.Message) {
504 a.mu.Lock()
505 defer a.mu.Unlock()
506 a.messages = append(a.messages[:0], msgs...)
507 a.rev++
508}
509
510// AppendUserContext adds a user-role message to the transcript without
511// starting a model turn. Hosts use it for context gathered outside the agent
512// loop, such as the output of an explicitly invoked shell command.
513func (a *Agent) AppendUserContext(text string, meta map[string]string) {
514 msg := provider.Message{
515 Role: provider.RoleUser,
516 Content: []provider.Content{provider.TextBlock{Text: text}},
517 Time: time.Now(),
518 Meta: meta,
519 }
520 a.mu.Lock()
521 a.messages = append(a.messages, msg)
522 a.rev++
523 a.mu.Unlock()
524 a.fireMessageAppended(msg)
525}
526
527// Cost returns the cumulative usage.
528func (a *Agent) Cost() provider.Usage {
529 a.mu.Lock()
530 defer a.mu.Unlock()
531 return a.cost.Total
532}
533
534func (a *Agent) addUsage(usage provider.Usage) provider.Usage {
535 a.mu.Lock()
536 defer a.mu.Unlock()
537 return a.cost.Add(usage)
538}
539
540// SeedCost sets the cumulative usage as a baseline before the first
541// turn runs. Used when transferring state from another agent (model
542// or provider switch) so the running cost meter doesn't reset to 0.
543func (a *Agent) SeedCost(u provider.Usage) {
544 a.mu.Lock()
545 defer a.mu.Unlock()
546 a.cost.Seed(u)
547}
548
549// LastTurnUsage returns the per-turn usage of the most recent
550// completed turn. Drives the "context used" gauge in the status bar
551// without waiting for the next turn to land.
552func (a *Agent) LastTurnUsage() provider.Usage {
553 a.mu.Lock()
554 defer a.mu.Unlock()
555 return a.cost.LastTurn
556}
557
558// SeedLastTurnUsage primes the per-turn snapshot. Used on resume so
559// the gauge reflects the prompt size of the last turn in the session
560// file instead of starting at zero.
561func (a *Agent) SeedLastTurnUsage(u provider.Usage) {
562 a.mu.Lock()
563 defer a.mu.Unlock()
564 a.cost.LastTurn = u
565}
566
567// fireMessageAppended invokes OnMessageAppended without holding the
568// agent mutex, so the host's persistence callback can take its own
569// locks without deadlocking the agent loop. Tolerates a nil hook so
570// non-persisting callers (tests, RPC mode) don't have to set it.
571func (a *Agent) fireMessageAppended(m provider.Message) {
572 if a.OnMessageAppended != nil {
573 a.OnMessageAppended(m)
574 }
575}
576
577// Prompt sends a user message and runs the agent loop until the model
578// stops or an error occurs. Events are delivered via sink in order.
579// sink must not block the caller for long; buffer as needed.
580func (a *Agent) Prompt(ctx context.Context, text string, images []provider.ImageBlock, sink func(AgentEvent)) error {
581 if text == "" && len(images) == 0 {
582 return errors.New("prompt is empty")
583 }
584 if sink == nil {
585 sink = func(AgentEvent) {}
586 }
587 sink = a.wrapSink(sink)
588 content := []provider.Content{}
589 if text != "" {
590 content = append(content, provider.TextBlock{Text: text})
591 }
592 for _, img := range images {
593 content = append(content, img)
594 }
595 user := provider.Message{Role: provider.RoleUser, Content: content, Time: time.Now()}
596
597 a.mu.Lock()
598 a.messages = append(a.messages, user)
599 a.rev++
600 a.mu.Unlock()
601 a.fireMessageAppended(user)
602 sink(EvUserMessage{Message: user})
603
604 return a.runLoop(ctx, sink, a.beginTurn())
605}
606
607// Continue runs the agent loop against the existing transcript. Used
608// after appending tool results manually or to retry.
609func (a *Agent) Continue(ctx context.Context, sink func(AgentEvent)) error {
610 if sink == nil {
611 sink = func(AgentEvent) {}
612 }
613 sink = a.wrapSink(sink)
614 return a.runLoop(ctx, sink, a.beginTurn())
615}
616
617// wrapSink composes the per-call sink with a.OnEvent (if set) so the
618// extension manager (or any other observer) sees every AgentEvent
619// without having to thread itself through every Prompt callsite.
620func (a *Agent) wrapSink(sink func(AgentEvent)) func(AgentEvent) {
621 if a.OnEvent == nil {
622 return sink
623 }
624 obs := a.OnEvent
625 return func(ev AgentEvent) {
626 obs(ev)
627 sink(ev)
628 }
629}
630
631func (a *Agent) runLoop(ctx context.Context, sink func(AgentEvent), requestContext provider.RequestContext) error {
632 a.resetDeniedToolCalls()
633 recoveryUsed := false
634 for step := 1; a.MaxSteps <= 0 || step <= a.MaxSteps; step++ {
635 // Messages queued while the agent was busy are delivered
636 // before the next model call. This is the safe boundary:
637 // any previous tool batch has already completed and its
638 // results have been appended, but no new provider request has
639 // started yet.
640 if pending := a.drainQueuedMessages(); len(pending) > 0 {
641 a.appendQueuedAsUser(pending, sink)
642 }
643
644 sink(EvTurnStart{Step: step, TurnID: requestContext.TurnID})
645 turnContext := ""
646 if a.BeforeTurnContext != nil {
647 var allowed bool
648 var reason string
649 allowed, reason, turnContext = a.BeforeTurnContext(ctx, step)
650 if !allowed {
651 if reason == "" {
652 reason = "turn blocked by extension guard"
653 }
654 sink(EvTurnEnd{Stop: provider.StopError, Err: fmt.Errorf("%s", reason)})
655 sink(EvDone{})
656 return nil
657 }
658 } else if a.BeforeTurn != nil {
659 if allowed, reason := a.BeforeTurn(step); !allowed {
660 if reason == "" {
661 reason = "turn blocked by extension guard"
662 }
663 sink(EvTurnEnd{Stop: provider.StopError, Err: fmt.Errorf("%s", reason)})
664 sink(EvDone{})
665 return nil
666 }
667 }
668
669 a.appendDynamicContext(turnContext)
670
671 var (
672 stop provider.StopReason
673 assistantMsg provider.Message
674 err error
675 )
676 maxAttempts := max(a.MaxRetries+1, 1)
677 for attempt := 0; ; attempt++ {
678 stop, assistantMsg, err = a.oneTurn(ctx, sink, turnContext, requestContext, attempt, maxAttempts)
679 sink(EvTurnEnd{Stop: stop, Err: err})
680 if err == nil {
681 break
682 }
683 retryable := a.canRetryError(err, attempt)
684 var retryCtxErr error
685 if retryable {
686 retryCtxErr = ctx.Err()
687 if retryCtxErr != nil {
688 retryable = false
689 }
690 }
691 reason := classifyRetryReason(err)
692 if !errors.Is(err, context.Canceled) {
693 a.fireRetryLifecycle(RetryLifecycleRecord{
694 Event: RetryLifecycleRequestFailed,
695 Scope: RetryScopeAgent,
696 Attempt: attempt + 1,
697 MaxAttempts: maxAttempts,
698 Reason: reason,
699 Terminal: !retryable,
700 })
701 }
702 if retryCtxErr != nil {
703 return retryCtxErr
704 }
705 if !retryable {
706 break
707 }
708 a.dropLastAssistantMessage()
709 delay := a.retryDelay(attempt)
710 if retryErr := ctx.Err(); retryErr != nil {
711 return retryErr
712 }
713 a.fireRetryLifecycle(RetryLifecycleRecord{
714 Event: RetryLifecycleRetryScheduled,
715 Scope: RetryScopeAgent,
716 Attempt: attempt + 2,
717 MaxAttempts: maxAttempts,
718 Reason: reason,
719 DelayMS: delay.Milliseconds(),
720 })
721 sink(EvRetryScheduled{
722 Scope: RetryScopeAgent,
723 Attempt: attempt + 2,
724 MaxAttempts: maxAttempts,
725 Delay: delay,
726 })
727 if sleepErr := sleepRetry(ctx, delay); sleepErr != nil {
728 return sleepErr
729 }
730 }
731 if err != nil {
732 return err
733 }
734
735 if stop == provider.StopContinue {
736 if err := ctx.Err(); err != nil {
737 sink(EvDone{})
738 return err
739 }
740 continue
741 }
742 if stop == provider.StopToolUse {
743 // Execute each tool call, append a single tool-results message, continue.
744 toolMsg, hadError := a.executeTools(ctx, assistantMsg, sink)
745 a.mu.Lock()
746 a.messages = append(a.messages, toolMsg)
747 a.rev++
748 // OpenAI's chat-completions tool message shape is text-centric.
749 // Vision models reliably consume images when they arrive as user
750 // content, so when a tool result contains images we mirror them
751 // into a synthetic user message immediately after the tool result.
752 // This keeps the transcript self-contained for providers that can
753 // see image blocks in tool messages while making OpenAI vision
754 // models actually receive the image bytes.
755 //
756 // The OpenAI Responses route ("openai-codex") has the same
757 // text-centric tool-output shape: a function_call_output only
758 // carries a string, so images in a tool result never reach the
759 // model. Both providers serialize images correctly when they
760 // arrive as user content, so the mirror covers them both.
761 var imageMirror provider.Message
762 if a.Client != nil && (a.Client.Name() == "openai" || a.Client.Name() == "openai-codex") {
763 if mirror := mirrorToolImagesAsUser(toolMsg); len(mirror.Content) > 0 {
764 a.messages = append(a.messages, mirror)
765 a.rev++
766 imageMirror = mirror
767 }
768 }
769 a.mu.Unlock()
770 a.fireMessageAppended(toolMsg)
771 if len(imageMirror.Content) > 0 {
772 a.fireMessageAppended(imageMirror)
773 }
774 // If context was cancelled during tool execution, bail out.
775 if err := ctx.Err(); err != nil {
776 sink(EvDone{})
777 return err
778 }
779 _ = hadError
780 continue
781 }
782
783 // If the assistant stopped without tool calls but a message was
784 // queued while it was speaking, loop once more so that message
785 // is appended and answered instead of waiting until a later
786 // top-level prompt.
787 if ctx.Err() == nil && a.QueuedMessageCount() > 0 {
788 continue
789 }
790
791 // Bounded normal-turn recovery: a successful normal stop whose
792 // current terminal assistant message carries no final answer gets
793 // one continuation per top-level invocation. Never on errors
794 // (err != nil returns above), cancellation, output/context
795 // limits, or a permission/confirmation refusal.
796 if stop == provider.StopEnd && !a.sawDeniedToolCall() {
797 if reason, incomplete := classifyTerminalAssistant(assistantMsg); incomplete {
798 if !recoveryUsed {
799 if err := ctx.Err(); err != nil {
800 sink(EvDone{})
801 return err
802 }
803 recoveryUsed = true
804 a.AppendUserContext(turnRecoveryPrompt, map[string]string{turnRecoveryMetaKey: "true"})
805 sink(EvTurnRecovery{Reason: reason, Attempt: 1})
806 continue
807 }
808 sink(EvDone{})
809 return ErrIncompleteTurn
810 }
811 }
812
813 // Terminal stop (end, length, error, aborted).
814 sink(EvDone{})
815 return nil
816 }
817 if a.MaxSteps > 0 {
818 sink(EvDone{})
819 return fmt.Errorf("%w: %d", ErrMaxSteps, a.MaxSteps)
820 }
821 return nil
822}
823
824func (a *Agent) canRetryError(err error, attempt int) bool {
825 if err == nil || a.MaxRetries <= 0 || attempt >= a.MaxRetries {
826 return false
827 }
828 var openErr *requestOpenError
829 if errors.As(err, &openErr) {
830 return false
831 }
832 // An idle timeout already bounds a full period without provider progress.
833 // Retrying it would restart that deadline for every attempt, allowing one
834 // silent model continuation to keep an agent running for more than thirty
835 // minutes under the default retry policy.
836 if errors.Is(err, ErrStreamIdleTimeout) || errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) {
837 return false
838 }
839 msg := strings.ToLower(err.Error())
840 if msg == "" || isNonRetryableProviderLimit(msg) {
841 return false
842 }
843 if isStreamNetworkError(err, msg) {
844 return true
845 }
846 needles := []string{
847 "overloaded", "provider returned error", "rate limit", "ratelimit", "too many requests",
848 "429", "http 429", "500", "http 500", "502", "http 502", "503", "http 503", "504", "http 504",
849 "service unavailable", "server error", "internal error", "ended without", "stream ended before",
850 "did not get a response", "timed out", "timeout", "terminated",
851 // OpenAI's ChatGPT/Codex backend returns this generic message (with a
852 // request ID) for transient server failures and explicitly says
853 // "You can retry your request".
854 "an error occurred while processing your request",
855 // Explicit retry guidance emitted by provider backends (OpenAI
856 // Responses, AWS Bedrock stream exceptions) with varying prefixes.
857 "you can retry your request", "try your request again", "please retry your request",
858 // Capacity messages from the ChatGPT/Codex backend, e.g.
859 // "Our servers are currently overloaded. Please try again later."
860 // The trailing advice also shows up on its own for transient
861 // capacity failures; usage/quota limits are filtered out above by
862 // isNonRetryableProviderLimit before this list is consulted.
863 "servers are currently overloaded", "servers are busy", "try again later",
864 }
865 for _, needle := range needles {
866 if strings.Contains(msg, needle) {
867 return true
868 }
869 }
870 return false
871}
872
873// isStreamNetworkError detects transport errors from an opened provider stream.
874// Keep typed EOF checks because WebSocket readers return io.EOF on a peer close,
875// while provider-specific transports may only preserve a textual error.
876func isStreamNetworkError(err error, msg string) bool {
877 if errors.Is(err, io.EOF) || errors.Is(err, io.ErrUnexpectedEOF) {
878 return true
879 }
880 for _, needle := range []string{
881 "connection", "network", "unexpected eof", "broken pipe", "socket hang up",
882 "transport failure", "upstream connect",
883 } {
884 if strings.Contains(msg, needle) {
885 return true
886 }
887 }
888 return false
889}
890
891func isNonRetryableProviderLimit(msg string) bool {
892 needles := []string{
893 "usage limit", "monthly usage limit", "freeusagelimit", "gousagelimit",
894 "available balance", "insufficient_quota", "out of budget", "quota exceeded", "billing",
895 }
896 for _, needle := range needles {
897 if strings.Contains(msg, needle) {
898 return true
899 }
900 }
901 return false
902}
903
904func classifyRetryReason(err error) RetryReason {
905 if err == nil {
906 return RetryReasonUnknown
907 }
908 msg := strings.ToLower(err.Error())
909 if isNonRetryableProviderLimit(msg) {
910 return RetryReasonQuota
911 }
912 if strings.Contains(msg, "overload") || strings.Contains(msg, "servers are busy") {
913 return RetryReasonOverload
914 }
915 if strings.Contains(msg, "rate limit") || strings.Contains(msg, "ratelimit") || strings.Contains(msg, "too many requests") || strings.Contains(msg, "429") {
916 return RetryReasonRateLimit
917 }
918 if errors.Is(err, context.DeadlineExceeded) || strings.Contains(msg, "timeout") || strings.Contains(msg, "timed out") {
919 return RetryReasonTimeout
920 }
921 if isStreamNetworkError(err, msg) {
922 return RetryReasonNetwork
923 }
924 for _, needle := range []string{"context window", "context length", "maximum context", "too many tokens"} {
925 if strings.Contains(msg, needle) {
926 return RetryReasonContextWindow
927 }
928 }
929 for _, needle := range []string{"unauthorized", "authentication", "invalid api key", "invalid_api_key", "http 401", "http 403"} {
930 if strings.Contains(msg, needle) {
931 return RetryReasonAuth
932 }
933 }
934 for _, needle := range []string{"server error", "internal error", "service unavailable", "http 500", "http 502", "http 503", "http 504", "provider returned error"} {
935 if strings.Contains(msg, needle) {
936 return RetryReasonServer
937 }
938 }
939 for _, needle := range []string{"bad request", "unsupported", "invalid parameter", "http 400", "http 404", "http 422"} {
940 if strings.Contains(msg, needle) {
941 return RetryReasonClient
942 }
943 }
944 return RetryReasonUnknown
945}
946
947func (a *Agent) fireRetryLifecycle(record RetryLifecycleRecord) {
948 if a.OnRetryLifecycle != nil {
949 a.OnRetryLifecycle(record)
950 }
951}
952
953func (a *Agent) retryDelay(attempt int) time.Duration {
954 base := a.RetryBaseDelay
955 if base <= 0 {
956 base = 2 * time.Second
957 }
958 return base * time.Duration(1<<attempt)
959}
960
961func sleepRetry(ctx context.Context, d time.Duration) error {
962 if d <= 0 {
963 return ctx.Err()
964 }
965 t := time.NewTimer(d)
966 defer t.Stop()
967 select {
968 case <-ctx.Done():
969 return ctx.Err()
970 case <-t.C:
971 return nil
972 }
973}
974
975func (a *Agent) dropLastAssistantMessage() {
976 a.mu.Lock()
977 defer a.mu.Unlock()
978 if n := len(a.messages); n > 0 && a.messages[n-1].Role == provider.RoleAssistant {
979 a.messages = a.messages[:n-1]
980 a.rev++
981 }
982}
983
984// requestLifecycleSink adapts provider-owned pre-stream retries into core
985// events without letting provider concerns leak into mode renderers.
986type requestLifecycleSink struct {
987 sink func(AgentEvent)
988 observe func(RetryLifecycleRecord)
989 provider string
990 model string
991 lastReason RetryReason
992}
993
994func (s *requestLifecycleSink) CacheDiagnostics(diagnostics provider.CacheDiagnostics) {
995 s.sink(EvCacheDiagnostics{
996 Eligible: diagnostics.Eligible,
997 Mode: diagnostics.Mode,
998 Transport: diagnostics.Transport,
999 Continuation: diagnostics.Continuation,
1000 })
1001}
1002
1003func (s *requestLifecycleSink) RequestAttempt(attempt, maxAttempts int) {
1004 s.sink(EvRequestStarted{
1005 Provider: s.provider,
1006 Model: s.model,
1007 Scope: RetryScopeProvider,
1008 Attempt: attempt,
1009 MaxAttempts: maxAttempts,
1010 })
1011}
1012
1013func (s *requestLifecycleSink) RequestFailed(attempt, maxAttempts int, reason provider.RequestFailureReason, terminal bool) {
1014 s.lastReason = retryReasonFromProvider(reason)
1015 if s.observe != nil {
1016 s.observe(RetryLifecycleRecord{
1017 Event: RetryLifecycleRequestFailed,
1018 Scope: RetryScopeProvider,
1019 Attempt: attempt,
1020 MaxAttempts: maxAttempts,
1021 Reason: s.lastReason,
1022 Terminal: terminal,
1023 })
1024 }
1025}
1026
1027func (s *requestLifecycleSink) RetryScheduled(attempt, maxAttempts int, delay time.Duration) {
1028 s.sink(EvRetryScheduled{
1029 Scope: RetryScopeProvider,
1030 Attempt: attempt,
1031 MaxAttempts: maxAttempts,
1032 Delay: delay,
1033 })
1034 if s.observe != nil {
1035 reason := s.lastReason
1036 if reason == "" {
1037 reason = RetryReasonUnknown
1038 }
1039 s.observe(RetryLifecycleRecord{
1040 Event: RetryLifecycleRetryScheduled,
1041 Scope: RetryScopeProvider,
1042 Attempt: attempt,
1043 MaxAttempts: maxAttempts,
1044 Reason: reason,
1045 DelayMS: delay.Milliseconds(),
1046 })
1047 }
1048 s.lastReason = ""
1049}
1050
1051func retryReasonFromProvider(reason provider.RequestFailureReason) RetryReason {
1052 switch reason {
1053 case provider.RequestFailureOverload:
1054 return RetryReasonOverload
1055 case provider.RequestFailureRateLimit:
1056 return RetryReasonRateLimit
1057 case provider.RequestFailureQuota:
1058 return RetryReasonQuota
1059 case provider.RequestFailureServer:
1060 return RetryReasonServer
1061 case provider.RequestFailureNetwork:
1062 return RetryReasonNetwork
1063 case provider.RequestFailureTimeout:
1064 return RetryReasonTimeout
1065 case provider.RequestFailureClient:
1066 return RetryReasonClient
1067 default:
1068 return RetryReasonUnknown
1069 }
1070}
1071
1072// oneTurn calls the LLM once, forwards events, returns the stop reason
1073// and the assembled assistant message (already appended to the transcript).
1074func newStreamIdleTimer(timeout time.Duration) (<-chan time.Time, func(), func()) {
1075 timer := time.NewTimer(timeout)
1076 stop := func() { timer.Stop() }
1077 reset := func() {
1078 if !timer.Stop() {
1079 select {
1080 case <-timer.C:
1081 default:
1082 }
1083 }
1084 timer.Reset(timeout)
1085 }
1086 return timer.C, stop, reset
1087}
1088
1089func (a *Agent) oneTurn(ctx context.Context, sink func(AgentEvent), turnContext string, requestContext provider.RequestContext, attempt, maxAttempts int) (provider.StopReason, provider.Message, error) {
1090 fastMode := a.FastModeEnabled()
1091 if err := provider.ValidateFastMode(a.Client.Name(), fastMode); err != nil {
1092 return provider.StopError, provider.Message{}, err
1093 }
1094 system, tools := a.PromptConfig()
1095 // Repair pairs before projecting the copied provider-input view. The
1096 // repair can add stub results for aborted calls; those results must also
1097 // remain in the request so every tool call still has a matching result.
1098 messages := repairToolUseResultPairs(a.Messages())
1099 requestLifecycle := &requestLifecycleSink{
1100 sink: sink,
1101 observe: a.fireRetryLifecycle,
1102 provider: a.Client.Name(),
1103 model: a.Model,
1104 }
1105 req := provider.Request{
1106 Model: a.Model,
1107 System: system,
1108 // Repair any dangling tool_use blocks before sending. A turn
1109 // aborted mid-flight (cancel, connection drop, ECONNREFUSED to a
1110 // dev server, etc.) can leave an assistant tool_use with no
1111 // matching tool_result in the live transcript. The load-time
1112 // repair in OpenSession only runs on restart, so without this the
1113 // next in-process request is rejected by providers like Anthropic
1114 // with "tool_use ids were found without tool_result blocks". The
1115 // repair is pure and a no-op on already-valid transcripts.
1116 Messages: projectProviderMessages(messages),
1117 Tools: tools.Specs(),
1118 Reasoning: a.Reasoning,
1119 FastMode: fastMode,
1120 MaxTokens: a.MaxTokens,
1121 Temperature: a.Temperature,
1122 Lifecycle: requestLifecycle,
1123 Context: requestContext,
1124 }
1125 sink(EvRequestStarted{
1126 Provider: a.Client.Name(),
1127 Model: a.Model,
1128 Scope: RetryScopeAgent,
1129 Attempt: attempt + 1,
1130 MaxAttempts: maxAttempts,
1131 })
1132 streamCtx, cancelStream := context.WithCancel(ctx)
1133 defer cancelStream()
1134 a.beginProviderRequest()
1135 stream, err := a.Client.Stream(streamCtx, req)
1136 if err != nil {
1137 return provider.StopError, provider.Message{}, &requestOpenError{err: err}
1138 }
1139 // A provider may already have queued events when cancellation wins. Keep a
1140 // consumer attached after this turn returns so bounded producer channels
1141 // cannot strand the provider goroutine and its response body.
1142 defer func() { go drainProviderStream(stream) }()
1143
1144 sink(EvAssistantStart{})
1145
1146 var (
1147 stop provider.StopReason
1148 finalErr error
1149 finalMsg provider.Message
1150 )
1151
1152 var idle <-chan time.Time
1153 var resetIdle func()
1154 if a.StreamIdleTimeout > 0 {
1155 newTimer := a.streamIdleTimer
1156 if newTimer == nil {
1157 newTimer = newStreamIdleTimer
1158 }
1159 var stopIdle func()
1160 idle, stopIdle, resetIdle = newTimer(a.StreamIdleTimeout)
1161 defer stopIdle()
1162 }
1163 handleStreamEvent := func(ev provider.Event) (done bool) {
1164 if resetIdle != nil {
1165 resetIdle()
1166 }
1167 switch e := ev.(type) {
1168 case provider.EventStart:
1169 // nothing
1170 case provider.EventTextDelta:
1171 sink(EvTextDelta{Delta: e.Delta})
1172 case provider.EventToolStart:
1173 sink(EvToolUseStart{ID: e.ID, Name: e.Name})
1174 case provider.EventToolArgs:
1175 sink(EvToolUseArgs{ID: e.ID, Delta: e.Delta})
1176 case provider.EventToolEnd:
1177 sink(EvToolUseEnd{ID: e.ID})
1178 case provider.EventUsage:
1179 cum := a.addUsage(e.Usage)
1180 sink(EvUsage{Usage: e.Usage, Cumulative: cum})
1181 if a.OnUsage != nil {
1182 a.OnUsage(cum)
1183 }
1184 case provider.EventDone:
1185 stop = e.Stop
1186 finalErr = e.Err
1187 finalMsg = provider.WithMessageOrigin(e.Message, a.Client.Name(), req.Model)
1188 return true
1189 }
1190 return false
1191 }
1192 for {
1193 // Consume no more than one buffered event before honoring cancellation.
1194 select {
1195 case ev, ok := <-stream:
1196 if !ok || handleStreamEvent(ev) {
1197 goto streamDone
1198 }
1199 default:
1200 }
1201 if err := streamCtx.Err(); err != nil {
1202 return provider.StopError, finalMsg, err
1203 }
1204 select {
1205 case <-streamCtx.Done():
1206 return provider.StopError, finalMsg, streamCtx.Err()
1207 case <-idle:
1208 return provider.StopError, finalMsg, ErrStreamIdleTimeout
1209 case ev, ok := <-stream:
1210 if !ok || handleStreamEvent(ev) {
1211 goto streamDone
1212 }
1213 }
1214 }
1215streamDone:
1216
1217 // Append assistant message to transcript. Aborted turns (Esc / Ctrl+C)
1218 // produce partial content. When the partial message is text only we
1219 // keep whatever was streamed up to the cancel so the user does not
1220 // lose visible work (a cut-off summary is still useful). If the
1221 // partial message already contained tool-call blocks we drop the
1222 // whole thing, because an unmatched tool_use would fail the next
1223 // turn with a tool_result mismatch error.
1224 keep := len(finalMsg.Content) > 0
1225 if stop == provider.StopAborted && keep {
1226 hasToolCall := false
1227 for _, c := range finalMsg.Content {
1228 if _, ok := c.(provider.ToolCallBlock); ok {
1229 hasToolCall = true
1230 break
1231 }
1232 }
1233 if hasToolCall {
1234 keep = false
1235 }
1236 }
1237 if keep {
1238 emit := finalMsg
1239 suppress := false
1240
1241 // BeforeAssistantMessage hook: extensions can suppress or
1242 // rewrite the visible text. The transcript keeps the
1243 // model's original output so the model still sees what it
1244 // said on subsequent turns.
1245 if a.BeforeAssistantMessage != nil {
1246 orig := extractText(finalMsg)
1247 if orig != "" {
1248 allowed, _, replacement := a.BeforeAssistantMessage(orig)
1249 if !allowed {
1250 suppress = true
1251 } else if replacement != "" && replacement != orig {
1252 emit = replaceText(finalMsg, replacement)
1253 }
1254 }
1255 }
1256
1257 a.mu.Lock()
1258 a.messages = append(a.messages, finalMsg)
1259 a.rev++
1260 a.mu.Unlock()
1261 a.fireMessageAppended(finalMsg)
1262 if !suppress {
1263 sink(EvAssistantMessage{Message: emit})
1264 }
1265 // Now surface tool calls as EvToolCall events so UIs can render them
1266 // in order before the tool results arrive.
1267 for _, c := range finalMsg.Content {
1268 if tc, ok := c.(provider.ToolCallBlock); ok {
1269 sink(EvToolCall{ID: tc.ID, Name: tc.Name, Args: tc.Arguments})
1270 }
1271 }
1272 }
1273
1274 return stop, finalMsg, finalErr
1275}
1276
1277func drainProviderStream(stream <-chan provider.Event) {
1278 for range stream {
1279 }
1280}
1281
1282// ToolResultCommitError supplies a user-safe explanation when a host rejects
1283// a tool result during durable-state commit. Internal storage errors remain in
1284// Err; callers without this type retain the generic failure message.
1285type ToolResultCommitError struct {
1286 Message string
1287 Err error
1288}
1289
1290func (e *ToolResultCommitError) Error() string {
1291 if e == nil || e.Err == nil {
1292 return "tool result commit failed"
1293 }
1294 return e.Err.Error()
1295}
1296
1297func (e *ToolResultCommitError) Unwrap() error {
1298 if e == nil {
1299 return nil
1300 }
1301 return e.Err
1302}
1303
1304func toolResultCommitMessage(err error) string {
1305 var safe *ToolResultCommitError
1306 if errors.As(err, &safe) && strings.TrimSpace(safe.Message) != "" {
1307 return safe.Message
1308 }
1309 return "tool result state could not be persisted"
1310}
1311
1312// executeTools runs every tool call in the assistant message and returns
1313// a single tool-role message carrying all results.
1314func (a *Agent) executeTools(ctx context.Context, msg provider.Message, sink func(AgentEvent)) (provider.Message, bool) {
1315 var results []provider.Content
1316 var addedTools []string
1317 hadError := false
1318 tools := a.ToolsSnapshot()
1319
1320 for _, c := range msg.Content {
1321 tc, ok := c.(provider.ToolCallBlock)
1322 if !ok {
1323 continue
1324 }
1325 res := a.runOneTool(ctx, tc, tools, sink)
1326 if a.CommitToolResult != nil {
1327 if err := a.CommitToolResult(tc.ID, res); err != nil {
1328 res = ToolResult{
1329 Content: []provider.Content{provider.TextBlock{Text: toolResultCommitMessage(err)}},
1330 IsError: true,
1331 Timing: res.Timing,
1332 }
1333 }
1334 }
1335 if res.IsError {
1336 hadError = true
1337 }
1338 results = append(results, provider.ToolResultBlock{
1339 CallID: tc.ID,
1340 Content: res.Content,
1341 IsError: res.IsError,
1342 Timing: res.Timing,
1343 Context: res.Context,
1344 })
1345 for _, name := range res.ActivateTools {
1346 if _, err := tools.Get(name); err == nil && !containsString(addedTools, name) {
1347 addedTools = append(addedTools, name)
1348 }
1349 }
1350 if a.OnToolResult != nil {
1351 a.OnToolResult(tc.ID, res)
1352 }
1353 if update, ok := res.Details.(PlanUpdate); ok && !res.IsError {
1354 sink(EvPlanUpdate{Update: update})
1355 }
1356 sink(EvToolResult{ID: tc.ID, Result: res})
1357 }
1358
1359 return provider.Message{
1360 Role: provider.RoleTool,
1361 Content: results,
1362 Time: time.Now(),
1363 AddedToolNames: addedTools,
1364 }, hadError
1365}
1366
1367func (a *Agent) runOneTool(ctx context.Context, tc provider.ToolCallBlock, tools Registry, sink func(AgentEvent)) (res ToolResult) {
1368 started := time.Now()
1369 defer func() {
1370 if recover() != nil {
1371 res = ToolResult{
1372 Content: []provider.Content{provider.TextBlock{Text: "tool execution failed"}},
1373 IsError: true,
1374 }
1375 }
1376 res.Timing = &provider.ToolTiming{
1377 StartedAt: started,
1378 CompletedAt: time.Now(),
1379 Duration: time.Since(started),
1380 }
1381 }()
1382
1383 tool, err := tools.Get(tc.Name)
1384 if err != nil {
1385 return ToolResult{
1386 Content: []provider.Content{provider.TextBlock{Text: err.Error()}},
1387 IsError: true,
1388 }
1389 }
1390
1391 args := tc.Arguments
1392
1393 // Intercept hook: an extension or other guard can refuse the
1394 // call before any side effect happens, OR rewrite the args
1395 // seen by the tool. The model sees the reason as the tool
1396 // error, learns from it, and (typically) proposes a different
1397 // action; rewrites are invisible to the model (they apply only
1398 // to the execution).
1399 if a.BeforeToolExecute != nil {
1400 allowed, reason, modified := a.BeforeToolExecute(tc)
1401 if !allowed {
1402 // A guard or confirmation denial is tracked separately from an
1403 // ordinary tool error: a denied invocation must not gain a
1404 // recovery prompt urging more actions.
1405 a.markDeniedToolCall()
1406 if reason == "" {
1407 reason = "tool call refused by extension guard"
1408 }
1409 return ToolResult{
1410 Content: []provider.Content{provider.TextBlock{Text: reason}},
1411 IsError: true,
1412 }
1413 }
1414 if len(modified) > 0 && json.Valid(modified) {
1415 if policy, ok := tool.(ToolArgumentRewritePolicy); ok && !policy.AllowArgumentRewrite() {
1416 return ToolResult{
1417 Content: []provider.Content{provider.TextBlock{Text: "tool arguments cannot be rewritten"}},
1418 IsError: true,
1419 }
1420 }
1421 args = modified
1422 }
1423 }
1424
1425 if len(args) == 0 {
1426 args = json.RawMessage("{}")
1427 }
1428
1429 // Tool panics are recovered by the invocation-level defer above so
1430 // guard, lookup, progress, and Execute panics all receive timing.
1431 sink(EvToolExecutionStarted{ID: tc.ID, Name: tc.Name})
1432 out, err := tool.Execute(ctx, args, func(text string) {
1433 sink(EvToolProgress{ID: tc.ID, Text: text})
1434 })
1435 if err != nil {
1436 // A refused execution (permission scope, allowlist, jail, or
1437 // admission gate) is tracked separately from an ordinary tool
1438 // error: like a guard or confirmation denial, it must not gain
1439 // a recovery prompt urging more actions.
1440 if asToolDeniedError(err) {
1441 a.markDeniedToolCall()
1442 }
1443 if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) {
1444 return ToolResult{
1445 Content: []provider.Content{provider.TextBlock{Text: "aborted: " + err.Error()}},
1446 IsError: true,
1447 }
1448 }
1449 return ToolResult{
1450 Content: []provider.Content{provider.TextBlock{Text: err.Error()}},
1451 IsError: true,
1452 }
1453 }
1454 return out
1455}
1456
1457// extractText concatenates all TextBlock content in a message. Used
1458// by BeforeAssistantMessage so guards see a single string instead of
1459// having to walk provider.Content themselves.
1460func mirrorToolImagesAsUser(msg provider.Message) provider.Message {
1461 var content []provider.Content
1462 for _, c := range msg.Content {
1463 tr, ok := c.(provider.ToolResultBlock)
1464 if !ok {
1465 continue
1466 }
1467 for _, inner := range tr.Content {
1468 switch v := inner.(type) {
1469 case provider.TextBlock:
1470 // Keep short textual context so the model understands why
1471 // the images appeared, but don't duplicate giant read
1472 // outputs verbatim.
1473 if len(v.Text) > 0 && len(v.Text) <= 500 {
1474 content = append(content, v)
1475 }
1476 case provider.ImageBlock:
1477 content = append(content, v)
1478 }
1479 }
1480 }
1481 if len(content) == 0 {
1482 return provider.Message{}
1483 }
1484 prefix := provider.TextBlock{Text: "Tool output included the following image content:"}
1485 content = append([]provider.Content{prefix}, content...)
1486 return provider.Message{Role: provider.RoleUser, Content: content, Time: time.Now()}
1487}
1488
1489const (
1490 maxTurnContextBytes = 16 * 1024
1491 turnContextTruncatedMarker = "\n[extension context truncated]"
1492)
1493
1494func boundedTurnContext(text string) string {
1495 text = strings.TrimSpace(text)
1496 if len(text) <= maxTurnContextBytes {
1497 return text
1498 }
1499 limit := maxTurnContextBytes - len(turnContextTruncatedMarker)
1500 cut := limit
1501 for cut > 0 && !utf8.RuneStart(text[cut]) {
1502 cut--
1503 }
1504 return strings.TrimSpace(text[:cut]) + turnContextTruncatedMarker
1505}
1506
1507func containsString(values []string, want string) bool {
1508 for _, value := range values {
1509 if value == want {
1510 return true
1511 }
1512 }
1513 return false
1514}
1515
1516func extractText(msg provider.Message) string {
1517 var out string
1518 for _, c := range msg.Content {
1519 if tb, ok := c.(provider.TextBlock); ok {
1520 if out != "" {
1521 out += "\n"
1522 }
1523 out += tb.Text
1524 }
1525 }
1526 return out
1527}
1528
1529// replaceText returns a copy of msg with every TextBlock replaced by
1530// a single TextBlock containing replacement. Non-text content (tool
1531// calls, etc.) is preserved in order.
1532func replaceText(msg provider.Message, replacement string) provider.Message {
1533 out := provider.Message{Role: msg.Role}
1534 out.Content = make([]provider.Content, 0, len(msg.Content))
1535 replaced := false
1536 for _, c := range msg.Content {
1537 if _, ok := c.(provider.TextBlock); ok {
1538 if !replaced {
1539 out.Content = append(out.Content, provider.TextBlock{Text: replacement})
1540 replaced = true
1541 }
1542 continue
1543 }
1544 out.Content = append(out.Content, c)
1545 }
1546 if !replaced {
1547 out.Content = append(out.Content, provider.TextBlock{Text: replacement})
1548 }
1549 return out
1550}