Files
odidere/internal/conversation/conversation.go
T
2026-10-23 23:41:35 +00:00

1709 lines
46 KiB
Go

// 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.Inference] 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 run.
mu sync.Mutex
// cancel is used to stop inference.
// When set, it indicates inference is ongoing.
cancel context.CancelFunc
// run shares the ongoing Inference's events with subscribers.
// When set, it indicates inference is ongoing.
run *run
}
// 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.
// Err is nil on success.
type Done struct {
Err error `json:"err"`
}
func (Done) Type() EventType { return EventDone }
// run records events from a [Conversation.Inference] call.
// [Conversation.Subscribe] uses these events to send past and live updates to
// subscribers.
type run struct {
// mu serializes access to the other fields.
mu sync.Mutex
// 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
}
// finish marks the run complete and records err.
// It closes changed to wake waiting subscribers.
// The channel stays closed because the run is complete.
func (r *run) finish(err error) {
r.mu.Lock()
defer r.mu.Unlock()
r.done = true
r.err = err
close(r.changed)
}
// Subscribe streams events that synchronize a client's message cache with
// the saved conversation and any ongoing [Conversation.Inference] call.
//
// 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 [Conversation.Inference] is running when
// Subscribe loads the database, Subscribe streams the run's
// [MessageSaved] and [AssistantUpdate] events, then ends with [Done].
// If no inference is 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
// [Conversation.Inference] call.
func (c *Conversation) Subscribe(
ctx context.Context,
manifest []MessageManifest,
) (<-chan Event, error) {
// Check for an active Inference run.
c.mu.Lock()
run := c.run
c.mu.Unlock()
// Inference holds run.mu while it commits a message and appends the
// MessageSaved event to run.events. Holding run.mu across this read
// ensures that any message committed after this point will appear in
// run.events, and any message already in rows will be recorded in the
// diff's sent map.
if run != nil {
run.mu.Lock()
}
rows, err := c.dbr.SelectMessagesConversationId(ctx, c.id)
if run != nil {
run.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 run == nil {
return
}
// Replay and follow run.events from index 0.
pos := 0
for {
// Copy the slice header for unread events and the
// current notification channel. Because Inference
// only appends to run.events, reading batch after
// Unlock is safe.
run.mu.Lock()
batch := run.events[pos:]
pos = len(run.events)
wait, done, err := run.changed, run.done, run.err
run.mu.Unlock()
// Skip events for messages that the initial diff
// already sent. If the client's manifest already had
// a message from this run, it is not in sent and will
// be re-sent here; the client deduplicates by ID and
// UpdatedAt.
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 := sent[id]; ok {
continue
}
}
if !send(ctx, ch, e) {
return
}
}
if done {
send(ctx, ch, Done{Err: err})
return
}
// Wait for Inference to close wait (when it appends
// new events or finishes), or for the subscriber to
// cancel ctx.
select {
case <-wait:
case <-ctx.Done():
return
}
}
}()
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
}
// Inference 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.
func (c *Conversation) Inference(
ctx context.Context,
client *llm.Client,
req InferenceRequest,
) error {
// Reject invalid input before accepting work.
if client == nil {
return fmt.Errorf("missing llm client")
}
if err := req.Validate(); err != nil {
return 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)
}
// Reserve the conversation until work and save attempts finish.
// The run shares this call's events with subscribers.
r := &run{changed: make(chan struct{})}
c.mu.Lock()
if c.cancel != nil {
c.mu.Unlock()
cancel()
return ErrBusy
}
c.cancel = cancel
c.run = r
c.mu.Unlock()
// Every failing path below must assign this err, not shadow it:
// the deferred run.finish call reports it to subscribers.
var err error
defer func() {
// Every exit path, including errors, finishes the run.
r.finish(err)
cancel()
c.mu.Lock()
c.cancel = nil
c.run = nil
c.mu.Unlock()
}()
// 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
r.mu.Lock()
saved, err = c.insertMessage(
ctx, nil, *req.Message, nil, nil,
)
if err == nil {
r.events = append(r.events, MessageSaved{Message: *saved})
close(r.changed)
r.changed = make(chan struct{})
}
r.mu.Unlock()
if err != nil {
return fmt.Errorf("insert message: %w", err)
}
log.InfoContext(
ctx,
"persisted new user message",
slog.String("id", saved.Id),
)
// 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() {
r.mu.Lock()
r.events = append(r.events, AssistantUpdate{
MessageId: aid.String(),
Chunk: stream.Current(),
})
close(r.changed)
r.changed = make(chan struct{})
r.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.
r.mu.Lock()
row, err = c.insertMessage(
ctx, &aid, res.Message, &res.Metadata, nil,
)
if err == nil {
r.events = append(r.events, MessageSaved{Message: *row})
close(r.changed)
r.changed = make(chan struct{})
}
r.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.run is set for the whole of Inference, the only caller path to
// callTool.
c.mu.Lock()
r := c.run
c.mu.Unlock()
// Hold r.mu across the commit so a concurrent Subscribe's diff
// sees the row and its event together.
r.mu.Lock()
saved, err := c.insertMessage(saveCtx, nil, msg, nil, update)
if err == nil {
r.events = append(r.events, MessageSaved{Message: *saved})
close(r.changed)
r.changed = make(chan struct{})
}
r.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
}
return c.insertMessage(ctx, nil, msg, nil, 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
}