refactor(chatloop): extract chat run loop from processor

This commit is contained in:
Kyle Carberry
2026-02-24 17:25:00 -05:00
parent 384affd2ed
commit cc8b7f90e1
4 changed files with 1046 additions and 708 deletions
+21 -631
View File
@@ -9,7 +9,6 @@ import (
"io"
"net/http"
"net/url"
"strconv"
"strings"
"sync"
"time"
@@ -30,6 +29,7 @@ import (
"golang.org/x/xerrors"
"cdr.dev/slog/v3"
"github.com/coder/coder/v2/coderd/chatd/chatloop"
"github.com/coder/coder/v2/coderd/chatd/chatprompt"
"github.com/coder/coder/v2/coderd/chatd/chatprovider"
"github.com/coder/coder/v2/coderd/database"
@@ -65,7 +65,6 @@ const (
"Omit pleasantries and next-step suggestions."
contextCompressionSystemSummaryPrefix = "Summary of earlier chat context:"
contextCompressionSummaryTruncatedTag = "[summary truncated for live stream]"
interruptedToolResultErrorMessage = "tool call was interrupted before it produced a result"
maxCreateWorkspaceBuildLogLines = 120
maxCreateWorkspaceBuildLogChars = 16 * 1024
@@ -1599,502 +1598,32 @@ func (p *Processor) runChatWithAgent(
p.publishMessage(chat.ID, toolMessage)
return nil
}
var (
stepStateMu sync.Mutex
streamToolNames map[string]string
streamReasoningTitles map[string]string
streamReasoningText map[string]string
stepToolResults []chatprompt.ToolResultBlock
stepAssistantDraft []fantasy.Content
stepToolCallIndexByID map[string]int
)
resetStepState := func() {
stepStateMu.Lock()
streamToolNames = make(map[string]string)
streamReasoningTitles = make(map[string]string)
streamReasoningText = make(map[string]string)
stepToolResults = nil
stepAssistantDraft = nil
stepToolCallIndexByID = make(map[string]int)
stepStateMu.Unlock()
}
setReasoningTitleFromText := func(id string, text string) {
if id == "" || strings.TrimSpace(text) == "" {
return
}
stepStateMu.Lock()
defer stepStateMu.Unlock()
if streamReasoningTitles[id] != "" {
return
}
streamReasoningText[id] += text
if !strings.ContainsAny(streamReasoningText[id], "\r\n") {
return
}
title := chatprompt.ReasoningTitleFromFirstLine(streamReasoningText[id])
if title == "" {
return
}
streamReasoningTitles[id] = title
}
appendDraftText := func(text string) {
if text == "" {
return
}
stepStateMu.Lock()
defer stepStateMu.Unlock()
if len(stepAssistantDraft) > 0 {
lastIndex := len(stepAssistantDraft) - 1
switch last := stepAssistantDraft[lastIndex].(type) {
case fantasy.TextContent:
last.Text += text
stepAssistantDraft[lastIndex] = last
return
case *fantasy.TextContent:
last.Text += text
stepAssistantDraft[lastIndex] = fantasy.TextContent{Text: last.Text}
return
}
}
stepAssistantDraft = append(stepAssistantDraft, fantasy.TextContent{Text: text})
}
appendDraftReasoning := func(text string) {
if text == "" {
return
}
stepStateMu.Lock()
defer stepStateMu.Unlock()
if len(stepAssistantDraft) > 0 {
lastIndex := len(stepAssistantDraft) - 1
switch last := stepAssistantDraft[lastIndex].(type) {
case fantasy.ReasoningContent:
last.Text += text
stepAssistantDraft[lastIndex] = last
return
case *fantasy.ReasoningContent:
last.Text += text
stepAssistantDraft[lastIndex] = fantasy.ReasoningContent{Text: last.Text}
return
}
}
stepAssistantDraft = append(stepAssistantDraft, fantasy.ReasoningContent{Text: text})
}
upsertDraftToolCall := func(toolCallID, toolName, input string, appendInput bool) {
if toolCallID == "" {
return
}
stepStateMu.Lock()
defer stepStateMu.Unlock()
if strings.TrimSpace(toolName) != "" {
streamToolNames[toolCallID] = toolName
}
index, exists := stepToolCallIndexByID[toolCallID]
if !exists {
stepToolCallIndexByID[toolCallID] = len(stepAssistantDraft)
stepAssistantDraft = append(stepAssistantDraft, fantasy.ToolCallContent{
ToolCallID: toolCallID,
ToolName: toolName,
Input: input,
})
return
}
if index < 0 || index >= len(stepAssistantDraft) {
stepToolCallIndexByID[toolCallID] = len(stepAssistantDraft)
stepAssistantDraft = append(stepAssistantDraft, fantasy.ToolCallContent{
ToolCallID: toolCallID,
ToolName: toolName,
Input: input,
})
return
}
existingCall, ok := fantasy.AsContentType[fantasy.ToolCallContent](stepAssistantDraft[index])
if !ok {
if ptrCall, ptrOK := fantasy.AsContentType[*fantasy.ToolCallContent](stepAssistantDraft[index]); ptrOK && ptrCall != nil {
existingCall = *ptrCall
ok = true
}
}
if !ok {
stepToolCallIndexByID[toolCallID] = len(stepAssistantDraft)
stepAssistantDraft = append(stepAssistantDraft, fantasy.ToolCallContent{
ToolCallID: toolCallID,
ToolName: toolName,
Input: input,
})
return
}
if strings.TrimSpace(toolName) != "" {
existingCall.ToolName = toolName
}
if appendInput {
existingCall.Input += input
} else if input != "" || existingCall.Input == "" {
existingCall.Input = input
}
stepAssistantDraft[index] = existingCall
}
appendDraftSource := func(source fantasy.SourceContent) {
stepStateMu.Lock()
stepAssistantDraft = append(stepAssistantDraft, source)
stepStateMu.Unlock()
}
persistInterruptedStep := func() error {
stepStateMu.Lock()
draft := append([]fantasy.Content(nil), stepAssistantDraft...)
toolResults := append([]chatprompt.ToolResultBlock(nil), stepToolResults...)
toolNameByCallID := make(map[string]string, len(streamToolNames))
for id, name := range streamToolNames {
toolNameByCallID[id] = name
}
stepStateMu.Unlock()
if len(draft) == 0 && len(toolResults) == 0 {
return nil
}
resultByToolCallID := make(map[string]struct{}, len(toolResults))
for _, toolResult := range toolResults {
if toolResult.ToolCallID == "" {
continue
}
resultByToolCallID[toolResult.ToolCallID] = struct{}{}
}
missingResults := make([]chatprompt.ToolResultBlock, 0)
for _, block := range draft {
toolCall, ok := fantasy.AsContentType[fantasy.ToolCallContent](block)
if !ok {
if ptrCall, ptrOK := fantasy.AsContentType[*fantasy.ToolCallContent](block); ptrOK && ptrCall != nil {
toolCall = *ptrCall
ok = true
}
}
if !ok || toolCall.ToolCallID == "" {
continue
}
if _, exists := resultByToolCallID[toolCall.ToolCallID]; exists {
continue
}
toolName := strings.TrimSpace(toolCall.ToolName)
if toolName == "" {
toolName = strings.TrimSpace(toolNameByCallID[toolCall.ToolCallID])
}
missingResults = append(missingResults, chatprompt.ToolResultBlock{
ToolCallID: toolCall.ToolCallID,
ToolName: toolName,
Result: map[string]any{
"status": "interrupted",
"error": interruptedToolResultErrorMessage,
},
IsError: true,
})
resultByToolCallID[toolCall.ToolCallID] = struct{}{}
}
toolResults = append(toolResults, missingResults...)
persistCtx := context.WithoutCancel(ctx)
if err := persistAssistant(
persistCtx,
stepAssistantContent(draft, toolResults),
fantasy.Usage{},
sql.NullInt64{},
); err != nil {
return err
}
for _, toolResult := range toolResults {
if err := persistToolResult(persistCtx, toolResult); err != nil {
return err
}
}
return nil
}
resetStepState()
agent := fantasy.NewAgent(
model,
fantasy.WithTools(p.agentTools(model, &currentChat, &chatStateMu, getWorkspaceConn)...),
fantasy.WithStopConditions(fantasy.StepCountIs(maxChatSteps)),
)
sentinelPrompt := "__chatd_agent_prompt_sentinel_" + uuid.NewString()
applyAnthropicCaching := shouldApplyAnthropicPromptCaching(model)
streamCall := streamCallOptionsFromChatModelConfig(model, modelConfig)
streamCall.Prompt = sentinelPrompt
streamCall.Messages = prompt
streamCall.PrepareStep = func(
stepCtx context.Context,
options fantasy.PrepareStepFunctionOptions,
) (context.Context, fantasy.PrepareStepResult, error) {
return stepCtx, prepareAgentStepResult(
options.Messages,
sentinelPrompt,
reportOnly,
applyAnthropicCaching,
), nil
}
streamCall.OnStepStart = func(_ int) error {
resetStepState()
return nil
}
streamCall.OnTextDelta = func(_ string, text string) error {
appendDraftText(text)
p.publishMessagePart(chat.ID, string(fantasy.MessageRoleAssistant), codersdk.ChatMessagePart{
Type: codersdk.ChatMessagePartTypeText,
Text: text,
})
return nil
}
streamCall.OnReasoningDelta = func(id string, text string) error {
appendDraftReasoning(text)
setReasoningTitleFromText(id, text)
stepStateMu.Lock()
title := streamReasoningTitles[id]
stepStateMu.Unlock()
p.publishMessagePart(chat.ID, string(fantasy.MessageRoleAssistant), codersdk.ChatMessagePart{
Type: codersdk.ChatMessagePartTypeReasoning,
Text: text,
Title: title,
})
return nil
}
streamCall.OnReasoningStart = func(id string, reasoning fantasy.ReasoningContent) error {
_ = id
_ = reasoning
return nil
}
streamCall.OnReasoningEnd = func(id string, reasoning fantasy.ReasoningContent) error {
_ = reasoning
stepStateMu.Lock()
if streamReasoningTitles[id] == "" {
// At the end of reasoning we have the full text, so we can
// safely evaluate first-line title format even if no newline
// ever arrived in deltas.
streamReasoningTitles[id] = chatprompt.ReasoningTitleFromFirstLine(
streamReasoningText[id],
)
}
title := streamReasoningTitles[id]
stepStateMu.Unlock()
if title != "" {
// Publish a title-only reasoning part so clients can update the
// reasoning header when metadata arrives at the end of streaming.
p.publishMessagePart(chat.ID, string(fantasy.MessageRoleAssistant), codersdk.ChatMessagePart{
Type: codersdk.ChatMessagePartTypeReasoning,
Title: title,
})
}
return nil
}
streamCall.OnToolInputStart = func(id, toolName string) error {
upsertDraftToolCall(id, toolName, "", false)
return nil
}
streamCall.OnToolInputDelta = func(id, delta string) error {
stepStateMu.Lock()
toolName := streamToolNames[id]
stepStateMu.Unlock()
upsertDraftToolCall(id, toolName, delta, true)
p.publishMessagePart(chat.ID, string(fantasy.MessageRoleAssistant), codersdk.ChatMessagePart{
Type: codersdk.ChatMessagePartTypeToolCall,
ToolCallID: id,
ToolName: toolName,
ArgsDelta: delta,
})
return nil
}
streamCall.OnToolCall = func(toolCall fantasy.ToolCallContent) error {
upsertDraftToolCall(toolCall.ToolCallID, toolCall.ToolName, toolCall.Input, false)
p.publishMessagePart(
chat.ID,
string(fantasy.MessageRoleAssistant),
chatprompt.PartFromContent(toolCall),
)
return nil
}
streamCall.OnSource = func(source fantasy.SourceContent) error {
appendDraftSource(source)
p.publishMessagePart(
chat.ID,
string(fantasy.MessageRoleAssistant),
chatprompt.PartFromContent(source),
)
return nil
}
streamCall.OnToolResult = func(result fantasy.ToolResultContent) error {
toolResult := chatprompt.ResultFromContent(result)
if strings.TrimSpace(result.ClientMetadata) != "" {
var block chatprompt.ToolResultBlock
if err := json.Unmarshal([]byte(result.ClientMetadata), &block); err == nil {
if block.ToolCallID == "" {
block.ToolCallID = result.ToolCallID
}
if block.ToolName == "" {
block.ToolName = result.ToolName
}
toolResult = block
}
}
p.publishMessagePart(chat.ID, string(fantasy.MessageRoleTool), chatprompt.PartFromResult(toolResult))
stepStateMu.Lock()
if toolResult.ToolCallID != "" && strings.TrimSpace(toolResult.ToolName) != "" {
streamToolNames[toolResult.ToolCallID] = toolResult.ToolName
}
stepToolResults = append(stepToolResults, toolResult)
stepStateMu.Unlock()
return chatloop.Run(ctx, chatloop.RunOptions{
Model: model,
Messages: prompt,
Tools: p.agentTools(model, &currentChat, &chatStateMu, getWorkspaceConn),
StreamCall: streamCall,
MaxSteps: maxChatSteps,
return nil
}
streamCall.OnStepFinish = func(stepResult fantasy.StepResult) error {
stepStateMu.Lock()
toolResults := append([]chatprompt.ToolResultBlock(nil), stepToolResults...)
stepStateMu.Unlock()
ReportOnly: reportOnly,
ContextLimitFallback: modelConfigContextLimit,
contextLimit := extractContextLimit(stepResult.ProviderMetadata)
if !contextLimit.Valid && modelConfigContextLimit > 0 {
contextLimit = sql.NullInt64{
Int64: modelConfigContextLimit,
Valid: true,
}
}
if err := persistAssistant(
ctx,
stepAssistantContent(stepResult.Content, toolResults),
stepResult.Usage,
contextLimit,
); err != nil {
return err
}
for _, toolResult := range toolResults {
if err := persistToolResult(ctx, toolResult); err != nil {
return err
}
}
return nil
}
result, err := agent.Stream(ctx, streamCall)
if err != nil {
if errors.Is(err, context.Canceled) && errors.Is(context.Cause(ctx), ErrChatInterrupted) {
if persistErr := persistInterruptedStep(); persistErr != nil {
p.logger.Warn(ctx, "failed to persist interrupted chat step", slog.Error(persistErr))
}
return nil, ErrChatInterrupted
}
return nil, xerrors.Errorf("stream response: %w", err)
}
return result, nil
}
func prepareAgentStepResult(
messages []fantasy.Message,
sentinel string,
reportOnly bool,
anthropicCaching bool,
) fantasy.PrepareStepResult {
filtered := make([]fantasy.Message, 0, len(messages))
removed := false
for _, message := range messages {
if !removed &&
message.Role == fantasy.MessageRoleUser &&
len(message.Content) == 1 {
textPart, ok := fantasy.AsMessagePart[fantasy.TextPart](message.Content[0])
if ok && textPart.Text == sentinel {
removed = true
continue
}
}
filtered = append(filtered, message)
}
result := fantasy.PrepareStepResult{
Messages: filtered,
}
if anthropicCaching {
result.Messages = addAnthropicPromptCaching(result.Messages)
}
if reportOnly {
result.ActiveTools = []string{"subagent_report"}
}
return result
}
func shouldApplyAnthropicPromptCaching(model fantasy.LanguageModel) bool {
if model == nil {
return false
}
return model.Provider() == fantasyanthropic.Name
}
func addAnthropicPromptCaching(messages []fantasy.Message) []fantasy.Message {
for i := range messages {
messages[i].ProviderOptions = nil
}
providerOption := fantasy.ProviderOptions{
fantasyanthropic.Name: &fantasyanthropic.ProviderCacheControlOptions{
CacheControl: fantasyanthropic.CacheControl{Type: "ephemeral"},
PersistAssistant: persistAssistant,
PersistToolResult: persistToolResult,
PublishMessagePart: func(
role fantasy.MessageRole,
part codersdk.ChatMessagePart,
) {
p.publishMessagePart(chat.ID, string(role), part)
},
}
lastSystemRoleIdx := -1
systemMessageUpdated := false
for i, msg := range messages {
if msg.Role == fantasy.MessageRoleSystem {
lastSystemRoleIdx = i
} else if !systemMessageUpdated && lastSystemRoleIdx >= 0 {
messages[lastSystemRoleIdx].ProviderOptions = providerOption
systemMessageUpdated = true
}
if i > len(messages)-3 {
messages[i].ProviderOptions = providerOption
}
}
return messages
}
func stepAssistantContent(content []fantasy.Content, toolResults []chatprompt.ToolResultBlock) []fantasy.Content {
toolResultIDs := make(map[string]struct{}, len(toolResults))
for _, toolResult := range toolResults {
if toolResult.ToolCallID == "" {
continue
}
toolResultIDs[toolResult.ToolCallID] = struct{}{}
}
filtered := make([]fantasy.Content, 0, len(content))
for _, block := range content {
toolResult, ok := fantasy.AsContentType[fantasy.ToolResultContent](block)
if !ok {
filtered = append(filtered, block)
continue
}
if _, tracked := toolResultIDs[toolResult.ToolCallID]; tracked {
continue
}
filtered = append(filtered, block)
}
return filtered
InterruptedError: ErrChatInterrupted,
OnInterruptedPersistError: func(err error) {
p.logger.Warn(ctx, "failed to persist interrupted chat step", slog.Error(err))
},
})
}
func usageNullInt64(value int64, valid bool) sql.NullInt64 {
@@ -2107,145 +1636,6 @@ func usageNullInt64(value int64, valid bool) sql.NullInt64 {
}
}
func extractContextLimit(metadata fantasy.ProviderMetadata) sql.NullInt64 {
if len(metadata) == 0 {
return sql.NullInt64{}
}
encoded, err := json.Marshal(metadata)
if err != nil || len(encoded) == 0 {
return sql.NullInt64{}
}
var payload any
if err := json.Unmarshal(encoded, &payload); err != nil {
return sql.NullInt64{}
}
limit, ok := findContextLimitValue(payload)
if !ok {
return sql.NullInt64{}
}
return sql.NullInt64{
Int64: limit,
Valid: true,
}
}
func findContextLimitValue(value any) (int64, bool) {
var (
limit int64
found bool
)
collectContextLimitValues(value, func(candidate int64) {
if !found || candidate > limit {
limit = candidate
found = true
}
})
return limit, found
}
func collectContextLimitValues(value any, onValue func(int64)) {
switch typed := value.(type) {
case map[string]any:
for key, child := range typed {
if isContextLimitKey(key) {
if numeric, ok := numericContextLimitValue(child); ok {
onValue(numeric)
}
}
collectContextLimitValues(child, onValue)
}
case []any:
for _, child := range typed {
collectContextLimitValues(child, onValue)
}
}
}
func isContextLimitKey(key string) bool {
normalized := normalizeMetadataKey(key)
if normalized == "" {
return false
}
switch normalized {
case
"contextlimit",
"contextwindow",
"contextlength",
"maxcontext",
"maxcontexttokens",
"maxinputtokens",
"maxinputtoken",
"inputtokenlimit":
return true
}
return strings.Contains(normalized, "context") &&
(strings.Contains(normalized, "limit") ||
strings.Contains(normalized, "window") ||
strings.Contains(normalized, "length") ||
strings.HasPrefix(normalized, "max"))
}
func normalizeMetadataKey(key string) string {
var b strings.Builder
b.Grow(len(key))
for _, r := range key {
switch {
case r >= 'a' && r <= 'z':
b.WriteRune(r)
case r >= 'A' && r <= 'Z':
b.WriteRune(r + ('a' - 'A'))
case r >= '0' && r <= '9':
b.WriteRune(r)
}
}
return b.String()
}
func numericContextLimitValue(value any) (int64, bool) {
switch typed := value.(type) {
case int64:
return positiveInt64(typed)
case int32:
return positiveInt64(int64(typed))
case int:
return positiveInt64(int64(typed))
case float64:
casted := int64(typed)
if typed > 0 && float64(casted) == typed {
return casted, true
}
case string:
parsed, err := strconv.ParseInt(strings.TrimSpace(typed), 10, 64)
if err == nil {
return positiveInt64(parsed)
}
case json.Number:
parsed, err := typed.Int64()
if err == nil {
return positiveInt64(parsed)
}
}
return 0, false
}
func positiveInt64(value int64) (int64, bool) {
if value <= 0 {
return 0, false
}
return value, true
}
func (p *Processor) maybeGenerateChatTitle(
ctx context.Context,
chat database.Chat,
-77
View File
@@ -7,7 +7,6 @@ import (
"time"
"charm.land/fantasy"
fantasyanthropic "charm.land/fantasy/providers/anthropic"
fantasyopenai "charm.land/fantasy/providers/openai"
"github.com/google/uuid"
@@ -457,59 +456,6 @@ func TestChatMessagesToPrompt_RepairsLegacyOrphanToolResult(t *testing.T) {
require.Equal(t, sanitizedToolCallID, toolResultParts[0].ToolCallID)
}
func TestPrepareAgentStepResult_ReportOnly(t *testing.T) {
t.Parallel()
sentinel := "__sentinel__"
result := prepareAgentStepResult(
[]fantasy.Message{
{
Role: fantasy.MessageRoleUser,
Content: []fantasy.MessagePart{
fantasy.TextPart{Text: sentinel},
},
},
{
Role: fantasy.MessageRoleUser,
Content: []fantasy.MessagePart{
fantasy.TextPart{Text: "real message"},
},
},
},
sentinel,
true,
false,
)
require.Equal(t, []string{"subagent_report"}, result.ActiveTools)
require.Len(t, result.Messages, 1)
require.Equal(t, fantasy.MessageRoleUser, result.Messages[0].Role)
}
func TestPrepareAgentStepResult_AnthropicCaching(t *testing.T) {
t.Parallel()
result := prepareAgentStepResult(
[]fantasy.Message{
textMessage(fantasy.MessageRoleSystem, "sys-1"),
textMessage(fantasy.MessageRoleSystem, "sys-2"),
textMessage(fantasy.MessageRoleUser, "hello"),
textMessage(fantasy.MessageRoleAssistant, "working"),
textMessage(fantasy.MessageRoleUser, "continue"),
},
"__sentinel__",
false,
true,
)
require.Len(t, result.Messages, 5)
require.False(t, hasAnthropicEphemeralCacheControl(result.Messages[0]))
require.True(t, hasAnthropicEphemeralCacheControl(result.Messages[1]))
require.False(t, hasAnthropicEphemeralCacheControl(result.Messages[2]))
require.True(t, hasAnthropicEphemeralCacheControl(result.Messages[3]))
require.True(t, hasAnthropicEphemeralCacheControl(result.Messages[4]))
}
func TestParseChatModelConfig_ParsesCallConfig(t *testing.T) {
t.Parallel()
@@ -1076,29 +1022,6 @@ func messageToolResultParts(message fantasy.Message) []fantasy.ToolResultPart {
return results
}
func textMessage(role fantasy.MessageRole, text string) fantasy.Message {
return fantasy.Message{
Role: role,
Content: []fantasy.MessagePart{
fantasy.TextPart{Text: text},
},
}
}
func hasAnthropicEphemeralCacheControl(message fantasy.Message) bool {
if len(message.ProviderOptions) == 0 {
return false
}
options, ok := message.ProviderOptions[fantasyanthropic.Name]
if !ok {
return false
}
cacheOptions, ok := options.(*fantasyanthropic.ProviderCacheControlOptions)
return ok && cacheOptions.CacheControl.Type == "ephemeral"
}
type titleTestModel struct {
generateFn func(context.Context, fantasy.Call) (*fantasy.Response, error)
provider string
+714
View File
@@ -0,0 +1,714 @@
package chatloop
import (
"context"
"database/sql"
"encoding/json"
"errors"
"strconv"
"strings"
"sync"
"charm.land/fantasy"
fantasyanthropic "charm.land/fantasy/providers/anthropic"
"github.com/google/uuid"
"golang.org/x/xerrors"
"github.com/coder/coder/v2/coderd/chatd/chatprompt"
"github.com/coder/coder/v2/codersdk"
)
const (
interruptedToolResultErrorMessage = "tool call was interrupted before it produced a result"
reportOnlyToolName = "subagent_report"
)
// RunOptions configures a single streaming chat loop run.
type RunOptions struct {
Model fantasy.LanguageModel
Messages []fantasy.Message
Tools []fantasy.AgentTool
StreamCall fantasy.AgentStreamCall
MaxSteps int
ReportOnly bool
ContextLimitFallback int64
PersistAssistant func(
persistCtx context.Context,
content []fantasy.Content,
usage fantasy.Usage,
contextLimit sql.NullInt64,
) error
PersistToolResult func(context.Context, chatprompt.ToolResultBlock) error
PublishMessagePart func(
role fantasy.MessageRole,
part codersdk.ChatMessagePart,
)
InterruptedError error
OnInterruptedPersistError func(error)
}
// Run executes the chat step-stream loop and delegates persistence/publishing to callbacks.
func Run(ctx context.Context, opts RunOptions) (*fantasy.AgentResult, error) {
if opts.Model == nil {
return nil, xerrors.New("chat model is required")
}
if opts.PersistAssistant == nil {
return nil, xerrors.New("persist assistant callback is required")
}
if opts.PersistToolResult == nil {
return nil, xerrors.New("persist tool result callback is required")
}
if opts.MaxSteps <= 0 {
opts.MaxSteps = 1
}
publishMessagePart := func(role fantasy.MessageRole, part codersdk.ChatMessagePart) {
if opts.PublishMessagePart == nil {
return
}
opts.PublishMessagePart(role, part)
}
var (
stepStateMu sync.Mutex
streamToolNames map[string]string
streamReasoningTitles map[string]string
streamReasoningText map[string]string
stepToolResults []chatprompt.ToolResultBlock
stepAssistantDraft []fantasy.Content
stepToolCallIndexByID map[string]int
)
resetStepState := func() {
stepStateMu.Lock()
streamToolNames = make(map[string]string)
streamReasoningTitles = make(map[string]string)
streamReasoningText = make(map[string]string)
stepToolResults = nil
stepAssistantDraft = nil
stepToolCallIndexByID = make(map[string]int)
stepStateMu.Unlock()
}
setReasoningTitleFromText := func(id string, text string) {
if id == "" || strings.TrimSpace(text) == "" {
return
}
stepStateMu.Lock()
defer stepStateMu.Unlock()
if streamReasoningTitles[id] != "" {
return
}
streamReasoningText[id] += text
if !strings.ContainsAny(streamReasoningText[id], "\r\n") {
return
}
title := chatprompt.ReasoningTitleFromFirstLine(streamReasoningText[id])
if title == "" {
return
}
streamReasoningTitles[id] = title
}
appendDraftText := func(text string) {
if text == "" {
return
}
stepStateMu.Lock()
defer stepStateMu.Unlock()
if len(stepAssistantDraft) > 0 {
lastIndex := len(stepAssistantDraft) - 1
switch last := stepAssistantDraft[lastIndex].(type) {
case fantasy.TextContent:
last.Text += text
stepAssistantDraft[lastIndex] = last
return
case *fantasy.TextContent:
last.Text += text
stepAssistantDraft[lastIndex] = fantasy.TextContent{Text: last.Text}
return
}
}
stepAssistantDraft = append(stepAssistantDraft, fantasy.TextContent{Text: text})
}
appendDraftReasoning := func(text string) {
if text == "" {
return
}
stepStateMu.Lock()
defer stepStateMu.Unlock()
if len(stepAssistantDraft) > 0 {
lastIndex := len(stepAssistantDraft) - 1
switch last := stepAssistantDraft[lastIndex].(type) {
case fantasy.ReasoningContent:
last.Text += text
stepAssistantDraft[lastIndex] = last
return
case *fantasy.ReasoningContent:
last.Text += text
stepAssistantDraft[lastIndex] = fantasy.ReasoningContent{Text: last.Text}
return
}
}
stepAssistantDraft = append(stepAssistantDraft, fantasy.ReasoningContent{Text: text})
}
upsertDraftToolCall := func(toolCallID, toolName, input string, appendInput bool) {
if toolCallID == "" {
return
}
stepStateMu.Lock()
defer stepStateMu.Unlock()
if strings.TrimSpace(toolName) != "" {
streamToolNames[toolCallID] = toolName
}
index, exists := stepToolCallIndexByID[toolCallID]
if !exists {
stepToolCallIndexByID[toolCallID] = len(stepAssistantDraft)
stepAssistantDraft = append(stepAssistantDraft, fantasy.ToolCallContent{
ToolCallID: toolCallID,
ToolName: toolName,
Input: input,
})
return
}
if index < 0 || index >= len(stepAssistantDraft) {
stepToolCallIndexByID[toolCallID] = len(stepAssistantDraft)
stepAssistantDraft = append(stepAssistantDraft, fantasy.ToolCallContent{
ToolCallID: toolCallID,
ToolName: toolName,
Input: input,
})
return
}
existingCall, ok := fantasy.AsContentType[fantasy.ToolCallContent](stepAssistantDraft[index])
if !ok {
if ptrCall, ptrOK := fantasy.AsContentType[*fantasy.ToolCallContent](stepAssistantDraft[index]); ptrOK && ptrCall != nil {
existingCall = *ptrCall
ok = true
}
}
if !ok {
stepToolCallIndexByID[toolCallID] = len(stepAssistantDraft)
stepAssistantDraft = append(stepAssistantDraft, fantasy.ToolCallContent{
ToolCallID: toolCallID,
ToolName: toolName,
Input: input,
})
return
}
if strings.TrimSpace(toolName) != "" {
existingCall.ToolName = toolName
}
if appendInput {
existingCall.Input += input
} else if input != "" || existingCall.Input == "" {
existingCall.Input = input
}
stepAssistantDraft[index] = existingCall
}
appendDraftSource := func(source fantasy.SourceContent) {
stepStateMu.Lock()
stepAssistantDraft = append(stepAssistantDraft, source)
stepStateMu.Unlock()
}
persistInterruptedStep := func() error {
stepStateMu.Lock()
draft := append([]fantasy.Content(nil), stepAssistantDraft...)
toolResults := append([]chatprompt.ToolResultBlock(nil), stepToolResults...)
toolNameByCallID := make(map[string]string, len(streamToolNames))
for id, name := range streamToolNames {
toolNameByCallID[id] = name
}
stepStateMu.Unlock()
if len(draft) == 0 && len(toolResults) == 0 {
return nil
}
resultByToolCallID := make(map[string]struct{}, len(toolResults))
for _, toolResult := range toolResults {
if toolResult.ToolCallID == "" {
continue
}
resultByToolCallID[toolResult.ToolCallID] = struct{}{}
}
missingResults := make([]chatprompt.ToolResultBlock, 0)
for _, block := range draft {
toolCall, ok := fantasy.AsContentType[fantasy.ToolCallContent](block)
if !ok {
if ptrCall, ptrOK := fantasy.AsContentType[*fantasy.ToolCallContent](block); ptrOK && ptrCall != nil {
toolCall = *ptrCall
ok = true
}
}
if !ok || toolCall.ToolCallID == "" {
continue
}
if _, exists := resultByToolCallID[toolCall.ToolCallID]; exists {
continue
}
toolName := strings.TrimSpace(toolCall.ToolName)
if toolName == "" {
toolName = strings.TrimSpace(toolNameByCallID[toolCall.ToolCallID])
}
missingResults = append(missingResults, chatprompt.ToolResultBlock{
ToolCallID: toolCall.ToolCallID,
ToolName: toolName,
Result: map[string]any{
"status": "interrupted",
"error": interruptedToolResultErrorMessage,
},
IsError: true,
})
resultByToolCallID[toolCall.ToolCallID] = struct{}{}
}
toolResults = append(toolResults, missingResults...)
persistCtx := context.WithoutCancel(ctx)
if err := opts.PersistAssistant(
persistCtx,
stepAssistantContent(draft, toolResults),
fantasy.Usage{},
sql.NullInt64{},
); err != nil {
return err
}
for _, toolResult := range toolResults {
if err := opts.PersistToolResult(persistCtx, toolResult); err != nil {
return err
}
}
return nil
}
resetStepState()
agent := fantasy.NewAgent(
opts.Model,
fantasy.WithTools(opts.Tools...),
fantasy.WithStopConditions(fantasy.StepCountIs(opts.MaxSteps)),
)
applyAnthropicCaching := shouldApplyAnthropicPromptCaching(opts.Model)
sentinelPrompt := "__chatd_agent_prompt_sentinel_" + uuid.NewString()
streamCall := opts.StreamCall
streamCall.Prompt = sentinelPrompt
streamCall.Messages = opts.Messages
streamCall.PrepareStep = func(
stepCtx context.Context,
options fantasy.PrepareStepFunctionOptions,
) (context.Context, fantasy.PrepareStepResult, error) {
return stepCtx, prepareStepResult(
options.Messages,
sentinelPrompt,
opts.ReportOnly,
applyAnthropicCaching,
), nil
}
streamCall.OnStepStart = func(_ int) error {
resetStepState()
return nil
}
streamCall.OnTextDelta = func(_ string, text string) error {
appendDraftText(text)
publishMessagePart(fantasy.MessageRoleAssistant, codersdk.ChatMessagePart{
Type: codersdk.ChatMessagePartTypeText,
Text: text,
})
return nil
}
streamCall.OnReasoningDelta = func(id string, text string) error {
appendDraftReasoning(text)
setReasoningTitleFromText(id, text)
stepStateMu.Lock()
title := streamReasoningTitles[id]
stepStateMu.Unlock()
publishMessagePart(fantasy.MessageRoleAssistant, codersdk.ChatMessagePart{
Type: codersdk.ChatMessagePartTypeReasoning,
Text: text,
Title: title,
})
return nil
}
streamCall.OnReasoningEnd = func(id string, _ fantasy.ReasoningContent) error {
stepStateMu.Lock()
if streamReasoningTitles[id] == "" {
// At the end of reasoning we have the full text, so we can
// safely evaluate first-line title format even if no newline
// ever arrived in deltas.
streamReasoningTitles[id] = chatprompt.ReasoningTitleFromFirstLine(
streamReasoningText[id],
)
}
title := streamReasoningTitles[id]
stepStateMu.Unlock()
if title != "" {
// Publish a title-only reasoning part so clients can update the
// reasoning header when metadata arrives at the end of streaming.
publishMessagePart(fantasy.MessageRoleAssistant, codersdk.ChatMessagePart{
Type: codersdk.ChatMessagePartTypeReasoning,
Title: title,
})
}
return nil
}
streamCall.OnToolInputStart = func(id, toolName string) error {
upsertDraftToolCall(id, toolName, "", false)
return nil
}
streamCall.OnToolInputDelta = func(id, delta string) error {
stepStateMu.Lock()
toolName := streamToolNames[id]
stepStateMu.Unlock()
upsertDraftToolCall(id, toolName, delta, true)
publishMessagePart(fantasy.MessageRoleAssistant, codersdk.ChatMessagePart{
Type: codersdk.ChatMessagePartTypeToolCall,
ToolCallID: id,
ToolName: toolName,
ArgsDelta: delta,
})
return nil
}
streamCall.OnToolCall = func(toolCall fantasy.ToolCallContent) error {
upsertDraftToolCall(toolCall.ToolCallID, toolCall.ToolName, toolCall.Input, false)
publishMessagePart(
fantasy.MessageRoleAssistant,
chatprompt.PartFromContent(toolCall),
)
return nil
}
streamCall.OnSource = func(source fantasy.SourceContent) error {
appendDraftSource(source)
publishMessagePart(
fantasy.MessageRoleAssistant,
chatprompt.PartFromContent(source),
)
return nil
}
streamCall.OnToolResult = func(result fantasy.ToolResultContent) error {
toolResult := chatprompt.ResultFromContent(result)
if strings.TrimSpace(result.ClientMetadata) != "" {
var block chatprompt.ToolResultBlock
if err := json.Unmarshal([]byte(result.ClientMetadata), &block); err == nil {
if block.ToolCallID == "" {
block.ToolCallID = result.ToolCallID
}
if block.ToolName == "" {
block.ToolName = result.ToolName
}
toolResult = block
}
}
publishMessagePart(fantasy.MessageRoleTool, chatprompt.PartFromResult(toolResult))
stepStateMu.Lock()
if toolResult.ToolCallID != "" && strings.TrimSpace(toolResult.ToolName) != "" {
streamToolNames[toolResult.ToolCallID] = toolResult.ToolName
}
stepToolResults = append(stepToolResults, toolResult)
stepStateMu.Unlock()
return nil
}
streamCall.OnStepFinish = func(stepResult fantasy.StepResult) error {
stepStateMu.Lock()
toolResults := append([]chatprompt.ToolResultBlock(nil), stepToolResults...)
stepStateMu.Unlock()
contextLimit := extractContextLimit(stepResult.ProviderMetadata)
if !contextLimit.Valid && opts.ContextLimitFallback > 0 {
contextLimit = sql.NullInt64{
Int64: opts.ContextLimitFallback,
Valid: true,
}
}
if err := opts.PersistAssistant(
ctx,
stepAssistantContent(stepResult.Content, toolResults),
stepResult.Usage,
contextLimit,
); err != nil {
return err
}
for _, toolResult := range toolResults {
if err := opts.PersistToolResult(ctx, toolResult); err != nil {
return err
}
}
return nil
}
result, err := agent.Stream(ctx, streamCall)
if err != nil {
if opts.InterruptedError != nil &&
errors.Is(err, context.Canceled) &&
errors.Is(context.Cause(ctx), opts.InterruptedError) {
if persistErr := persistInterruptedStep(); persistErr != nil {
if opts.OnInterruptedPersistError != nil {
opts.OnInterruptedPersistError(persistErr)
}
}
return nil, opts.InterruptedError
}
return nil, xerrors.Errorf("stream response: %w", err)
}
return result, nil
}
func prepareStepResult(
messages []fantasy.Message,
sentinel string,
reportOnly bool,
anthropicCaching bool,
) fantasy.PrepareStepResult {
filtered := make([]fantasy.Message, 0, len(messages))
removed := false
for _, message := range messages {
if !removed &&
message.Role == fantasy.MessageRoleUser &&
len(message.Content) == 1 {
textPart, ok := fantasy.AsMessagePart[fantasy.TextPart](message.Content[0])
if ok && textPart.Text == sentinel {
removed = true
continue
}
}
filtered = append(filtered, message)
}
result := fantasy.PrepareStepResult{
Messages: filtered,
}
if anthropicCaching {
result.Messages = addAnthropicPromptCaching(result.Messages)
}
if reportOnly {
result.ActiveTools = []string{reportOnlyToolName}
}
return result
}
func shouldApplyAnthropicPromptCaching(model fantasy.LanguageModel) bool {
if model == nil {
return false
}
return model.Provider() == fantasyanthropic.Name
}
func addAnthropicPromptCaching(messages []fantasy.Message) []fantasy.Message {
for i := range messages {
messages[i].ProviderOptions = nil
}
providerOption := fantasy.ProviderOptions{
fantasyanthropic.Name: &fantasyanthropic.ProviderCacheControlOptions{
CacheControl: fantasyanthropic.CacheControl{Type: "ephemeral"},
},
}
lastSystemRoleIdx := -1
systemMessageUpdated := false
for i, msg := range messages {
if msg.Role == fantasy.MessageRoleSystem {
lastSystemRoleIdx = i
} else if !systemMessageUpdated && lastSystemRoleIdx >= 0 {
messages[lastSystemRoleIdx].ProviderOptions = providerOption
systemMessageUpdated = true
}
if i > len(messages)-3 {
messages[i].ProviderOptions = providerOption
}
}
return messages
}
func stepAssistantContent(content []fantasy.Content, toolResults []chatprompt.ToolResultBlock) []fantasy.Content {
toolResultIDs := make(map[string]struct{}, len(toolResults))
for _, toolResult := range toolResults {
if toolResult.ToolCallID == "" {
continue
}
toolResultIDs[toolResult.ToolCallID] = struct{}{}
}
filtered := make([]fantasy.Content, 0, len(content))
for _, block := range content {
toolResult, ok := fantasy.AsContentType[fantasy.ToolResultContent](block)
if !ok {
filtered = append(filtered, block)
continue
}
if _, tracked := toolResultIDs[toolResult.ToolCallID]; tracked {
continue
}
filtered = append(filtered, block)
}
return filtered
}
func extractContextLimit(metadata fantasy.ProviderMetadata) sql.NullInt64 {
if len(metadata) == 0 {
return sql.NullInt64{}
}
encoded, err := json.Marshal(metadata)
if err != nil || len(encoded) == 0 {
return sql.NullInt64{}
}
var payload any
if err := json.Unmarshal(encoded, &payload); err != nil {
return sql.NullInt64{}
}
limit, ok := findContextLimitValue(payload)
if !ok {
return sql.NullInt64{}
}
return sql.NullInt64{
Int64: limit,
Valid: true,
}
}
func findContextLimitValue(value any) (int64, bool) {
var (
limit int64
found bool
)
collectContextLimitValues(value, func(candidate int64) {
if !found || candidate > limit {
limit = candidate
found = true
}
})
return limit, found
}
func collectContextLimitValues(value any, onValue func(int64)) {
switch typed := value.(type) {
case map[string]any:
for key, child := range typed {
if isContextLimitKey(key) {
if numeric, ok := numericContextLimitValue(child); ok {
onValue(numeric)
}
}
collectContextLimitValues(child, onValue)
}
case []any:
for _, child := range typed {
collectContextLimitValues(child, onValue)
}
}
}
func isContextLimitKey(key string) bool {
normalized := normalizeMetadataKey(key)
if normalized == "" {
return false
}
switch normalized {
case
"contextlimit",
"contextwindow",
"contextlength",
"maxcontext",
"maxcontexttokens",
"maxinputtokens",
"maxinputtoken",
"inputtokenlimit":
return true
}
return strings.Contains(normalized, "context") &&
(strings.Contains(normalized, "limit") ||
strings.Contains(normalized, "window") ||
strings.Contains(normalized, "length") ||
strings.HasPrefix(normalized, "max"))
}
func normalizeMetadataKey(key string) string {
var b strings.Builder
b.Grow(len(key))
for _, r := range key {
switch {
case r >= 'a' && r <= 'z':
b.WriteRune(r)
case r >= 'A' && r <= 'Z':
b.WriteRune(r + ('a' - 'A'))
case r >= '0' && r <= '9':
b.WriteRune(r)
}
}
return b.String()
}
func numericContextLimitValue(value any) (int64, bool) {
switch typed := value.(type) {
case int64:
return positiveInt64(typed)
case int32:
return positiveInt64(int64(typed))
case int:
return positiveInt64(int64(typed))
case float64:
casted := int64(typed)
if typed > 0 && float64(casted) == typed {
return casted, true
}
case string:
parsed, err := strconv.ParseInt(strings.TrimSpace(typed), 10, 64)
if err == nil {
return positiveInt64(parsed)
}
case json.Number:
parsed, err := typed.Int64()
if err == nil {
return positiveInt64(parsed)
}
}
return 0, false
}
func positiveInt64(value int64) (int64, bool) {
if value <= 0 {
return 0, false
}
return value, true
}
+311
View File
@@ -0,0 +1,311 @@
package chatloop
import (
"context"
"database/sql"
"iter"
"strings"
"testing"
"charm.land/fantasy"
fantasyanthropic "charm.land/fantasy/providers/anthropic"
"github.com/stretchr/testify/require"
"golang.org/x/xerrors"
"github.com/coder/coder/v2/coderd/chatd/chatprompt"
)
func TestRun_ReportOnlyPrepareBehavior(t *testing.T) {
t.Parallel()
var capturedCall fantasy.Call
model := &loopTestModel{
provider: fantasyanthropic.Name,
streamFn: func(_ context.Context, call fantasy.Call) (fantasy.StreamResponse, error) {
capturedCall = call
return streamFromParts([]fantasy.StreamPart{
{Type: fantasy.StreamPartTypeTextStart, ID: "text-1"},
{Type: fantasy.StreamPartTypeTextDelta, ID: "text-1", Delta: "done"},
{Type: fantasy.StreamPartTypeTextEnd, ID: "text-1"},
{Type: fantasy.StreamPartTypeFinish, FinishReason: fantasy.FinishReasonStop},
}), nil
},
}
persistAssistantCalls := 0
persistToolResultCalls := 0
persistedContextLimit := sql.NullInt64{}
_, err := Run(context.Background(), RunOptions{
Model: model,
Messages: []fantasy.Message{
textMessage(fantasy.MessageRoleSystem, "sys-1"),
textMessage(fantasy.MessageRoleSystem, "sys-2"),
textMessage(fantasy.MessageRoleUser, "hello"),
textMessage(fantasy.MessageRoleAssistant, "working"),
textMessage(fantasy.MessageRoleUser, "continue"),
},
Tools: []fantasy.AgentTool{
newNoopTool(reportOnlyToolName),
newNoopTool("read_file"),
},
MaxSteps: 3,
ReportOnly: true,
ContextLimitFallback: 4096,
PersistAssistant: func(
_ context.Context,
_ []fantasy.Content,
_ fantasy.Usage,
contextLimit sql.NullInt64,
) error {
persistAssistantCalls++
persistedContextLimit = contextLimit
return nil
},
PersistToolResult: func(_ context.Context, _ chatprompt.ToolResultBlock) error {
persistToolResultCalls++
return nil
},
})
require.NoError(t, err)
require.Equal(t, 1, persistAssistantCalls)
require.Equal(t, 0, persistToolResultCalls)
require.True(t, persistedContextLimit.Valid)
require.Equal(t, int64(4096), persistedContextLimit.Int64)
require.NotEmpty(t, capturedCall.Prompt)
require.False(t, containsPromptSentinel(capturedCall.Prompt))
require.Len(t, capturedCall.Tools, 1)
require.Equal(t, reportOnlyToolName, capturedCall.Tools[0].GetName())
require.Len(t, capturedCall.Prompt, 5)
require.False(t, hasAnthropicEphemeralCacheControl(capturedCall.Prompt[0]))
require.True(t, hasAnthropicEphemeralCacheControl(capturedCall.Prompt[1]))
require.False(t, hasAnthropicEphemeralCacheControl(capturedCall.Prompt[2]))
require.True(t, hasAnthropicEphemeralCacheControl(capturedCall.Prompt[3]))
require.True(t, hasAnthropicEphemeralCacheControl(capturedCall.Prompt[4]))
}
func TestRun_InterruptedStepPersistsSyntheticToolResult(t *testing.T) {
t.Parallel()
started := make(chan struct{})
model := &loopTestModel{
provider: "fake",
streamFn: func(ctx context.Context, _ fantasy.Call) (fantasy.StreamResponse, error) {
return iter.Seq[fantasy.StreamPart](func(yield func(fantasy.StreamPart) bool) {
parts := []fantasy.StreamPart{
{
Type: fantasy.StreamPartTypeToolInputStart,
ID: "interrupt-tool-1",
ToolCallName: "read_file",
},
{
Type: fantasy.StreamPartTypeToolInputDelta,
ID: "interrupt-tool-1",
ToolCallName: "read_file",
Delta: `{"path":"main.go"`,
},
{Type: fantasy.StreamPartTypeTextStart, ID: "text-1"},
{Type: fantasy.StreamPartTypeTextDelta, ID: "text-1", Delta: "partial assistant output"},
}
for _, part := range parts {
if !yield(part) {
return
}
}
select {
case <-started:
default:
close(started)
}
<-ctx.Done()
_ = yield(fantasy.StreamPart{
Type: fantasy.StreamPartTypeError,
Error: ctx.Err(),
})
}), nil
},
}
interruptedErr := xerrors.New("chat interrupted")
ctx, cancel := context.WithCancelCause(context.Background())
defer cancel(nil)
go func() {
<-started
cancel(interruptedErr)
}()
persistedAssistantCtxErr := xerrors.New("unset")
var persistedAssistantContent []fantasy.Content
persistedToolResults := make([]chatprompt.ToolResultBlock, 0, 1)
_, err := Run(ctx, RunOptions{
Model: model,
Messages: []fantasy.Message{
textMessage(fantasy.MessageRoleUser, "hello"),
},
Tools: []fantasy.AgentTool{
newNoopTool("read_file"),
},
MaxSteps: 3,
PersistAssistant: func(
persistCtx context.Context,
content []fantasy.Content,
_ fantasy.Usage,
_ sql.NullInt64,
) error {
persistedAssistantCtxErr = persistCtx.Err()
persistedAssistantContent = append([]fantasy.Content(nil), content...)
return nil
},
PersistToolResult: func(_ context.Context, result chatprompt.ToolResultBlock) error {
persistedToolResults = append(persistedToolResults, result)
return nil
},
InterruptedError: interruptedErr,
})
require.ErrorIs(t, err, interruptedErr)
require.NoError(t, persistedAssistantCtxErr)
require.NotEmpty(t, persistedAssistantContent)
var (
foundText bool
foundToolCall bool
)
for _, block := range persistedAssistantContent {
if text, ok := fantasy.AsContentType[fantasy.TextContent](block); ok {
if strings.Contains(text.Text, "partial assistant output") {
foundText = true
}
continue
}
if toolCall, ok := fantasy.AsContentType[fantasy.ToolCallContent](block); ok {
if toolCall.ToolCallID == "interrupt-tool-1" &&
toolCall.ToolName == "read_file" &&
strings.Contains(toolCall.Input, `"path":"main.go"`) {
foundToolCall = true
}
}
}
require.True(t, foundText)
require.True(t, foundToolCall)
require.Len(t, persistedToolResults, 1)
require.Equal(t, "interrupt-tool-1", persistedToolResults[0].ToolCallID)
require.Equal(t, "read_file", persistedToolResults[0].ToolName)
require.True(t, persistedToolResults[0].IsError)
resultMap, ok := persistedToolResults[0].Result.(map[string]any)
require.True(t, ok)
require.Equal(t, "interrupted", resultMap["status"])
errorText, ok := resultMap["error"].(string)
require.True(t, ok)
require.Contains(t, strings.ToLower(errorText), "interrupted")
}
type loopTestModel struct {
provider string
model string
streamFn func(context.Context, fantasy.Call) (fantasy.StreamResponse, error)
}
func (m *loopTestModel) Provider() string {
if m.provider != "" {
return m.provider
}
return "fake"
}
func (m *loopTestModel) Model() string {
if m.model != "" {
return m.model
}
return "fake"
}
func (*loopTestModel) Generate(context.Context, fantasy.Call) (*fantasy.Response, error) {
return &fantasy.Response{}, nil
}
func (m *loopTestModel) Stream(ctx context.Context, call fantasy.Call) (fantasy.StreamResponse, error) {
if m.streamFn != nil {
return m.streamFn(ctx, call)
}
return streamFromParts([]fantasy.StreamPart{{
Type: fantasy.StreamPartTypeFinish,
FinishReason: fantasy.FinishReasonStop,
}}), nil
}
func (*loopTestModel) GenerateObject(context.Context, fantasy.ObjectCall) (*fantasy.ObjectResponse, error) {
return nil, xerrors.New("not implemented")
}
func (*loopTestModel) StreamObject(context.Context, fantasy.ObjectCall) (fantasy.ObjectStreamResponse, error) {
return nil, xerrors.New("not implemented")
}
func streamFromParts(parts []fantasy.StreamPart) fantasy.StreamResponse {
return iter.Seq[fantasy.StreamPart](func(yield func(fantasy.StreamPart) bool) {
for _, part := range parts {
if !yield(part) {
return
}
}
})
}
func newNoopTool(name string) fantasy.AgentTool {
return fantasy.NewAgentTool(
name,
"test noop tool",
func(context.Context, struct{}, fantasy.ToolCall) (fantasy.ToolResponse, error) {
return fantasy.ToolResponse{}, nil
},
)
}
func textMessage(role fantasy.MessageRole, text string) fantasy.Message {
return fantasy.Message{
Role: role,
Content: []fantasy.MessagePart{
fantasy.TextPart{Text: text},
},
}
}
func containsPromptSentinel(prompt []fantasy.Message) bool {
for _, message := range prompt {
if message.Role != fantasy.MessageRoleUser || len(message.Content) != 1 {
continue
}
textPart, ok := fantasy.AsMessagePart[fantasy.TextPart](message.Content[0])
if !ok {
continue
}
if strings.HasPrefix(textPart.Text, "__chatd_agent_prompt_sentinel_") {
return true
}
}
return false
}
func hasAnthropicEphemeralCacheControl(message fantasy.Message) bool {
if len(message.ProviderOptions) == 0 {
return false
}
options, ok := message.ProviderOptions[fantasyanthropic.Name]
if !ok {
return false
}
cacheOptions, ok := options.(*fantasyanthropic.ProviderCacheControlOptions)
return ok && cacheOptions.CacheControl.Type == "ephemeral"
}