// Package agent implements the LLM tool-calling loop, cancellation and // summarisation of long sessions. package agent import ( "context" "errors" "fmt" "strings" "sync" "github.com/crypto/byteclaw/internal/llm" "github.com/crypto/byteclaw/internal/store" "github.com/crypto/byteclaw/internal/tools" ) // ErrCancelled is returned by Run when the user stopped the answer. var ErrCancelled = errors.New("cancelled by user") // Event is streamed to clients while the agent works. type Event struct { Type string `json:"type"` // session | delta | tool_call | tool_result | summarized | done | cancelled | error Text string `json:"text,omitempty"` Tool string `json:"tool,omitempty"` Args string `json:"args,omitempty"` CallID string `json:"call_id,omitempty"` Session string `json:"session_id,omitempty"` } // SummarizeConfig controls automatic summarisation. type SummarizeConfig struct { Enabled bool MaxMessages int MaxTokens int KeepRecent int } type Agent struct { // LLM is used when Resolve is nil. LLM llm.Provider // Resolve picks the provider for a session (e.g. per-session override). Resolve func(ctx context.Context, sessionID string) (llm.Provider, error) Tools *tools.Registry Store *store.Store SystemPrompt string MaxIterations int HistoryLimit int Summarize SummarizeConfig mu sync.Mutex locks map[string]*sync.Mutex runs map[string]context.CancelCauseFunc } func (a *Agent) sessionLock(id string) *sync.Mutex { a.mu.Lock() defer a.mu.Unlock() if a.locks == nil { a.locks = map[string]*sync.Mutex{} } l, ok := a.locks[id] if !ok { l = &sync.Mutex{} a.locks[id] = l } return l } func (a *Agent) provider(ctx context.Context, sessionID string) (llm.Provider, error) { if a.Resolve != nil { return a.Resolve(ctx, sessionID) } if a.LLM == nil { return nil, errors.New("no LLM provider configured") } return a.LLM, nil } // Cancel stops the running answer of a session. Returns false if none is running. func (a *Agent) Cancel(sessionID string) bool { a.mu.Lock() defer a.mu.Unlock() if c, ok := a.runs[sessionID]; ok { c(ErrCancelled) return true } return false } // Running reports whether an answer is in progress for the session. func (a *Agent) Running(sessionID string) bool { a.mu.Lock() defer a.mu.Unlock() _, ok := a.runs[sessionID] return ok } func (a *Agent) register(sessionID string, c context.CancelCauseFunc) { a.mu.Lock() defer a.mu.Unlock() if a.runs == nil { a.runs = map[string]context.CancelCauseFunc{} } a.runs[sessionID] = c } func (a *Agent) unregister(sessionID string) { a.mu.Lock() defer a.mu.Unlock() delete(a.runs, sessionID) } // Run appends the user message to the session and loops until the model // answers without tool calls. Returns the final assistant text. If the run is // cancelled via Cancel, the partial answer is stored (marked cancelled) and // ErrCancelled is returned together with the partial text. func (a *Agent) Run(parent context.Context, sessionID, userText string, emit func(Event)) (string, error) { if emit == nil { emit = func(Event) {} } l := a.sessionLock(sessionID) l.Lock() defer l.Unlock() ctx, cancel := context.WithCancelCause(parent) defer cancel(nil) a.register(sessionID, cancel) defer a.unregister(sessionID) // Writes after a cancellation must still succeed. saveCtx := context.WithoutCancel(ctx) if _, err := a.Store.AppendMessage(saveCtx, sessionID, llm.Message{Role: llm.RoleUser, Content: userText}); err != nil { return "", err } prov, err := a.provider(ctx, sessionID) if err != nil { return "", err } if a.Summarize.Enabled { if sm, err := a.summarizeLocked(ctx, prov, sessionID, false); err == nil && sm != nil { emit(Event{Type: "summarized", Text: sm.Content}) } } var specs []llm.ToolSpec if a.Tools != nil { specs = a.Tools.Specs() } cancelled := func() bool { return errors.Is(context.Cause(ctx), ErrCancelled) } for i := 0; i < a.MaxIterations; i++ { if cancelled() { return "", ErrCancelled } msgs, err := a.history(saveCtx, sessionID) if err != nil { return "", err } var partial strings.Builder resp, err := prov.Chat(ctx, llm.Request{Messages: msgs, Tools: specs}, func(t string) { partial.WriteString(t) emit(Event{Type: "delta", Text: t}) }) if err != nil { if cancelled() { text := partial.String() if text != "" { if _, serr := a.Store.AppendMessage(saveCtx, sessionID, llm.Message{Role: llm.RoleAssistant, Content: text, Cancelled: true}); serr != nil { return text, serr } } return text, ErrCancelled } return "", err } asst := llm.Message{Role: llm.RoleAssistant, Content: resp.Content, ToolCalls: resp.ToolCalls} if _, err := a.Store.AppendMessage(saveCtx, sessionID, asst); err != nil { return "", err } if len(resp.ToolCalls) == 0 { return resp.Content, nil } // Some models (seen with GLM/DeepSeek on Dahl) emit the same call several times in // parallel; identical name+arguments run once and share the result. seen := map[string]string{} for _, tc := range resp.ToolCalls { emit(Event{Type: "tool_call", Tool: tc.Name, Args: tc.Arguments, CallID: tc.ID}) key := tc.Name + "\x00" + tc.Arguments out, dup := seen[key] switch { case dup: case cancelled(): out = "error: cancelled by user" default: out = a.Tools.Call(ctx, tc.Name, tc.Arguments) seen[key] = out } emit(Event{Type: "tool_result", Tool: tc.Name, Text: clip(out, 2000), CallID: tc.ID}) // Every tool call gets a result so the history stays valid for the provider. if _, err := a.Store.AppendMessage(saveCtx, sessionID, llm.Message{Role: llm.RoleTool, ToolCallID: tc.ID, Name: tc.Name, Content: out}); err != nil { return "", err } } } return "", fmt.Errorf("stopped after %d iterations without a final answer", a.MaxIterations) } func (a *Agent) history(ctx context.Context, sessionID string) ([]llm.Message, error) { var after int64 summary := "" if sm, err := a.Store.LatestSummary(ctx, sessionID); err == nil { after, summary = sm.UptoMessageID, sm.Content } else if !errors.Is(err, store.ErrNotFound) { return nil, err } stored, err := a.Store.MessagesAfter(ctx, sessionID, after, a.HistoryLimit) if err != nil { return nil, err } // A truncated window must not start with orphaned tool results. for len(stored) > 0 && stored[0].Role == llm.RoleTool { stored = stored[1:] } msgs := make([]llm.Message, 0, len(stored)+2) if a.SystemPrompt != "" { msgs = append(msgs, llm.Message{Role: llm.RoleSystem, Content: a.SystemPrompt}) } if summary != "" { msgs = append(msgs, llm.Message{Role: llm.RoleSystem, Content: "Summary of the earlier conversation:\n" + summary}) } for _, m := range stored { mm := m.Message mm.Cancelled = false msgs = append(msgs, mm) } return msgs, nil } func clip(s string, n int) string { if len(s) <= n { return s } return s[:n] + "…" }