Maven/internal/store/factvectors.go

package store

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"strconv"
	"strings"
	"time"
)

// metaKeyFactVectorShape names the shape the stored fact vectors were written
// in. It exists so the repair below runs once per box instead of on every
// start: the rows it fixes were written by a code path that no longer exists,
// and once fixed nothing writes that shape again.
const metaKeyFactVectorShape = "fact_vector_shape"

// factVectorShapeFact is the shape FactRecallText produces. Anything else in
// the marker (including nothing, which is every box written before #493) means
// the fact vectors still hold utterances.
const factVectorShapeFact = "fact-text (#493)"

// FactVectorRepair is what one repair run did, for logging.
type FactVectorRepair struct {
	Skipped   bool // marker already matched — nothing to do
	Rewritten int  // rows re-embedded from the fact they name
	Dropped   int  // rows deleted: voided, superseded, or naming no fact at all
	Kept      int  // rows already holding the right text
	Took      time.Duration
}

// RepairFactVectors brings the fact rows of memory_vectors in line with the
// facts they name, and is the operator recovery a poisoned box had no path to
// (#470 point 4, #493).
//
// Three defects put wrong text in that index, and all three are write-path
// fixes that do nothing for rows already stored:
//
//   - the indexed text was the utterance, so every fact row reads back a
//     sentence rather than a value;
//   - a void left its vector behind, so retracted junk kept answering;
//   - a correction left its vector behind, so the superseded value did.
//
// So each fact row is resolved against the fact store and one of three things
// happens. It is dropped when the key has no fact, when the newest row for the
// key is a void marker, or when a newer vector for the same key exists — a
// superseded value has no business claiming a turn. It is re-embedded when its
// text is not what FactRecallText says the fact is. Otherwise it is left alone.
//
// Idempotent, and safe to interrupt: every step compares before writing and the
// marker is written last, so a run that dies partway is simply redone.
func (s *Store) RepairFactVectors(ctx context.Context, embed EmbedFunc) (FactVectorRepair, error) {
	start := time.Now()
	var res FactVectorRepair

	shape, err := s.Meta(ctx, metaKeyFactVectorShape)
	if err != nil {
		return res, err
	}
	if shape == factVectorShapeFact {
		res.Skipped = true
		res.Took = time.Since(start)
		return res, nil
	}

	rows, err := s.db.QueryContext(ctx, `SELECT id, meta FROM memory_vectors`)
	if err != nil {
		return res, fmt.Errorf("repair fact vectors: read: %w", err)
	}
	type factVec struct {
		id, key string
		meta    map[string]string
		ts      int64
	}
	var vecs []factVec
	newest := map[string]int64{} // key → newest ts seen for it
	for rows.Next() {
		var id, metaJSON string
		if err := rows.Scan(&id, &metaJSON); err != nil {
			rows.Close()
			return res, fmt.Errorf("repair fact vectors: row: %w", err)
		}
		meta := map[string]string{}
		if err := json.Unmarshal([]byte(metaJSON), &meta); err != nil {
			rows.Close()
			return res, fmt.Errorf("repair fact vectors: meta for %q: %w", id, err)
		}
		if meta["type"] != "fact" {
			continue
		}
		key, ts, ok := splitFactVectorID(id)
		if !ok {
			continue
		}
		vecs = append(vecs, factVec{id: id, key: key, meta: meta, ts: ts})
		if ts > newest[key] {
			newest[key] = ts
		}
	}
	rows.Close()
	if err := rows.Err(); err != nil {
		return res, fmt.Errorf("repair fact vectors: rows: %w", err)
	}

	for _, v := range vecs {
		drop := v.ts < newest[v.key]
		var want string
		if !drop {
			f, ferr := s.LatestFact(ctx, v.key)
			switch {
			case errors.Is(ferr, ErrNoFact):
				drop = true
			case ferr != nil:
				return res, fmt.Errorf("repair fact vectors: fact %q: %w", v.key, ferr)
			case DecodeFactValue(f.Value) == "voided":
				drop = true
			default:
				want = FactRecallText(v.key, f.Value)
			}
		}
		if drop {
			if err := s.VectorMemory().Delete(ctx, v.id); err != nil {
				return res, err
			}
			res.Dropped++
			continue
		}
		if v.meta["text"] == want {
			res.Kept++
			continue
		}
		vec, err := embed(ctx, want)
		if err != nil {
			return res, fmt.Errorf("repair fact vectors: embed %q: %w", v.id, err)
		}
		// The whole meta blob is rewritten in Go rather than patched in SQL,
		// because json_set needs the JSON1 extension and this store is opened
		// through sqlcipher.
		v.meta["text"] = want
		metaJSON, err := json.Marshal(v.meta)
		if err != nil {
			return res, fmt.Errorf("repair fact vectors: meta %q: %w", v.id, err)
		}
		if _, err := s.db.ExecContext(ctx,
			`UPDATE memory_vectors SET vec = ?, meta = ? WHERE id = ?`,
			encodeVec(vec), string(metaJSON), v.id); err != nil {
			return res, fmt.Errorf("repair fact vectors: write %q: %w", v.id, err)
		}
		res.Rewritten++
	}

	if err := s.SetMeta(ctx, metaKeyFactVectorShape, factVectorShapeFact); err != nil {
		return res, err
	}
	res.Took = time.Since(start)
	return res, nil
}

// splitFactVectorID reads the key and write time back out of a fact vector's
// id, which the write path builds as `fact:<key>:<unix>`. A key may hold a
// colon, the timestamp may not, so the split is from the right.
func splitFactVectorID(id string) (key string, ts int64, ok bool) {
	rest, found := strings.CutPrefix(id, "fact:")
	if !found {
		return "", 0, false
	}
	cut := strings.LastIndex(rest, ":")
	if cut <= 0 {
		return "", 0, false
	}
	ts, err := strconv.ParseInt(rest[cut+1:], 10, 64)
	if err != nil {
		return "", 0, false
	}
	return rest[:cut], ts, true
}