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