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