loadSessionItems loads all items for a session from session_items. Used both as the public read path (q == s.db) and recursively from inside loadSession when resolving sub-sessions inside a transaction.
(ctx context.Context, q querier, sessionID string)
| 715 | // Used both as the public read path (q == s.db) and recursively from inside |
| 716 | // loadSession when resolving sub-sessions inside a transaction. |
| 717 | func (s *SQLiteSessionStore) loadSessionItems(ctx context.Context, q querier, sessionID string) ([]Item, error) { |
| 718 | rows, err := q.QueryContext(ctx, |
| 719 | `SELECT position, item_type, agent_name, message_json, implicit, subsession_id, summary_text, COALESCE(first_kept_entry, 0) |
| 720 | FROM session_items WHERE session_id = ? ORDER BY position`, sessionID) |
| 721 | if err != nil { |
| 722 | return nil, err |
| 723 | } |
| 724 | defer rows.Close() |
| 725 | |
| 726 | // First, collect all raw row data so we can close the result set |
| 727 | // before making any recursive calls (SQLite doesn't allow concurrent queries) |
| 728 | var rawRows []sessionItemRow |
| 729 | for rows.Next() { |
| 730 | var row sessionItemRow |
| 731 | if err := rows.Scan(&row.position, &row.itemType, &row.agentName, &row.messageJSON, &row.implicit, &row.subsessionID, &row.summaryText, &row.firstKeptEntry); err != nil { |
| 732 | return nil, err |
| 733 | } |
| 734 | rawRows = append(rawRows, row) |
| 735 | } |
| 736 | if err := rows.Err(); err != nil { |
| 737 | return nil, err |
| 738 | } |
| 739 | |
| 740 | if len(rawRows) == 0 { |
| 741 | return nil, nil |
| 742 | } |
| 743 | |
| 744 | // Now process the collected rows, making recursive calls as needed |
| 745 | var items []Item |
| 746 | for _, row := range rawRows { |
| 747 | switch row.itemType { |
| 748 | case "message": |
| 749 | var chatMsg chat.Message |
| 750 | if err := json.Unmarshal([]byte(row.messageJSON.String), &chatMsg); err != nil { |
| 751 | return nil, fmt.Errorf("unmarshaling message at position %d: %w", row.position, err) |
| 752 | } |
| 753 | items = append(items, Item{ |
| 754 | Message: &Message{ |
| 755 | AgentName: row.agentName.String, |
| 756 | Message: chatMsg, |
| 757 | Implicit: row.implicit, |
| 758 | }, |
| 759 | }) |
| 760 | |
| 761 | case "subsession": |
| 762 | // Skip if subsession_id is NULL (can happen if the sub-session was deleted |
| 763 | // and the foreign key set the reference to NULL) |
| 764 | if !row.subsessionID.Valid || row.subsessionID.String == "" { |
| 765 | slog.WarnContext(ctx, "Skipping subsession item with NULL reference", "session_id", sessionID, "position", row.position) |
| 766 | continue |
| 767 | } |
| 768 | // Recursively load sub-session |
| 769 | subSession, err := s.loadSession(ctx, q, row.subsessionID.String) |
| 770 | if err != nil { |
| 771 | if errors.Is(err, ErrNotFound) { |
| 772 | // Sub-session was deleted but item reference remains (orphaned reference) |
| 773 | slog.WarnContext(ctx, "Skipping orphaned subsession reference", "session_id", sessionID, "subsession_id", row.subsessionID.String) |
| 774 | continue |
no test coverage detected