// Package conversation manages the saved history, model inference, tool // execution, and live event streaming for a single conversation. // // A [Conversation] coordinates access to one conversation in SQLite: // - [Conversation.Reserve] reserves the conversation for one // [Inference], and [Inference.Run] runs the model and tool-call loop. // Only one inference call or message edit may run at a time; // concurrent calls return [ErrBusy]. // - [Conversation.Subscribe] synchronizes a client's cached messages with // saved history and streams live updates from an active inference run. // - Message and tool-call edit methods modify saved history when no // inference is running, using optimistic timestamp checks // ([ErrRevisionConflict]) to prevent lost updates. package conversation import ( "context" "database/sql" "encoding/json/jsontext" "encoding/json/v2" "errors" "fmt" "log/slog" "runtime" "slices" "sync" "time" "uuid" "code.chimeric.al/chimerical/odidere/internal/db" "code.chimeric.al/chimerical/odidere/internal/llm" "code.chimeric.al/chimerical/odidere/internal/tool" ) const ( // interruptedWarning indicates a tool call started but did not finish // normally, so its effects are unknown. interruptedWarning = `▸ WARNING: the tool started, but its outcome is unknown. It may have had effects. Do not assume it is safe to repeat.` // resultSaveTimeout is the time limit for saving one tool result. resultSaveTimeout = time.Second ) var ( // ErrApprovalRequired indicates that saved tool calls await user // approval. ErrApprovalRequired = errors.New("tool approval required") // ErrBusy indicates that inference is ongoing. ErrBusy = errors.New("conversation busy") // ErrMaxIterations indicates that the model-request limit was reached // without a final assistant message. ErrMaxIterations = errors.New("max iterations exceeded") // ErrRevisionConflict indicates that the message was changed before // the caller's revision. ErrRevisionConflict = errors.New("message revision conflict") ) // DecodeMessages converts saved rows into llm.Messages, validating each. // It stops at the first row that fails to unmarshal or validate; the // returned slice is nil on error. func DecodeMessages(rows []db.Message) ([]llm.Message, error) { msgs := make([]llm.Message, len(rows)) for i, row := range rows { if err := json.Unmarshal( row.Content, &msgs[i], ); err != nil { return nil, fmt.Errorf( "unmarshal message %q: %w", row.Id, err, ) } if err := msgs[i].Validate(); err != nil { return nil, fmt.Errorf( "invalid message %q: %w", row.Id, err, ) } } return msgs, nil } // New constructs an in-memory conversation manager. func New( cfg Config, sdb *db.SQLite, log *slog.Logger, tools *tool.Registry, ) (*Conversation, error) { if err := cfg.Validate(); err != nil { return nil, fmt.Errorf("invalid config: %w", err) } if sdb == nil { return nil, fmt.Errorf("missing database") } if log == nil { return nil, fmt.Errorf("missing logger") } if tools == nil { return nil, fmt.Errorf("missing tool registry") } return &Conversation{ id: cfg.Id, db: sdb, dbr: db.New(sdb.Reader), dbw: db.New(sdb.Writer), log: log, tools: tools, }, nil } // Config identifies the saved conversation to manage. type Config struct { Id string } // Validate checks the configuration without accessing the database. func (cfg Config) Validate() error { if cfg.Id == "" { return fmt.Errorf("missing conversation id") } return nil } // Conversation is an in-memory handle for one saved conversation. type Conversation struct { id string db *db.SQLite dbr *db.Queries dbw *db.Queries log *slog.Logger tools *tool.Registry // mu synchronizes access to cancel and inf. mu sync.Mutex // cancel is used to stop inference. // When set, it indicates inference is reserved or ongoing. cancel context.CancelFunc // inf shares the reserved Inference's events with subscribers. // When set, it indicates inference is reserved or ongoing. inf *Inference } // MessageManifest identifies one saved message cached by a client. type MessageManifest struct { Id string UpdatedAt time.Time } // EventType distinguishes the events delivered by Subscribe. type EventType string const ( EventAssistantUpdate EventType = "assistant_update" EventDone EventType = "done" EventMessageDeleted EventType = "message_deleted" EventMessageSaved EventType = "message_saved" ) // Event is a conversation update delivered to subscribers. type Event interface { // Type reports the kind of the event. Type() EventType } // MessageSaved reports a message was saved. type MessageSaved struct { Message db.Message `json:"message"` } func (MessageSaved) Type() EventType { return EventMessageSaved } // MessageDeleted reports a message ID the client's manifest holds but // the saved history no longer contains. type MessageDeleted struct { MessageId string `json:"message_id"` } func (MessageDeleted) Type() EventType { return EventMessageDeleted } // AssistantUpdate reports the next streamed fragment of one assistant // message. type AssistantUpdate struct { MessageId string `json:"message_id"` Chunk llm.StreamChunk `json:"chunk"` } func (AssistantUpdate) Type() EventType { return EventAssistantUpdate } // Done reports that the Inference the subscriber joined has returned. // Error is empty on success. type Done struct { Error string `json:"error,omitempty"` } func (Done) Type() EventType { return EventDone } // Inference is one reserved inference call and its events. // [Conversation.Reserve] creates it, and [Inference.Run] runs it. // [Inference.Events] and [Conversation.Subscribe] use its events to send // past and live updates to subscribers. type Inference struct { c *Conversation client *llm.Client req InferenceRequest // ctx is canceled by cancel, which Stop and Delete also call. ctx context.Context cancel context.CancelFunc // mu serializes access to the fields below. mu sync.Mutex // started reports whether Run has been called. started bool // changed notifies waiting subscribers of new events or completion. // Each publish closes changed and replaces it with a new channel; // finish closes it without replacement. changed chan struct{} // done reports whether the inference work has finished. done bool // err reports why inference failed, or nil on success. err error // events holds all published events in order. // Events are added, never removed. // Each subscriber keeps its own read position in events. events []Event } // publish appends e to the events and wakes waiting subscribers. // The caller must hold inf.mu. func (inf *Inference) publish(e Event) { inf.events = append(inf.events, e) close(inf.changed) inf.changed = make(chan struct{}) } // finish marks the call complete and records err. // It closes changed to wake waiting subscribers. // The channel stays closed because the call is complete. func (inf *Inference) finish(err error) { inf.mu.Lock() defer inf.mu.Unlock() inf.done = true inf.err = err close(inf.changed) } // Events streams the call's events from the start, then ends with // [Done]. The channel closes after Done or when ctx is canceled. // Canceling ctx stops delivery; it does not stop the call. func (inf *Inference) Events(ctx context.Context) <-chan Event { ch := make(chan Event, 64) go func() { defer close(ch) inf.follow(ctx, ch, nil) }() return ch } // follow sends the call's events to ch from index 0, skipping // [MessageSaved] and [AssistantUpdate] events whose message ID is in // skip. It waits for new events until the call finishes, then sends // [Done]. It returns false if ctx is canceled first. func (inf *Inference) follow( ctx context.Context, ch chan Event, skip map[string]struct{}, ) bool { pos := 0 for { // Copy the slice header for unread events and the current // notification channel. Because the call only appends to // events, reading batch after Unlock is safe. inf.mu.Lock() batch := inf.events[pos:] pos = len(inf.events) wait, done, err := inf.changed, inf.done, inf.err inf.mu.Unlock() for _, e := range batch { var id string switch e := e.(type) { case MessageSaved: id = e.Message.Id case AssistantUpdate: id = e.MessageId } if id != "" { if _, ok := skip[id]; ok { continue } } if !send(ctx, ch, e) { return false } } if done { var errStr string if err != nil { errStr = err.Error() } return send(ctx, ch, Done{Error: errStr}) } // Wait for the call to close wait (when it appends new // events or finishes), or for the subscriber to cancel ctx. select { case <-wait: case <-ctx.Done(): return false } } } // Subscribe streams events that synchronize a client's message cache with // the saved conversation and any reserved or ongoing [Inference]. // // manifest lists the message IDs and timestamps the caller already holds. // An empty manifest requests all saved messages. // // Events arrive on the returned channel in two phases: // 1. Cache diff: [MessageSaved] for each saved message missing from or // newer than manifest, followed by [MessageDeleted] for each ID in // manifest that is no longer saved. // 2. Active inference: if an [Inference] is reserved or running when // Subscribe loads the database, Subscribe streams its // [MessageSaved] and [AssistantUpdate] events, then ends with [Done]. // If no inference is reserved or running, Subscribe sends no [Done] // event. // // The channel closes when delivery finishes or ctx is canceled. Canceling // ctx stops delivery to this subscriber; it does not stop an ongoing // [Inference]. func (c *Conversation) Subscribe( ctx context.Context, manifest []MessageManifest, ) (<-chan Event, error) { // Check for a reserved or ongoing Inference. c.mu.Lock() inf := c.inf c.mu.Unlock() // Inference.Run holds inf.mu while it commits a message and // appends the MessageSaved event to inf.events. Holding inf.mu // across this read ensures that any message committed after this // point will appear in inf.events, and any message already in rows // will be recorded in the diff's sent map. if inf != nil { inf.mu.Lock() } rows, err := c.dbr.SelectMessagesConversationId(ctx, c.id) if inf != nil { inf.mu.Unlock() } if err != nil { return nil, fmt.Errorf("load messages: %w", err) } events, sent := diffMessages(rows, manifest) // Buffer events so bursts (such as the initial diff or fast token // streaming) do not cause a context switch on every event. ch := make(chan Event, 64) go func() { defer close(ch) for _, e := range events { if !send(ctx, ch, e) { return } } if inf == nil { return } // Skip events for messages that the initial diff already // sent. If the client's manifest already had a message from // this call, it is not in sent and will be re-sent here; the // client deduplicates by ID and UpdatedAt. inf.follow(ctx, ch, sent) }() return ch, nil } // send writes e to ch and returns true, or returns false if ctx is // canceled before ch can accept e. This prevents a slow or disconnected // subscriber from blocking the producer goroutine when ch is full. func send(ctx context.Context, ch chan Event, e Event) bool { select { case ch <- e: return true case <-ctx.Done(): return false } } // diffMessages compares saved rows against manifest. // // It returns: // - events: [MessageSaved] for each row missing from or newer than // manifest (in saved order), followed by [MessageDeleted] for each // ID in manifest that is not in rows. // - sent: the set of message IDs included in the [MessageSaved] events. func diffMessages( rows []db.Message, manifest []MessageManifest, ) (events []Event, sent map[string]struct{}) { var ( // known maps each client-cached message id to its timestamp. known = make(map[string]time.Time, len(manifest)) // saved holds ids of all messages currently in the database. saved = make(map[string]struct{}, len(rows)) ) for _, m := range manifest { known[m.Id] = m.UpdatedAt } // Emit MessageSaved for new or updated rows, in saved order. sent = map[string]struct{}{} for _, row := range rows { saved[row.Id] = struct{}{} if at, ok := known[row.Id]; ok && at.Equal(row.UpdatedAt) { continue } events = append(events, MessageSaved{Message: row}) sent[row.Id] = struct{}{} } // Emit MessageDeleted for client-cached ids no longer in the database. for id := range known { if _, ok := saved[id]; !ok { events = append(events, MessageDeleted{MessageId: id}) } } return events, sent } // InferenceRequest supplies input and settings for one inference/tool loop. // Runtime dependencies are passed separately to Inference. type InferenceRequest struct { // Message supplies new user input, including text and attachments. // nil continues from saved history without adding a message. Message *llm.Message // Model is the LLM model ID. If empty, the default model is used. Model string // MaxIterations caps the agent loop iterations. Zero means unlimited. MaxIterations int // ThinkingBudgetTokens is the llama.cpp reasoning budget. // (-1 = unlimited, 0 = off). ThinkingBudgetTokens *int // Timeout limits the inference time (e.g., "10m"). Timeout time.Duration // Stream selects provider streaming, not HTTP response delivery. // Failed streams never save partial assistant messages. // False requests a completed provider response. Stream bool } // Validate checks inference limits and optional user input. func (r InferenceRequest) Validate() error { if r.MaxIterations < 0 { return fmt.Errorf("negative max_iterations") } if r.ThinkingBudgetTokens != nil && *r.ThinkingBudgetTokens < -1 { return fmt.Errorf("invalid thinking_budget_tokens") } if r.Timeout < 0 { return fmt.Errorf("negative timeout") } if m := r.Message; m != nil { if m.Role != llm.RoleUser || len(m.ToolCalls) != 0 || m.ToolCallID != "" || m.Reasoning != "" || m.ReasoningContent != "" || m.Refusal != "" { return fmt.Errorf("invalid message") } if err := m.Validate(); err != nil { return fmt.Errorf("invalid message: %w", err) } } return nil } // Reserve validates req and reserves the conversation for one inference // call. It returns ErrBusy if an inference call is reserved or running. // // ctx bounds the call's work, not the call to Reserve: the call stops // when ctx is canceled, when req.Timeout expires, or on Stop or Delete. // Pass a context that outlives any single HTTP request, so a client // disconnect does not stop the call. // // The caller must call [Inference.Run] on the result. Until Run returns, // the conversation stays busy: other calls to Reserve and all edits // return ErrBusy, even if Run is never called. func (c *Conversation) Reserve( ctx context.Context, client *llm.Client, req InferenceRequest, ) (*Inference, error) { // Reject invalid input before accepting work. if client == nil { return nil, fmt.Errorf("missing llm client") } if err := req.Validate(); err != nil { return nil, fmt.Errorf("invalid inference request: %w", err) } // Limit the work duration and allow Stop to cancel it. var cancel context.CancelFunc if req.Timeout > 0 { ctx, cancel = context.WithTimeout(ctx, req.Timeout) } else { ctx, cancel = context.WithCancel(ctx) } // Check busy and reserve in one critical section, so no other // Reserve or edit can start between the check and the reservation. inf := &Inference{ c: c, client: client, req: req, ctx: ctx, cancel: cancel, changed: make(chan struct{}), } c.mu.Lock() defer c.mu.Unlock() if c.cancel != nil { cancel() return nil, ErrBusy } c.cancel = cancel c.inf = inf return inf, nil } // Run resolves outstanding tool calls in saved history, saves optional // new user input, and runs the model/tool loop, blocking until the work // and result-save attempts finish. Pending approvals return // ErrApprovalRequired after sibling calls finish their save attempts; // new user input is not saved while calls remain unresolved. // // When Run returns, it sends [Done] to subscribers and releases the // reservation. Call Run once; a second call returns an error. func (inf *Inference) Run() (err error) { inf.mu.Lock() started := inf.started inf.started = true inf.mu.Unlock() if started { return fmt.Errorf("inference already run") } c := inf.c defer func() { // Every exit path, including errors, finishes the call. // Every failing path below must assign err, not shadow it. inf.finish(err) inf.cancel() c.mu.Lock() c.cancel = nil c.inf = nil c.mu.Unlock() }() var ( ctx = inf.ctx client = inf.client req = inf.req ) // Check that the conversation exists. conv, err := c.dbr.SelectConversationId(ctx, c.id) if err != nil { return fmt.Errorf("load conversation: %w", err) } log := c.log.With( slog.String("conversation_id", conv.Id), ) // Load saved rows in conversation order. rows, err := c.dbr.SelectMessagesConversationId(ctx, c.id) if err != nil { return fmt.Errorf("load history: %w", err) } base := len(rows) // Convert db.Message into llm.Message; validate. history, err := DecodeMessages(rows) if err != nil { return fmt.Errorf("decode history: %w", err) } // Find the last message that is not a tool result. // For an assistant role with tool calls, run calls without a result. var ( aId string remaining []llm.ToolCall answered = map[string]struct{}{} ) for i, m := range slices.Backward(history) { if m.Role == llm.RoleTool { answered[m.ToolCallID] = struct{}{} continue } if len(m.ToolCalls) != 0 { aId = rows[i].Id for _, call := range m.ToolCalls { if _, ok := answered[call.ID]; !ok { remaining = append(remaining, call) } } } // Done: we've found the last non-tool message. break } // Resolve tool calls before adding input or requesting inference. if len(remaining) != 0 { log.InfoContext( ctx, "executing remaining tool calls", slog.Int("count", len(remaining)), ) var tres []llm.Message tres, err = c.callTools(ctx, log, aId, remaining) if err != nil { return fmt.Errorf("call tools: %w", err) } history = append(history, tres...) } // Save new user input before requesting a completion. if req.Message != nil { log.InfoContext(ctx, "persisting new message") var saved *db.Message inf.mu.Lock() saved, err = c.insertMessage( ctx, nil, *req.Message, nil, nil, ) if err == nil { inf.publish(MessageSaved{Message: *saved}) } inf.mu.Unlock() if err != nil { return fmt.Errorf("insert message: %w", err) } log.InfoContext( ctx, "persisted new user message", slog.String("id", saved.Id), ) // Give an untitled conversation its title before the model // runs, so a slow completion does not delay the title. A // failure here costs the title, not the run: the next user // message tries again. if err := c.setTitle(ctx, *req.Message); err != nil { log.WarnContext( ctx, "failed to derive conversation title", slog.Any("error", err), ) } // Continue from the updated history. history = append(history, *req.Message) } // Loop for inference. var ( completed bool start time.Time tools = c.tools.ToAPI() ) log.InfoContext(ctx, "starting inference") start = time.Now() for i := 0; req.MaxIterations == 0 || i < req.MaxIterations; i++ { var ( cp = llm.CompletionsParams{ Messages: history, Model: req.Model, ThinkingBudgetTokens: req.ThinkingBudgetTokens, Tools: tools, } // aid identifies the assistant message this iteration // produces, created before the request so the // assistant_update events and the saved row share it. aid = uuid.NewV4() res *llm.CompletionResult row *db.Message stream *llm.CompletionStream ) if req.Stream { stream, err = client.CompletionsStream(ctx, cp) if err != nil { err = fmt.Errorf("completions stream: %w", err) break } for stream.Next() { inf.mu.Lock() inf.publish(AssistantUpdate{ MessageId: aid.String(), Chunk: stream.Current(), }) inf.mu.Unlock() } err = errors.Join( stream.Err(), stream.Close(), ) if err != nil { break } sr := stream.Result() res = &sr } else { res, err = client.Completions(ctx, cp) if err != nil { err = fmt.Errorf("completions: %w", err) break } } // Reject invalid assistant output before saving it. if res.Message.Role != llm.RoleAssistant { err = fmt.Errorf( "invalid message role: %q", res.Message.Role) break } if err = res.Message.Validate(); err != nil { err = fmt.Errorf("invalid assistant message: %w", err) break } // Persist the assistant message before acting on it, so an // interruption leaves a committed record. inf.mu.Lock() row, err = c.insertMessage( ctx, &aid, res.Message, &res.Metadata, nil, ) if err == nil { inf.publish(MessageSaved{Message: *row}) } inf.mu.Unlock() if err != nil { err = fmt.Errorf("insert message: %w", err) break } history = append(history, res.Message) // If there are no more tool calls, we're done. if len(res.Message.ToolCalls) == 0 { completed = true break } // Resolve the reply's tool calls before the next request. tres, toolErr := c.callTools( ctx, log, row.Id, res.Message.ToolCalls, ) if toolErr != nil { err = errors.Join(toolErr, ctx.Err()) break } history = append(history, tres...) } duration := time.Since(start) if err == nil && !completed { err = ErrMaxIterations } if err != nil { log.ErrorContext( ctx, "inference failed", slog.Any("error", err), slog.Duration("duration", duration), ) return err } log.InfoContext( ctx, "inference complete", slog.Duration("duration", duration), slog.Int("messages", len(history)-base), ) return nil } // callTools runs the tool calls of the assistant message aId, up to // twice the CPU count concurrently, and collects each persisted result. // Each call saves its own result, so the returned order may differ from // the saved order; results are keyed by tool call ID. The error joins // every failure; if non-nil, at least one call has no result. func (c *Conversation) callTools( ctx context.Context, log *slog.Logger, aId string, calls []llm.ToolCall, ) ([]llm.Message, error) { // Load each call's execution status. executions, err := c.dbr.SelectExecutionsMessageId( ctx, aId, ) if err != nil { return nil, fmt.Errorf("load executions: %w", err) } callExecutions := make(map[string]*db.Execution, len(executions)) for i := range executions { callExecutions[executions[i].ToolCallId] = &executions[i] } // Run calls concurrently, up to twice the CPU count at once. type outcome struct { err error msg llm.Message } var ( completed = make(chan outcome, len(calls)) sem = make(chan struct{}, runtime.NumCPU()*2) ) for _, call := range calls { go func() { select { case sem <- struct{}{}: defer func() { <-sem }() case <-ctx.Done(): completed <- outcome{err: ctx.Err()} return } msg, err := c.callTool( ctx, log, aId, call, callExecutions[call.ID], ) completed <- outcome{msg: msg, err: err} }() } // Wait for every worker, even if another worker fails. var ( errs []error results []llm.Message ) for range calls { result := <-completed if result.err != nil { log.ErrorContext( ctx, "tool call failed", slog.Any("error", result.err), ) errs = append(errs, result.err) continue } results = append(results, result.msg) } return results, errors.Join(errs...) } // callTool runs one tool call and saves the tool message that reports // the outcome to the model. // // The record is a row in the executions table, saved before the tool // starts. The row shows whether the tool may have run: // - No row: the tool never started. It is safe to run again. // - Row with status 'pending': the tool requires approval and never // started; it runs once approved. // - Row with status 'running': the tool may have run and had effects. // It is never retried. // - Row with status 'completed': the tool finished or never started, // and the tool message is saved. // // A nil execution is a new call: recorded running, or pending if the // tool requires approval, in which case ErrApprovalRequired is returned // before running. A recorded execution decides the rest: pending returns // ErrApprovalRequired, approved is marked running and run, any other // status is refused. // // Results: // - A message and a nil error: the message is saved. The message // reports the tool's outcome: exit code and output, the error that // kept the tool from starting, or the error that interrupted it. // - An error: the outcome was not recorded. The tool may have run and // had effects. // // The message contains: // - If the tool finished: its output. // - If it never started: the error. // - If it started but was stopped, or its output was cut off: the // error, a warning that its effects are unknown, and any output. // // Limitations: // - callTool relies on [tool.Registry.Execute] returning a nil result // only when the process never started. func (c *Conversation) callTool( ctx context.Context, log *slog.Logger, aId string, call llm.ToolCall, execution *db.Execution, ) (llm.Message, error) { log = log.With( slog.String("tool_call_id", call.ID), slog.String("name", call.Function.Name), ) // Decide whether and how the call may run. A case that ends without // returning has cleared the call to run and continues to the tail. switch { case execution == nil: // Record a new call before running it or requesting approval. status := db.ExecutionStatusRunning if definition, ok := c.tools.Get(call.Function.Name); ok && definition.RequiresApproval { status = db.ExecutionStatusPending } var err error execution, err = c.dbw.InsertExecution( ctx, db.InsertExecutionParams{ MessageId: aId, ToolCallId: call.ID, Status: status, }, ) if err != nil { return llm.Message{}, fmt.Errorf( "insert execution for call %q: %w", call.ID, err, ) } if status == db.ExecutionStatusPending { return llm.Message{}, fmt.Errorf( "call %q: %w", call.ID, ErrApprovalRequired, ) } case execution.Status == db.ExecutionStatusPending: return llm.Message{}, fmt.Errorf( "call %q: %w", call.ID, ErrApprovalRequired, ) case execution.Status == db.ExecutionStatusApproved: if err := c.dbw.UpdateExecutionStatus( ctx, db.UpdateExecutionStatusParams{ Id: execution.Id, Status: db.ExecutionStatusRunning, }, ); err != nil { return llm.Message{}, fmt.Errorf( "start approved call %q: %w", call.ID, err, ) } default: // An existing row that may have run is never retried. return llm.Message{}, fmt.Errorf( "call %q has status %q but no result; will not retry", call.ID, execution.Status, ) } log.InfoContext( ctx, "calling tool", slog.String("args", call.Function.Arguments), ) // Run the tool. tres, toolErr := c.tools.Execute( ctx, call.Function.Name, call.Function.Arguments, ) var content string switch { case toolErr == nil: log.InfoContext( ctx, "called tool", slog.Duration( "duration", tres.FinishedAt.Sub(tres.StartedAt), ), ) content = tres.FormatResult() case tres == nil: log.ErrorContext( ctx, "tool did not start", slog.Any("error", toolErr), ) content = toolErr.Error() default: log.ErrorContext( ctx, "tool interrupted", slog.Any("error", toolErr), slog.Duration( "duration", tres.FinishedAt.Sub(tres.StartedAt), ), ) // Retain output produced before interruption. content = toolErr.Error() + "\n" + interruptedWarning + "\n" + tres.FormatResult() } // Build the tool result msg. msg := llm.Message{ Role: llm.RoleTool, Name: call.Function.Name, ToolCallID: call.ID, ContentParts: []llm.ContentPart{{ Type: llm.ContentTypeText, Text: content, }}, } if err := msg.Validate(); err != nil { return llm.Message{}, fmt.Errorf( "invalid message %q: %w", call.ID, err, ) } // Attempt to save the result, even in case of cancellation. // Tool calls are expensive and may have side-effects; avoid throwing // away completed work. The status update, if any, commits with the // result. saveCtx, cancel := context.WithTimeout( context.WithoutCancel(ctx), resultSaveTimeout, ) defer cancel() // Mark finished tools and failed launches as completed. // Keep interrupted calls as ExecutionStatusRunning; // they may have had effects. var update *db.UpdateExecutionStatusParams if toolErr == nil || tres == nil { update = &db.UpdateExecutionStatusParams{ Id: execution.Id, Status: db.ExecutionStatusCompleted, } } // c.inf is set for the whole of Inference.Run, the only caller path // to callTool. c.mu.Lock() inf := c.inf c.mu.Unlock() // Hold inf.mu across the commit so a concurrent Subscribe's diff // sees the row and its event together. inf.mu.Lock() saved, err := c.insertMessage(saveCtx, nil, msg, nil, update) if err == nil { inf.publish(MessageSaved{Message: *saved}) } inf.mu.Unlock() if err != nil { log.ErrorContext( ctx, "failed to save tool result", slog.Any("error", err), ) return llm.Message{}, fmt.Errorf( "save tool result %q: %w", call.ID, err, ) } return msg, nil } // Stop cancels ongoing inference. It returns without waiting. // Inference may still be finishing after Stop is called. // Stop copies cancel to prevent canceling a new inference request. func (c *Conversation) Stop() { c.mu.Lock() cancel := c.cancel c.mu.Unlock() if cancel != nil { cancel() } } // insertMessage stores msg under id, or a new version 4 UUID if id is // nil, with its optional inference metadata at the next sequence in // the conversation; nil metadata stores NULL. If executionUpdate is // non-nil, its status change commits in the same transaction. The // caller chooses the execution and a valid status transition. func (c *Conversation) insertMessage( ctx context.Context, id *uuid.UUID, msg llm.Message, metadata *llm.Metadata, executionUpdate *db.UpdateExecutionStatusParams, ) (*db.Message, error) { content, err := json.Marshal(msg) if err != nil { return nil, fmt.Errorf("marshal content: %w", err) } var md jsontext.Value if metadata != nil { encoded, err := json.Marshal( metadata, json.FormatNilSliceAsNull(true), ) if err != nil { return nil, fmt.Errorf("marshal metadata: %w", err) } md = jsontext.Value(encoded) } tx, err := c.db.Writer.BeginTx(ctx, nil) if err != nil { return nil, fmt.Errorf("begin tx: %w", err) } defer tx.Rollback() qtx := c.dbw.WithTx(tx) sequence, err := qtx.NextSequence(ctx, c.id) if err != nil { return nil, fmt.Errorf("get next message sequence: %w", err) } saved, err := qtx.InsertMessage(ctx, db.InsertMessageParams{ Content: content, ConversationId: c.id, Id: id, Metadata: md, Role: db.MessageRole(msg.Role), Sequence: sequence, }) if err != nil { return nil, fmt.Errorf("insert message: %w", err) } if executionUpdate != nil { if err := qtx.UpdateExecutionStatus( ctx, *executionUpdate, ); err != nil { return nil, fmt.Errorf( "update execution status: %w", err, ) } } if err := tx.Commit(); err != nil { return nil, fmt.Errorf("commit message tx: %w", err) } return saved, nil } // InsertMessage saves a new user or assistant message at the next // sequence, validating it first. It returns ErrBusy while Inference is // ongoing. func (c *Conversation) InsertMessage( ctx context.Context, msg llm.Message, ) (*db.Message, error) { if msg.Role != llm.RoleAssistant && msg.Role != llm.RoleUser { return nil, fmt.Errorf( "invalid message role: %q", msg.Role, ) } if err := msg.Validate(); err != nil { return nil, fmt.Errorf("invalid message: %w", err) } c.mu.Lock() defer c.mu.Unlock() if c.cancel != nil { return nil, ErrBusy } saved, err := c.insertMessage(ctx, nil, msg, nil, nil) if err != nil { return nil, err } // A user message can give an untitled conversation its title. // A failure here costs the title, not the saved message. if msg.Role == llm.RoleUser { if err := c.setTitle(ctx, msg); err != nil { c.log.WarnContext( ctx, "failed to derive conversation title", slog.Any("error", err), ) } } return saved, nil } // UpdateMessage replaces a message's model-visible content; its role may // not change. A mismatch between updatedAt and the saved updated_at // returns ErrRevisionConflict. Inference metadata is preserved, and may // no longer reflect initially generated output. It returns ErrBusy while // Inference is ongoing. func (c *Conversation) UpdateMessage( ctx context.Context, mId string, updatedAt time.Time, message llm.Message, ) (*db.Message, error) { if err := message.Validate(); err != nil { return nil, fmt.Errorf("invalid message: %w", err) } content, err := json.Marshal(message) if err != nil { return nil, fmt.Errorf("marshal message: %w", err) } c.mu.Lock() defer c.mu.Unlock() if c.cancel != nil { return nil, ErrBusy } row, err := c.dbr.SelectConversationMessageId(ctx, c.id, mId) if err != nil { return nil, fmt.Errorf("load message %q: %w", mId, err) } // Refuse to update a message that has been changed. if !row.UpdatedAt.Equal(updatedAt) { return nil, ErrRevisionConflict } // Validate the role update. if llm.Role(row.Role) != message.Role { return nil, fmt.Errorf("invalid role: %q", message.Role) } tx, err := c.db.Writer.BeginTx(ctx, nil) if err != nil { return nil, fmt.Errorf("begin tx: %w", err) } defer tx.Rollback() qtx := c.dbw.WithTx(tx) updated, err := qtx.UpdateMessageContent( ctx, db.UpdateMessageContentParams{ Id: row.Id, Content: content, }, ) if err != nil { return nil, fmt.Errorf("update message %q: %w", row.Id, err) } if err := tx.Commit(); err != nil { return nil, fmt.Errorf("commit message tx: %w", err) } return updated, nil } // DeleteMessage removes one saved message, checking for conflicting // changes. A mismatch between updatedAt and the saved updated_at returns // ErrRevisionConflict. It returns ErrBusy while Inference is ongoing. func (c *Conversation) DeleteMessage( ctx context.Context, mId string, updatedAt time.Time, ) error { c.mu.Lock() defer c.mu.Unlock() if c.cancel != nil { return ErrBusy } row, err := c.dbr.SelectConversationMessageId(ctx, c.id, mId) if err != nil { return fmt.Errorf("load message %q: %w", mId, err) } // Refuse to delete a message if it has been updated. if !row.UpdatedAt.Equal(updatedAt) { return ErrRevisionConflict } // Delete the row and close the sequence gap it leaves, in one // transaction. tx, err := c.db.Writer.BeginTx(ctx, nil) if err != nil { return fmt.Errorf("begin tx: %w", err) } defer tx.Rollback() qtx := c.dbw.WithTx(tx) if err := qtx.DeleteMessageId(ctx, row.Id); err != nil { return fmt.Errorf("delete message %q: %w", row.Id, err) } if err := qtx.UpdateMessagesSequence(ctx, c.id, row.Sequence); err != nil { return fmt.Errorf( "shift sequences after %d: %w", row.Sequence, err, ) } if err := tx.Commit(); err != nil { return fmt.Errorf("commit message tx: %w", err) } return nil } // AddToolCall appends a tool call to a saved assistant message. // mId identifies that assistant; updatedAt is the message's // updated_at from the caller's last read; a mismatch returns // ErrRevisionConflict. This edit does not execute the tool: it creates // no execution row and no result. It returns ErrBusy while Inference is // ongoing. func (c *Conversation) AddToolCall( ctx context.Context, mId string, updatedAt time.Time, call llm.ToolCall, ) (*db.Message, error) { c.mu.Lock() defer c.mu.Unlock() if c.cancel != nil { return nil, ErrBusy } row, err := c.dbr.SelectConversationMessageId(ctx, c.id, mId) if err != nil { return nil, fmt.Errorf("load message %q: %w", mId, err) } // Refuse to update a message that has been changed. if !row.UpdatedAt.Equal(updatedAt) { return nil, ErrRevisionConflict } if row.Role != db.MessageRoleAssistant { return nil, fmt.Errorf("invalid role: %q", row.Role) } var message llm.Message if err := json.Unmarshal( row.Content, &message, ); err != nil { return nil, fmt.Errorf( "unmarshal message %q: %w", row.Id, err, ) } if err := message.AddToolCall(call); err != nil { return nil, fmt.Errorf("message %q: %w", row.Id, err) } content, err := json.Marshal(message) if err != nil { return nil, fmt.Errorf("marshal updated message: %w", err) } tx, err := c.db.Writer.BeginTx(ctx, nil) if err != nil { return nil, fmt.Errorf("begin tx: %w", err) } defer tx.Rollback() qtx := c.dbw.WithTx(tx) updated, err := qtx.UpdateMessageContent( ctx, db.UpdateMessageContentParams{ Id: row.Id, Content: content, }, ) if err != nil { return nil, fmt.Errorf("update message %q: %w", row.Id, err) } if err := tx.Commit(); err != nil { return nil, fmt.Errorf("commit message tx: %w", err) } return updated, nil } // UpdateToolCall replaces a call in a saved assistant message. mId // identifies the message; callId identifies the call within it. The // replacement must keep the existing call ID: the call's execution // record and any saved result are keyed by it. updatedAt is the // assistant message's updated_at from the caller's last read; a mismatch // returns ErrRevisionConflict. It returns ErrBusy while Inference is // ongoing. func (c *Conversation) UpdateToolCall( ctx context.Context, mId string, callId string, updatedAt time.Time, call llm.ToolCall, ) (*db.Message, error) { c.mu.Lock() defer c.mu.Unlock() if c.cancel != nil { return nil, ErrBusy } row, err := c.dbr.SelectConversationMessageId(ctx, c.id, mId) if err != nil { return nil, fmt.Errorf("load message %q: %w", mId, err) } // Refuse to update a message that has been changed. if !row.UpdatedAt.Equal(updatedAt) { return nil, ErrRevisionConflict } // Validate the role update. if row.Role != db.MessageRoleAssistant { return nil, fmt.Errorf("invalid role: %q", row.Role) } var message llm.Message if err := json.Unmarshal( row.Content, &message, ); err != nil { return nil, fmt.Errorf( "unmarshal message %q: %w", row.Id, err, ) } if err := message.UpdateToolCall(call); err != nil { return nil, fmt.Errorf("message %q: %w", row.Id, err) } content, err := json.Marshal(message) if err != nil { return nil, fmt.Errorf("marshal updated message: %w", err) } tx, err := c.db.Writer.BeginTx(ctx, nil) if err != nil { return nil, fmt.Errorf("begin tx: %w", err) } defer tx.Rollback() qtx := c.dbw.WithTx(tx) updated, err := qtx.UpdateMessageContent( ctx, db.UpdateMessageContentParams{ Id: row.Id, Content: content, }, ) if err != nil { return nil, fmt.Errorf("update message %q: %w", row.Id, err) } if err := tx.Commit(); err != nil { return nil, fmt.Errorf("commit message tx: %w", err) } return updated, nil } // DeleteToolCall removes a call from a saved assistant message, deleting // the call's saved result and execution record with it. mId identifies // the parent assistant. updatedAt is the assistant's updated_at from the // caller's last read; a mismatch returns ErrRevisionConflict. It returns // ErrBusy while Inference is ongoing. func (c *Conversation) DeleteToolCall( ctx context.Context, mId string, callId string, updatedAt time.Time, ) error { c.mu.Lock() defer c.mu.Unlock() if c.cancel != nil { return ErrBusy } row, err := c.dbr.SelectConversationMessageId(ctx, c.id, mId) if err != nil { return fmt.Errorf("load assistant %q: %w", mId, err) } // Refuse to delete if the assistant has been changed. if !row.UpdatedAt.Equal(updatedAt) { return ErrRevisionConflict } if row.Role != db.MessageRoleAssistant { return fmt.Errorf("message %q is not an assistant", mId) } var message llm.Message if err := json.Unmarshal( row.Content, &message, ); err != nil { return fmt.Errorf( "unmarshal assistant %q: %w", row.Id, err, ) } var hasCall bool for _, call := range message.ToolCalls { if call.ID == callId { hasCall = true break } } if !hasCall { return fmt.Errorf( "assistant %q has no call %q", row.Id, callId, ) } // Remove the call from the assistant's content. calls := make([]llm.ToolCall, 0, len(message.ToolCalls)-1) for _, call := range message.ToolCalls { if call.ID != callId { calls = append(calls, call) } } message.ToolCalls = calls content, err := json.Marshal(message) if err != nil { return fmt.Errorf("marshal assistant: %w", err) } // Delete the call, its result, and its execution record together. tx, err := c.db.Writer.BeginTx(ctx, nil) if err != nil { return fmt.Errorf("begin tx: %w", err) } defer tx.Rollback() qtx := c.dbw.WithTx(tx) sequence, err := qtx.SelectMessageSequenceToolResult( ctx, c.id, row.Sequence, ) if err != nil { return fmt.Errorf("find tool result sequence: %w", err) } result, err := qtx.SelectConversationToolResult( ctx, c.id, row.Sequence, sequence, callId, ) if err != nil && !errors.Is(err, sql.ErrNoRows) { return fmt.Errorf("load tool result %q: %w", callId, err) } if _, err := qtx.UpdateMessageContent(ctx, db.UpdateMessageContentParams{ Id: row.Id, Content: content, }); err != nil { return fmt.Errorf("update assistant %q: %w", row.Id, err) } if result != nil { if err := qtx.DeleteMessageId(ctx, result.Id); err != nil { return fmt.Errorf("delete result %q: %w", result.Id, err) } if err := qtx.UpdateMessagesSequence( ctx, c.id, result.Sequence, ); err != nil { return fmt.Errorf( "shift sequences after %d: %w", result.Sequence, err, ) } } if err := qtx.DeleteExecutionMessageIdToolCallId( ctx, row.Id, callId, ); err != nil { return fmt.Errorf( "delete execution for call %q: %w", callId, err, ) } if err := tx.Commit(); err != nil { return fmt.Errorf("commit tool call tx: %w", err) } return nil } // AddToolResult saves textual output for a call in the assistant message // mId, right after that assistant's last saved tool result; later // messages shift down by one. It does not execute the tool, and it fails // if the call already has a result. updatedAt is the assistant message's // updated_at from the caller's last read; a mismatch returns // ErrRevisionConflict. The call's execution is marked completed, or // created as completed if missing. It returns ErrBusy while Inference is // ongoing. func (c *Conversation) AddToolResult( ctx context.Context, mId string, callId string, updatedAt time.Time, content string, ) (*db.Message, error) { c.mu.Lock() defer c.mu.Unlock() if c.cancel != nil { return nil, ErrBusy } row, err := c.dbr.SelectConversationMessageId(ctx, c.id, mId) if err != nil { return nil, fmt.Errorf("load assistant %q: %w", mId, err) } // Refuse to add a result if the assistant has been changed. if !row.UpdatedAt.Equal(updatedAt) { return nil, ErrRevisionConflict } if row.Role != db.MessageRoleAssistant { return nil, fmt.Errorf("message %q is not an assistant", mId) } var message llm.Message if err := json.Unmarshal( row.Content, &message, ); err != nil { return nil, fmt.Errorf( "unmarshal assistant %q: %w", row.Id, err, ) } var call *llm.ToolCall for i := range message.ToolCalls { if message.ToolCalls[i].ID == callId { call = &message.ToolCalls[i] break } } if call == nil { return nil, fmt.Errorf( "assistant %q has no call %q", row.Id, callId, ) } // Build the tool result message. result := llm.Message{ Role: llm.RoleTool, Name: call.Function.Name, ToolCallID: call.ID, ContentParts: []llm.ContentPart{{ Type: llm.ContentTypeText, Text: content, }}, } if err := result.Validate(); err != nil { return nil, fmt.Errorf("invalid tool result: %w", err) } encoded, err := json.Marshal(result) if err != nil { return nil, fmt.Errorf( "marshal tool result %q: %w", callId, err, ) } // Find the position, free it, insert the result, and complete the // execution together, so the position and duplicate check stay valid. tx, err := c.db.Writer.BeginTx(ctx, nil) if err != nil { return nil, fmt.Errorf("begin tx: %w", err) } defer tx.Rollback() qtx := c.dbw.WithTx(tx) sequence, err := qtx.SelectMessageSequenceToolResult( ctx, c.id, row.Sequence, ) if err != nil { return nil, fmt.Errorf("find tool result sequence: %w", err) } if _, err := qtx.SelectConversationToolResult( ctx, c.id, row.Sequence, sequence, callId, ); err == nil { return nil, fmt.Errorf("call %q already has a result", callId) } else if !errors.Is(err, sql.ErrNoRows) { return nil, fmt.Errorf("check tool result %q: %w", callId, err) } if err := qtx.UpdateMessagesSequenceIncrement( ctx, c.id, sequence, ); err != nil { return nil, fmt.Errorf( "shift sequences from %d: %w", sequence, err, ) } msg, err := qtx.InsertMessage(ctx, db.InsertMessageParams{ Content: encoded, ConversationId: c.id, Role: db.MessageRoleTool, Sequence: sequence, }) if err != nil { return nil, fmt.Errorf("save tool result %q: %w", callId, err) } if err := qtx.InsertExecutionCompleted( ctx, row.Id, callId, ); err != nil { return nil, fmt.Errorf("complete execution %q: %w", callId, err) } if err := tx.Commit(); err != nil { return nil, fmt.Errorf("commit tool result tx: %w", err) } return msg, nil } // UpdateToolResult replaces the saved textual output for a call in the // assistant message mId. Unlike AddToolResult, updatedAt is the tool // result message's updated_at (not the assistant message's); a mismatch // returns ErrRevisionConflict. It does not execute the tool, and fails // if the call has no result. The execution is marked completed, or // created as completed if missing, settling an interrupted call's // running row. It returns ErrBusy while Inference is ongoing. func (c *Conversation) UpdateToolResult( ctx context.Context, mId string, callId string, updatedAt time.Time, content string, ) (*db.Message, error) { c.mu.Lock() defer c.mu.Unlock() if c.cancel != nil { return nil, ErrBusy } row, err := c.dbr.SelectConversationMessageId(ctx, c.id, mId) if err != nil { return nil, fmt.Errorf("load assistant %q: %w", mId, err) } if row.Role != db.MessageRoleAssistant { return nil, fmt.Errorf("message %q is not an assistant", mId) } var message llm.Message if err := json.Unmarshal( row.Content, &message, ); err != nil { return nil, fmt.Errorf( "unmarshal assistant %q: %w", row.Id, err, ) } var call *llm.ToolCall for i := range message.ToolCalls { if message.ToolCalls[i].ID == callId { call = &message.ToolCalls[i] break } } if call == nil { return nil, fmt.Errorf( "assistant %q has no call %q", row.Id, callId, ) } // Build the replacement tool result message. result := llm.Message{ Role: llm.RoleTool, Name: call.Function.Name, ToolCallID: call.ID, ContentParts: []llm.ContentPart{{ Type: llm.ContentTypeText, Text: content, }}, } if err := result.Validate(); err != nil { return nil, fmt.Errorf("invalid tool result: %w", err) } encoded, err := json.Marshal(result) if err != nil { return nil, fmt.Errorf( "marshal tool result %q: %w", callId, err, ) } // Load and update the result, and complete the execution // together, so the revision check stays valid. tx, err := c.db.Writer.BeginTx(ctx, nil) if err != nil { return nil, fmt.Errorf("begin tx: %w", err) } defer tx.Rollback() qtx := c.dbw.WithTx(tx) sequence, err := qtx.SelectMessageSequenceToolResult( ctx, c.id, row.Sequence, ) if err != nil { return nil, fmt.Errorf("find tool result sequence: %w", err) } saved, err := qtx.SelectConversationToolResult( ctx, c.id, row.Sequence, sequence, callId, ) if errors.Is(err, sql.ErrNoRows) { return nil, fmt.Errorf("call %q has no result", callId) } if err != nil { return nil, fmt.Errorf("load tool result %q: %w", callId, err) } // Refuse to update a result that has been changed. if !saved.UpdatedAt.Equal(updatedAt) { return nil, ErrRevisionConflict } updated, err := qtx.UpdateMessageContent( ctx, db.UpdateMessageContentParams{ Id: saved.Id, Content: encoded, }, ) if err != nil { return nil, fmt.Errorf("update tool result %q: %w", callId, err) } if err := qtx.InsertExecutionCompleted( ctx, row.Id, callId, ); err != nil { return nil, fmt.Errorf("complete execution %q: %w", callId, err) } if err := tx.Commit(); err != nil { return nil, fmt.Errorf("commit tool result tx: %w", err) } return updated, nil } // Delete removes the saved conversation and all its messages and // execution records. // If inference is ongoing, it is canceled. func (c *Conversation) Delete(ctx context.Context) error { c.mu.Lock() defer c.mu.Unlock() if c.cancel != nil { c.cancel() } err := c.dbw.DeleteConversationId(ctx, c.id) if err != nil { return fmt.Errorf("delete conversation: %w", err) } return nil }