Agent runtime: majordomo in-process, Ollama Cloud config, chat endpoint (#56) (#70)

Co-authored-by: Steve Dudenhoeffer <[email protected]>

internal/store/agent.go

@@ -0,0 +1,95 @@
+package store
+
+import (
+	"context"
+	"fmt"
+
+	"gitea.stevedudenhoeffer.com/steve/pansy/internal/domain"
+)
+
+// agentMessageColumns lists agent_messages columns in scan order.
+const agentMessageColumns = `id, conversation_id, role, body, change_set_id, created_at`
+
+// EnsureConversation returns the (garden, user) conversation id, creating it on
+// first use. One thread per person per garden: two people editing a shared
+// garden hold separate dialogues, and merging them would put words in someone's
+// mouth.
+func (d *DB) EnsureConversation(ctx context.Context, gardenID, userID int64) (int64, error) {
+	var id int64
+	err := d.sql.QueryRowContext(ctx,
+		`SELECT id FROM agent_conversations WHERE garden_id = ? AND user_id = ?`,
+		gardenID, userID).Scan(&id)
+	if err == nil {
+		return id, nil
+	}
+	if err := d.sql.QueryRowContext(ctx,
+		`INSERT INTO agent_conversations (garden_id, user_id) VALUES (?, ?)
+		 ON CONFLICT (garden_id, user_id) DO UPDATE SET updated_at = updated_at
+		 RETURNING id`,
+		gardenID, userID).Scan(&id); err != nil {
+		return 0, fmt.Errorf("store: ensure conversation: %w", err)
+	}
+	return id, nil
+}
+
+// AppendAgentMessage records one side of an exchange.
+func (d *DB) AppendAgentMessage(ctx context.Context, m *domain.AgentMessage) (*domain.AgentMessage, error) {
+	var out domain.AgentMessage
+	if err := d.sql.QueryRowContext(ctx,
+		`INSERT INTO agent_messages (conversation_id, role, body, change_set_id)
+		 VALUES (?, ?, ?, ?)
+		 RETURNING `+agentMessageColumns,
+		m.ConversationID, m.Role, m.Body, m.ChangeSetID,
+	).Scan(&out.ID, &out.ConversationID, &out.Role, &out.Body, &out.ChangeSetID, &out.CreatedAt); err != nil {
+		return nil, fmt.Errorf("store: append agent message: %w", err)
+	}
+	// Touch the conversation so a "most recent thread" read is possible later.
+	if _, err := d.sql.ExecContext(ctx,
+		`UPDATE agent_conversations SET updated_at = strftime('%Y-%m-%dT%H:%M:%SZ', 'now') WHERE id = ?`,
+		m.ConversationID); err != nil {
+		return nil, fmt.Errorf("store: touch conversation: %w", err)
+	}
+	return &out, nil
+}
+
+// ListAgentMessages returns the last `limit` messages of a conversation in
+// chronological order.
+//
+// Reading the TAIL rather than the head is the point: an old opening exchange is
+// the least useful context for "now do the same to the north bed", and an
+// uncapped read would grow without bound over a season.
+func (d *DB) ListAgentMessages(ctx context.Context, conversationID int64, limit int) ([]domain.AgentMessage, error) {
+	rows, err := d.sql.QueryContext(ctx,
+		`SELECT `+agentMessageColumns+` FROM (
+		   SELECT `+agentMessageColumns+` FROM agent_messages
+		    WHERE conversation_id = ? ORDER BY id DESC LIMIT ?
+		 ) ORDER BY id`,
+		conversationID, limit)
+	if err != nil {
+		return nil, fmt.Errorf("store: list agent messages: %w", err)
+	}
+	defer rows.Close()
+
+	msgs := []domain.AgentMessage{}
+	for rows.Next() {
+		var m domain.AgentMessage
+		if err := rows.Scan(&m.ID, &m.ConversationID, &m.Role, &m.Body, &m.ChangeSetID, &m.CreatedAt); err != nil {
+			return nil, fmt.Errorf("store: scan agent message: %w", err)
+		}
+		msgs = append(msgs, m)
+	}
+	if err := rows.Err(); err != nil {
+		return nil, fmt.Errorf("store: iterate agent messages: %w", err)
+	}
+	return msgs, nil
+}
+
+// DeleteConversation clears a thread (the messages cascade).
+func (d *DB) DeleteConversation(ctx context.Context, gardenID, userID int64) error {
+	if _, err := d.sql.ExecContext(ctx,
+		`DELETE FROM agent_conversations WHERE garden_id = ? AND user_id = ?`,
+		gardenID, userID); err != nil {
+		return fmt.Errorf("store: delete conversation: %w", err)
+	}
+	return nil
+}