1package core
2
3import (
4 "context"
5 "encoding/json"
6 "errors"
7 "fmt"
8 "strings"
9 "sync"
10 "time"
11 "unicode/utf8"
12
13 "github.com/bnema/zut/packages/provider"
14)
15
16// Agent is a stateful conversation bound to a provider client, a model,
17// and a set of tools.
18type Agent struct {
19 Client provider.Client
20 Model string
21 // System is mutex-protected; use SetSystemPrompt to write it and
22 // PromptConfig to read it.
23 System string
24 // Tools is mutex-protected; use SetTools to write it and PromptConfig
25 // or ToolsSnapshot to read it.
26 Tools Registry
27 MaxSteps int
28 Reasoning string
29 Temperature *float32
30 // FastMode requests the OpenAI fast service tier. The provider boundary
31 // rejects it for providers that do not support that contract.
32 FastMode bool
33
34 // ContextWindow is the effective input context capacity retained from
35 // model resolution. Hosts use it for proactive compaction even when the
36 // model was synthesized for an open-catalog provider.
37 ContextWindow int
38
39 // MaxTokens caps the model's output tokens per turn. Zero leaves
40 // the field unset on the provider request, letting each provider
41 // apply its own default (which can be conservative, e.g. Bedrock
42 // defaults to 4096, truncating long writes/edits). Hosts populate
43 // this from the resolved model's MaxOutput so large single-turn
44 // responses aren't silently cut off with stopReason=length.
45 MaxTokens int
46
47 // BeforeToolExecute, if set, is called immediately before each
48 // tool runs. Returning (allowed=false, reason) short-circuits
49 // the call with an error result containing reason. Optionally,
50 // returning a non-nil modifiedArgs replaces the JSON args the
51 // tool will see, which lets guards redact / augment / patch the
52 // model's request without rewriting the transcript. Empty or
53 // malformed modifiedArgs is ignored.
54 BeforeToolExecute func(call provider.ToolCallBlock) (allowed bool, reason string, modifiedArgs json.RawMessage)
55
56 // BeforeTurn, if set, is called before each turn's model call.
57 // Returning (allowed=false, reason) aborts the turn; reason is
58 // surfaced as an assistant-like status line. Used for rate-
59 // limiting, business-hour gates, and deny-by-default setups.
60 BeforeTurn func(step int) (allowed bool, reason string)
61
62 // BeforeTurnContext is the richer turn-start hook. It receives the
63 // current turn context so blocking interception can honor cancellation.
64 // It may return a bounded, non-transcript context string that is appended
65 // to the provider request's system prompt for this turn only. When set, it
66 // replaces BeforeTurn so a host can combine approval and context
67 // collection into one lifecycle round trip.
68 BeforeTurnContext func(ctx context.Context, step int) (allowed bool, reason, context string)
69
70 // OnToolResult, if set, receives extension/tool metadata immediately
71 // after each tool call completes. Details are host-owned metadata and
72 // are never sent to the provider; hosts may persist selected details
73 // with the active session.
74 OnToolResult func(id string, result ToolResult)
75
76 // BeforeAssistantMessage, if set, is called after the model's
77 // final assistant message is assembled but before it's appended
78 // to the transcript. Returning (allowed=false) suppresses both
79 // the transcript append and the UI event. A non-empty
80 // replacement rewrites the visible text for the user while
81 // leaving the model's original text in the transcript (so the
82 // model can still see what it said in subsequent turns).
83 BeforeAssistantMessage func(text string) (allowed bool, reason, replacement string)
84
85 // MaxRetries controls agent-level retries for transient provider
86 // failures that arrive after the HTTP stream opens (for example
87 // Anthropic overloaded_error). Zero disables this retry layer.
88 // RetryBaseDelay is doubled for each attempt; zero uses 2s.
89 MaxRetries int
90 RetryBaseDelay time.Duration
91
92 // OnEvent, if set, mirrors every AgentEvent the loop emits to
93 // this callback in addition to the per-Prompt sink. Used by the
94 // extension manager to fan events out to subscribed extensions
95 // without each caller having to compose sinks manually.
96 OnEvent func(AgentEvent)
97
98 // OnMessageAppended, if set, fires every time a message is
99 // appended to the in-memory transcript by the agent loop — the
100 // initial user prompt, each finalised assistant message, and
101 // each tool-results message (plus the synthetic OpenAI image
102 // mirror, if any). Hosts wire this to the on-disk session so
103 // that turns are durable as soon as they happen, instead of
104 // only being flushed on a clean exit.
105 OnMessageAppended func(provider.Message)
106
107 // OnUsage, if set, fires after every turn's usage row arrives,
108 // carrying the cumulative usage for the session. Hosts wire
109 // this to the on-disk session so the persisted total stays
110 // current and a crash recovers the right cost figure.
111 OnUsage func(cumulative provider.Usage)
112
113 // OnTranscriptCompacted, if set, fires after Compact replaces the
114 // in-memory transcript with the synthetic summary plus kept tail.
115 // Hosts wire this to append an explicit compaction checkpoint to
116 // the session log; per-message append hooks do not fire for this
117 // wholesale transcript replacement.
118 OnTranscriptCompacted func(messages []provider.Message)
119
120 mu sync.Mutex
121 messages []provider.Message
122 // rev increments whenever the transcript slice is replaced or a
123 // message is appended. The TUI uses it as a cheap redraw cache key
124 // so editor-only typing doesn't copy/rebuild a long transcript on
125 // every keypress.
126 rev uint64
127 cost CostTracker
128
129 // queued holds user messages submitted while the agent is busy.
130 // The loop appends them as normal user messages at safe
131 // boundaries: before the next model call after a tool batch, or
132 // after a text-only assistant turn finishes. It never interrupts
133 // a running tool or cancels an in-flight provider request.
134 queued []string
135}
136
137// NewAgent returns an Agent with sensible defaults.
138func NewAgent(client provider.Client, model, system string, tools Registry) *Agent {
139 return &Agent{
140 Client: client,
141 Model: model,
142 System: system,
143 Tools: tools,
144 MaxSteps: 0, // 0 = unlimited
145 MaxRetries: 3,
146 RetryBaseDelay: 2 * time.Second,
147 }
148}
149
150// SetFastMode changes the service-tier preference for the next model call.
151// It is safe to use while a turn is in flight; the active turn keeps the
152// value it already captured.
153func (a *Agent) SetFastMode(enabled bool) {
154 a.mu.Lock()
155 a.FastMode = enabled
156 a.mu.Unlock()
157}
158
159// FastModeEnabled returns the service-tier preference for a model call.
160func (a *Agent) FastModeEnabled() bool {
161 a.mu.Lock()
162 defer a.mu.Unlock()
163 return a.FastMode
164}
165
166// QueueMessage queues text to be injected as a user message at the
167// next safe boundary of the active agent loop. It is non-blocking in
168// the sense that it never waits for model/tool work; it only takes
169// the transcript mutex briefly. Empty/whitespace-only messages are
170// ignored.
171func (a *Agent) QueueMessage(text string) bool {
172 text = strings.TrimSpace(text)
173 if text == "" {
174 return false
175 }
176 a.mu.Lock()
177 a.queued = append(a.queued, text)
178 a.mu.Unlock()
179 return true
180}
181
182// PendingQueuedMessages returns a snapshot of user messages waiting
183// to be injected. Used by hosts to render the visible "sliding in"
184// chips without consuming them.
185func (a *Agent) PendingQueuedMessages() []string {
186 a.mu.Lock()
187 defer a.mu.Unlock()
188 out := make([]string, len(a.queued))
189 copy(out, a.queued)
190 return out
191}
192
193// QueuedMessageCount returns the number of messages waiting to be
194// injected at the next safe boundary.
195func (a *Agent) QueuedMessageCount() int {
196 a.mu.Lock()
197 defer a.mu.Unlock()
198 return len(a.queued)
199}
200
201// PopQueuedMessage removes and returns the most recently queued
202// message. Hosts use this for the slide-back keybinding.
203func (a *Agent) PopQueuedMessage() (string, bool) {
204 a.mu.Lock()
205 defer a.mu.Unlock()
206 n := len(a.queued)
207 if n == 0 {
208 return "", false
209 }
210 text := a.queued[n-1]
211 a.queued = a.queued[:n-1]
212 return text, true
213}
214
215// DrainQueuedMessages discards and returns every queued message.
216// Hosts use this on explicit cancel/clear so stale follow-ups do
217// not run after the user aborted the turn.
218func (a *Agent) DrainQueuedMessages() []string {
219 return a.drainQueuedMessages()
220}
221
222func (a *Agent) drainQueuedMessages() []string {
223 a.mu.Lock()
224 defer a.mu.Unlock()
225 out := make([]string, len(a.queued))
226 copy(out, a.queued)
227 a.queued = nil
228 return out
229}
230
231func (a *Agent) appendQueuedAsUser(texts []string, sink func(AgentEvent)) {
232 for _, text := range texts {
233 msg := provider.Message{
234 Role: provider.RoleUser,
235 Content: []provider.Content{provider.TextBlock{Text: text}},
236 Time: time.Now(),
237 }
238 a.mu.Lock()
239 a.messages = append(a.messages, msg)
240 a.rev++
241 a.mu.Unlock()
242 a.fireMessageAppended(msg)
243 if sink != nil {
244 sink(EvUserMessage{Message: msg})
245 }
246 }
247}
248
249// Messages returns a copy of the current transcript.
250func (a *Agent) Messages() []provider.Message {
251 a.mu.Lock()
252 defer a.mu.Unlock()
253 out := make([]provider.Message, len(a.messages))
254 copy(out, a.messages)
255 return out
256}
257
258// Revision returns a monotonically increasing transcript version.
259// It is cheap to query and changes whenever Messages() would return
260// different transcript content because of append/set operations.
261func (a *Agent) Revision() uint64 {
262 a.mu.Lock()
263 defer a.mu.Unlock()
264 return a.rev
265}
266
267// SetTools swaps the tool registry. Used by /reload-ext to hand
268// the agent a fresh registry after extension subprocesses have been
269// respawned (and their freshly-registered tools merged in).
270func (a *Agent) SetTools(reg Registry) {
271 a.mu.Lock()
272 a.Tools = reg
273 a.mu.Unlock()
274}
275
276// SetSystemPrompt replaces the system prompt used for subsequent turns.
277func (a *Agent) SetSystemPrompt(system string) {
278 a.mu.Lock()
279 a.System = system
280 a.mu.Unlock()
281}
282
283// SetPromptConfig atomically replaces the system prompt and tool registry.
284// It returns the previous registry so the caller can close resources that are
285// no longer owned by the agent.
286func (a *Agent) SetPromptConfig(system string, tools Registry) Registry {
287 a.mu.Lock()
288 defer a.mu.Unlock()
289 oldTools := a.Tools
290 a.System = system
291 a.Tools = tools
292 return oldTools
293}
294
295// PromptConfig returns a consistent snapshot of the system prompt and tools
296// used to construct a provider request.
297func (a *Agent) PromptConfig() (string, Registry) {
298 a.mu.Lock()
299 defer a.mu.Unlock()
300 tools := make(Registry, len(a.Tools))
301 for name, tool := range a.Tools {
302 tools[name] = tool
303 }
304 return a.System, tools
305}
306
307// ToolsSnapshot returns a copy of the current tool registry.
308func (a *Agent) ToolsSnapshot() Registry {
309 a.mu.Lock()
310 defer a.mu.Unlock()
311 tools := make(Registry, len(a.Tools))
312 for name, tool := range a.Tools {
313 tools[name] = tool
314 }
315 return tools
316}
317
318// SetMessages replaces the transcript (used when resuming a session).
319func (a *Agent) SetMessages(msgs []provider.Message) {
320 a.mu.Lock()
321 defer a.mu.Unlock()
322 a.messages = append(a.messages[:0], msgs...)
323 a.rev++
324}
325
326// AppendUserContext adds a user-role message to the transcript without
327// starting a model turn. Hosts use it for context gathered outside the agent
328// loop, such as the output of an explicitly invoked shell command.
329func (a *Agent) AppendUserContext(text string, meta map[string]string) {
330 msg := provider.Message{
331 Role: provider.RoleUser,
332 Content: []provider.Content{provider.TextBlock{Text: text}},
333 Time: time.Now(),
334 Meta: meta,
335 }
336 a.mu.Lock()
337 a.messages = append(a.messages, msg)
338 a.rev++
339 a.mu.Unlock()
340 a.fireMessageAppended(msg)
341}
342
343// Cost returns the cumulative usage.
344func (a *Agent) Cost() provider.Usage {
345 a.mu.Lock()
346 defer a.mu.Unlock()
347 return a.cost.Total
348}
349
350func (a *Agent) addUsage(usage provider.Usage) provider.Usage {
351 a.mu.Lock()
352 defer a.mu.Unlock()
353 return a.cost.Add(usage)
354}
355
356// SeedCost sets the cumulative usage as a baseline before the first
357// turn runs. Used when transferring state from another agent (model
358// or provider switch) so the running cost meter doesn't reset to 0.
359func (a *Agent) SeedCost(u provider.Usage) {
360 a.mu.Lock()
361 defer a.mu.Unlock()
362 a.cost.Seed(u)
363}
364
365// LastTurnUsage returns the per-turn usage of the most recent
366// completed turn. Drives the "context used" gauge in the status bar
367// without waiting for the next turn to land.
368func (a *Agent) LastTurnUsage() provider.Usage {
369 a.mu.Lock()
370 defer a.mu.Unlock()
371 return a.cost.LastTurn
372}
373
374// SeedLastTurnUsage primes the per-turn snapshot. Used on resume so
375// the gauge reflects the prompt size of the last turn in the session
376// file instead of starting at zero.
377func (a *Agent) SeedLastTurnUsage(u provider.Usage) {
378 a.mu.Lock()
379 defer a.mu.Unlock()
380 a.cost.LastTurn = u
381}
382
383// fireMessageAppended invokes OnMessageAppended without holding the
384// agent mutex, so the host's persistence callback can take its own
385// locks without deadlocking the agent loop. Tolerates a nil hook so
386// non-persisting callers (tests, RPC mode) don't have to set it.
387func (a *Agent) fireMessageAppended(m provider.Message) {
388 if a.OnMessageAppended != nil {
389 a.OnMessageAppended(m)
390 }
391}
392
393// Prompt sends a user message and runs the agent loop until the model
394// stops or an error occurs. Events are delivered via sink in order.
395// sink must not block the caller for long; buffer as needed.
396func (a *Agent) Prompt(ctx context.Context, text string, images []provider.ImageBlock, sink func(AgentEvent)) error {
397 if text == "" && len(images) == 0 {
398 return errors.New("prompt is empty")
399 }
400 if sink == nil {
401 sink = func(AgentEvent) {}
402 }
403 sink = a.wrapSink(sink)
404 content := []provider.Content{}
405 if text != "" {
406 content = append(content, provider.TextBlock{Text: text})
407 }
408 for _, img := range images {
409 content = append(content, img)
410 }
411 user := provider.Message{Role: provider.RoleUser, Content: content, Time: time.Now()}
412
413 a.mu.Lock()
414 a.messages = append(a.messages, user)
415 a.rev++
416 a.mu.Unlock()
417 a.fireMessageAppended(user)
418 sink(EvUserMessage{Message: user})
419
420 return a.runLoop(ctx, sink)
421}
422
423// Continue runs the agent loop against the existing transcript. Used
424// after appending tool results manually or to retry.
425func (a *Agent) Continue(ctx context.Context, sink func(AgentEvent)) error {
426 if sink == nil {
427 sink = func(AgentEvent) {}
428 }
429 sink = a.wrapSink(sink)
430 return a.runLoop(ctx, sink)
431}
432
433// wrapSink composes the per-call sink with a.OnEvent (if set) so the
434// extension manager (or any other observer) sees every AgentEvent
435// without having to thread itself through every Prompt callsite.
436func (a *Agent) wrapSink(sink func(AgentEvent)) func(AgentEvent) {
437 if a.OnEvent == nil {
438 return sink
439 }
440 obs := a.OnEvent
441 return func(ev AgentEvent) {
442 obs(ev)
443 sink(ev)
444 }
445}
446
447func (a *Agent) runLoop(ctx context.Context, sink func(AgentEvent)) error {
448 for step := 1; a.MaxSteps <= 0 || step <= a.MaxSteps; step++ {
449 // Messages queued while the agent was busy are delivered
450 // before the next model call. This is the safe boundary:
451 // any previous tool batch has already completed and its
452 // results have been appended, but no new provider request has
453 // started yet.
454 if pending := a.drainQueuedMessages(); len(pending) > 0 {
455 a.appendQueuedAsUser(pending, sink)
456 }
457
458 sink(EvTurnStart{Step: step})
459 turnContext := ""
460 if a.BeforeTurnContext != nil {
461 var allowed bool
462 var reason string
463 allowed, reason, turnContext = a.BeforeTurnContext(ctx, step)
464 if !allowed {
465 if reason == "" {
466 reason = "turn blocked by extension guard"
467 }
468 sink(EvTurnEnd{Stop: provider.StopError, Err: fmt.Errorf("%s", reason)})
469 sink(EvDone{})
470 return nil
471 }
472 } else if a.BeforeTurn != nil {
473 if allowed, reason := a.BeforeTurn(step); !allowed {
474 if reason == "" {
475 reason = "turn blocked by extension guard"
476 }
477 sink(EvTurnEnd{Stop: provider.StopError, Err: fmt.Errorf("%s", reason)})
478 sink(EvDone{})
479 return nil
480 }
481 }
482
483 var (
484 stop provider.StopReason
485 assistantMsg provider.Message
486 err error
487 )
488 maxAttempts := max(a.MaxRetries+1, 1)
489 for attempt := 0; ; attempt++ {
490 stop, assistantMsg, err = a.oneTurn(ctx, sink, turnContext, attempt, maxAttempts)
491 sink(EvTurnEnd{Stop: stop, Err: err})
492 if err == nil || !a.canRetryError(err, attempt) {
493 break
494 }
495 a.dropLastAssistantMessage()
496 delay := a.retryDelay(attempt)
497 if retryErr := ctx.Err(); retryErr != nil {
498 return retryErr
499 }
500 sink(EvRetryScheduled{
501 Scope: RetryScopeAgent,
502 Attempt: attempt + 2,
503 MaxAttempts: maxAttempts,
504 Delay: delay,
505 })
506 if sleepErr := sleepRetry(ctx, delay); sleepErr != nil {
507 return sleepErr
508 }
509 }
510 if err != nil {
511 return err
512 }
513
514 if stop == provider.StopToolUse {
515 // Execute each tool call, append a single tool-results message, continue.
516 toolMsg, hadError := a.executeTools(ctx, assistantMsg, sink)
517 a.mu.Lock()
518 a.messages = append(a.messages, toolMsg)
519 a.rev++
520 // OpenAI's chat-completions tool message shape is text-centric.
521 // Vision models reliably consume images when they arrive as user
522 // content, so when a tool result contains images we mirror them
523 // into a synthetic user message immediately after the tool result.
524 // This keeps the transcript self-contained for providers that can
525 // see image blocks in tool messages while making OpenAI vision
526 // models actually receive the image bytes.
527 //
528 // The OpenAI Responses route ("openai-codex") has the same
529 // text-centric tool-output shape: a function_call_output only
530 // carries a string, so images in a tool result never reach the
531 // model. Both providers serialize images correctly when they
532 // arrive as user content, so the mirror covers them both.
533 var imageMirror provider.Message
534 if a.Client != nil && (a.Client.Name() == "openai" || a.Client.Name() == "openai-codex") {
535 if mirror := mirrorToolImagesAsUser(toolMsg); len(mirror.Content) > 0 {
536 a.messages = append(a.messages, mirror)
537 a.rev++
538 imageMirror = mirror
539 }
540 }
541 a.mu.Unlock()
542 a.fireMessageAppended(toolMsg)
543 if len(imageMirror.Content) > 0 {
544 a.fireMessageAppended(imageMirror)
545 }
546 // If context was cancelled during tool execution, bail out.
547 if err := ctx.Err(); err != nil {
548 sink(EvDone{})
549 return err
550 }
551 _ = hadError
552 continue
553 }
554
555 // If the assistant stopped without tool calls but a message was
556 // queued while it was speaking, loop once more so that message
557 // is appended and answered instead of waiting until a later
558 // top-level prompt.
559 if ctx.Err() == nil && a.QueuedMessageCount() > 0 {
560 continue
561 }
562
563 // Terminal stop (end, length, error, aborted).
564 sink(EvDone{})
565 return nil
566 }
567 if a.MaxSteps > 0 {
568 sink(EvDone{})
569 return fmt.Errorf("max steps (%d) exceeded", a.MaxSteps)
570 }
571 return nil
572}
573
574func (a *Agent) canRetryError(err error, attempt int) bool {
575 if err == nil || a.MaxRetries <= 0 || attempt >= a.MaxRetries {
576 return false
577 }
578 if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) {
579 return false
580 }
581 msg := strings.ToLower(err.Error())
582 if msg == "" || isNonRetryableProviderLimit(msg) {
583 return false
584 }
585 needles := []string{
586 "overloaded", "provider returned error", "rate limit", "ratelimit", "too many requests",
587 "429", "http 429", "500", "http 500", "502", "http 502", "503", "http 503", "504", "http 504",
588 "service unavailable", "server error", "internal error", "network error", "connection error",
589 "connection refused", "connection lost", "fetch failed", "upstream connect", "reset before headers",
590 "socket hang up", "ended without", "stream ended before", "did not get a response", "timed out",
591 "timeout", "terminated", "unexpected eof", "transport failure",
592 // OpenAI's ChatGPT/Codex backend returns this generic message (with a
593 // request ID) for transient server failures and explicitly says
594 // "You can retry your request".
595 "an error occurred while processing your request",
596 // Explicit retry guidance emitted by provider backends (OpenAI
597 // Responses, AWS Bedrock stream exceptions) with varying prefixes.
598 "you can retry your request", "try your request again", "please retry your request",
599 // Capacity messages from the ChatGPT/Codex backend, e.g.
600 // "Our servers are currently overloaded. Please try again later."
601 // The trailing advice also shows up on its own for transient
602 // capacity failures; usage/quota limits are filtered out above by
603 // isNonRetryableProviderLimit before this list is consulted.
604 "servers are currently overloaded", "servers are busy", "try again later",
605 }
606 for _, needle := range needles {
607 if strings.Contains(msg, needle) {
608 return true
609 }
610 }
611 return false
612}
613
614func isNonRetryableProviderLimit(msg string) bool {
615 needles := []string{
616 "usage limit", "monthly usage limit", "freeusagelimit", "gousagelimit",
617 "available balance", "insufficient_quota", "out of budget", "quota exceeded", "billing",
618 }
619 for _, needle := range needles {
620 if strings.Contains(msg, needle) {
621 return true
622 }
623 }
624 return false
625}
626
627func (a *Agent) retryDelay(attempt int) time.Duration {
628 base := a.RetryBaseDelay
629 if base <= 0 {
630 base = 2 * time.Second
631 }
632 return base * time.Duration(1<<attempt)
633}
634
635func sleepRetry(ctx context.Context, d time.Duration) error {
636 if d <= 0 {
637 return ctx.Err()
638 }
639 t := time.NewTimer(d)
640 defer t.Stop()
641 select {
642 case <-ctx.Done():
643 return ctx.Err()
644 case <-t.C:
645 return nil
646 }
647}
648
649func (a *Agent) dropLastAssistantMessage() {
650 a.mu.Lock()
651 defer a.mu.Unlock()
652 if n := len(a.messages); n > 0 && a.messages[n-1].Role == provider.RoleAssistant {
653 a.messages = a.messages[:n-1]
654 a.rev++
655 }
656}
657
658// requestLifecycleSink adapts provider-owned pre-stream retries into core
659// events without letting provider concerns leak into mode renderers.
660type requestLifecycleSink struct {
661 sink func(AgentEvent)
662 provider string
663 model string
664}
665
666func (s requestLifecycleSink) RequestAttempt(attempt, maxAttempts int) {
667 s.sink(EvRequestStarted{
668 Provider: s.provider,
669 Model: s.model,
670 Scope: RetryScopeProvider,
671 Attempt: attempt,
672 MaxAttempts: maxAttempts,
673 })
674}
675
676func (s requestLifecycleSink) RetryScheduled(attempt, maxAttempts int, delay time.Duration) {
677 s.sink(EvRetryScheduled{
678 Scope: RetryScopeProvider,
679 Attempt: attempt,
680 MaxAttempts: maxAttempts,
681 Delay: delay,
682 })
683}
684
685// oneTurn calls the LLM once, forwards events, returns the stop reason
686// and the assembled assistant message (already appended to the transcript).
687func (a *Agent) oneTurn(ctx context.Context, sink func(AgentEvent), turnContext string, attempt, maxAttempts int) (provider.StopReason, provider.Message, error) {
688 fastMode := a.FastModeEnabled()
689 if err := provider.ValidateFastMode(a.Client.Name(), fastMode); err != nil {
690 return provider.StopError, provider.Message{}, err
691 }
692 system, tools := a.PromptConfig()
693 if contextText := boundedTurnContext(turnContext); contextText != "" {
694 if system != "" {
695 system += "\n\n"
696 }
697 system += "[Extension context for this turn]\n" + contextText
698 }
699 // Repair pairs before projecting the copied provider-input view. The
700 // repair can add stub results for aborted calls; those results must also
701 // remain in the request so every tool call still has a matching result.
702 messages := repairToolUseResultPairs(a.Messages())
703 req := provider.Request{
704 Model: a.Model,
705 System: system,
706 // Repair any dangling tool_use blocks before sending. A turn
707 // aborted mid-flight (cancel, connection drop, ECONNREFUSED to a
708 // dev server, etc.) can leave an assistant tool_use with no
709 // matching tool_result in the live transcript. The load-time
710 // repair in OpenSession only runs on restart, so without this the
711 // next in-process request is rejected by providers like Anthropic
712 // with "tool_use ids were found without tool_result blocks". The
713 // repair is pure and a no-op on already-valid transcripts.
714 Messages: projectToolResultMessages(messages),
715 Tools: tools.Specs(),
716 Reasoning: a.Reasoning,
717 FastMode: fastMode,
718 MaxTokens: a.MaxTokens,
719 Temperature: a.Temperature,
720 Lifecycle: requestLifecycleSink{
721 sink: sink,
722 provider: a.Client.Name(),
723 model: a.Model,
724 },
725 }
726 sink(EvRequestStarted{
727 Provider: a.Client.Name(),
728 Model: a.Model,
729 Scope: RetryScopeAgent,
730 Attempt: attempt + 1,
731 MaxAttempts: maxAttempts,
732 })
733 stream, err := a.Client.Stream(ctx, req)
734 if err != nil {
735 return provider.StopError, provider.Message{}, err
736 }
737
738 sink(EvAssistantStart{})
739
740 var (
741 stop provider.StopReason
742 finalErr error
743 finalMsg provider.Message
744 )
745
746 for ev := range stream {
747 switch e := ev.(type) {
748 case provider.EventStart:
749 // nothing
750 case provider.EventTextDelta:
751 sink(EvTextDelta{Delta: e.Delta})
752 case provider.EventToolStart:
753 sink(EvToolUseStart{ID: e.ID, Name: e.Name})
754 case provider.EventToolArgs:
755 sink(EvToolUseArgs{ID: e.ID, Delta: e.Delta})
756 case provider.EventToolEnd:
757 sink(EvToolUseEnd{ID: e.ID})
758 case provider.EventUsage:
759 cum := a.addUsage(e.Usage)
760 sink(EvUsage{Usage: e.Usage, Cumulative: cum})
761 if a.OnUsage != nil {
762 a.OnUsage(cum)
763 }
764 case provider.EventDone:
765 stop = e.Stop
766 finalErr = e.Err
767 finalMsg = e.Message
768 }
769 }
770
771 // Append assistant message to transcript. Aborted turns (Esc / Ctrl+C)
772 // produce partial content. When the partial message is text only we
773 // keep whatever was streamed up to the cancel so the user does not
774 // lose visible work (a cut-off summary is still useful). If the
775 // partial message already contained tool-call blocks we drop the
776 // whole thing, because an unmatched tool_use would fail the next
777 // turn with a tool_result mismatch error.
778 keep := len(finalMsg.Content) > 0
779 if stop == provider.StopAborted && keep {
780 hasToolCall := false
781 for _, c := range finalMsg.Content {
782 if _, ok := c.(provider.ToolCallBlock); ok {
783 hasToolCall = true
784 break
785 }
786 }
787 if hasToolCall {
788 keep = false
789 }
790 }
791 if keep {
792 emit := finalMsg
793 suppress := false
794
795 // BeforeAssistantMessage hook: extensions can suppress or
796 // rewrite the visible text. The transcript keeps the
797 // model's original output so the model still sees what it
798 // said on subsequent turns.
799 if a.BeforeAssistantMessage != nil {
800 orig := extractText(finalMsg)
801 if orig != "" {
802 allowed, _, replacement := a.BeforeAssistantMessage(orig)
803 if !allowed {
804 suppress = true
805 } else if replacement != "" && replacement != orig {
806 emit = replaceText(finalMsg, replacement)
807 }
808 }
809 }
810
811 a.mu.Lock()
812 a.messages = append(a.messages, finalMsg)
813 a.rev++
814 a.mu.Unlock()
815 a.fireMessageAppended(finalMsg)
816 if !suppress {
817 sink(EvAssistantMessage{Message: emit})
818 }
819 // Now surface tool calls as EvToolCall events so UIs can render them
820 // in order before the tool results arrive.
821 for _, c := range finalMsg.Content {
822 if tc, ok := c.(provider.ToolCallBlock); ok {
823 sink(EvToolCall{ID: tc.ID, Name: tc.Name, Args: tc.Arguments})
824 }
825 }
826 }
827
828 return stop, finalMsg, finalErr
829}
830
831// executeTools runs every tool call in the assistant message and returns
832// a single tool-role message carrying all results.
833func (a *Agent) executeTools(ctx context.Context, msg provider.Message, sink func(AgentEvent)) (provider.Message, bool) {
834 var results []provider.Content
835 var addedTools []string
836 hadError := false
837 tools := a.ToolsSnapshot()
838
839 for _, c := range msg.Content {
840 tc, ok := c.(provider.ToolCallBlock)
841 if !ok {
842 continue
843 }
844 res := a.runOneTool(ctx, tc, tools, sink)
845 if res.IsError {
846 hadError = true
847 }
848 results = append(results, provider.ToolResultBlock{
849 CallID: tc.ID,
850 Content: res.Content,
851 IsError: res.IsError,
852 })
853 for _, name := range res.ActivateTools {
854 if _, err := tools.Get(name); err == nil && !containsString(addedTools, name) {
855 addedTools = append(addedTools, name)
856 }
857 }
858 if a.OnToolResult != nil {
859 a.OnToolResult(tc.ID, res)
860 }
861 sink(EvToolResult{ID: tc.ID, Result: res})
862 }
863
864 return provider.Message{
865 Role: provider.RoleTool,
866 Content: results,
867 Time: time.Now(),
868 AddedToolNames: addedTools,
869 }, hadError
870}
871
872func (a *Agent) runOneTool(ctx context.Context, tc provider.ToolCallBlock, tools Registry, sink func(AgentEvent)) ToolResult {
873 tool, err := tools.Get(tc.Name)
874 if err != nil {
875 return ToolResult{
876 Content: []provider.Content{provider.TextBlock{Text: err.Error()}},
877 IsError: true,
878 }
879 }
880
881 args := tc.Arguments
882
883 // Intercept hook: an extension or other guard can refuse the
884 // call before any side effect happens, OR rewrite the args
885 // seen by the tool. The model sees the reason as the tool
886 // error, learns from it, and (typically) proposes a different
887 // action; rewrites are invisible to the model (they apply only
888 // to the execution).
889 if a.BeforeToolExecute != nil {
890 allowed, reason, modified := a.BeforeToolExecute(tc)
891 if !allowed {
892 if reason == "" {
893 reason = "tool call refused by extension guard"
894 }
895 return ToolResult{
896 Content: []provider.Content{provider.TextBlock{Text: reason}},
897 IsError: true,
898 }
899 }
900 if len(modified) > 0 && json.Valid(modified) {
901 args = modified
902 }
903 }
904
905 if len(args) == 0 {
906 args = json.RawMessage("{}")
907 }
908
909 // Recover panics so a buggy tool does not crash the agent.
910 var res ToolResult
911 func() {
912 defer func() {
913 if r := recover(); r != nil {
914 res = ToolResult{
915 Content: []provider.Content{provider.TextBlock{Text: fmt.Sprintf("panic: %v", r)}},
916 IsError: true,
917 }
918 }
919 }()
920 sink(EvToolExecutionStarted{ID: tc.ID, Name: tc.Name})
921 out, err := tool.Execute(ctx, args, func(text string) {
922 sink(EvToolProgress{ID: tc.ID, Text: text})
923 })
924 if err != nil {
925 if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) {
926 res = ToolResult{
927 Content: []provider.Content{provider.TextBlock{Text: "aborted: " + err.Error()}},
928 IsError: true,
929 }
930 return
931 }
932 res = ToolResult{
933 Content: []provider.Content{provider.TextBlock{Text: err.Error()}},
934 IsError: true,
935 }
936 return
937 }
938 res = out
939 }()
940 return res
941}
942
943// extractText concatenates all TextBlock content in a message. Used
944// by BeforeAssistantMessage so guards see a single string instead of
945// having to walk provider.Content themselves.
946func mirrorToolImagesAsUser(msg provider.Message) provider.Message {
947 var content []provider.Content
948 for _, c := range msg.Content {
949 tr, ok := c.(provider.ToolResultBlock)
950 if !ok {
951 continue
952 }
953 for _, inner := range tr.Content {
954 switch v := inner.(type) {
955 case provider.TextBlock:
956 // Keep short textual context so the model understands why
957 // the images appeared, but don't duplicate giant read
958 // outputs verbatim.
959 if len(v.Text) > 0 && len(v.Text) <= 500 {
960 content = append(content, v)
961 }
962 case provider.ImageBlock:
963 content = append(content, v)
964 }
965 }
966 }
967 if len(content) == 0 {
968 return provider.Message{}
969 }
970 prefix := provider.TextBlock{Text: "Tool output included the following image content:"}
971 content = append([]provider.Content{prefix}, content...)
972 return provider.Message{Role: provider.RoleUser, Content: content, Time: time.Now()}
973}
974
975const (
976 maxTurnContextBytes = 16 * 1024
977 turnContextTruncatedMarker = "\n[extension context truncated]"
978)
979
980func boundedTurnContext(text string) string {
981 text = strings.TrimSpace(text)
982 if len(text) <= maxTurnContextBytes {
983 return text
984 }
985 limit := maxTurnContextBytes - len(turnContextTruncatedMarker)
986 cut := limit
987 for cut > 0 && !utf8.RuneStart(text[cut]) {
988 cut--
989 }
990 return strings.TrimSpace(text[:cut]) + turnContextTruncatedMarker
991}
992
993func containsString(values []string, want string) bool {
994 for _, value := range values {
995 if value == want {
996 return true
997 }
998 }
999 return false
1000}
1001
1002func extractText(msg provider.Message) string {
1003 var out string
1004 for _, c := range msg.Content {
1005 if tb, ok := c.(provider.TextBlock); ok {
1006 if out != "" {
1007 out += "\n"
1008 }
1009 out += tb.Text
1010 }
1011 }
1012 return out
1013}
1014
1015// replaceText returns a copy of msg with every TextBlock replaced by
1016// a single TextBlock containing replacement. Non-text content (tool
1017// calls, etc.) is preserved in order.
1018func replaceText(msg provider.Message, replacement string) provider.Message {
1019 out := provider.Message{Role: msg.Role}
1020 out.Content = make([]provider.Content, 0, len(msg.Content))
1021 replaced := false
1022 for _, c := range msg.Content {
1023 if _, ok := c.(provider.TextBlock); ok {
1024 if !replaced {
1025 out.Content = append(out.Content, provider.TextBlock{Text: replacement})
1026 replaced = true
1027 }
1028 continue
1029 }
1030 out.Content = append(out.Content, c)
1031 }
1032 if !replaced {
1033 out.Content = append(out.Content, provider.TextBlock{Text: replacement})
1034 }
1035 return out
1036}