kenn-io--agentsview/cmd/agentsview/vector_push_source.go

// ABOUTME: pg push adapter exposing the local vectors.db active generation
// ABOUTME: to internal/postgres as a lazily-opened read-only VectorPushSource.
package main

import (
	"context"
	"fmt"
	"io"
	"log"
	"os"
	"sync"

	"go.kenn.io/agentsview/internal/config"
	"go.kenn.io/agentsview/internal/postgres"
	"go.kenn.io/agentsview/internal/vector"
)

// vectorPushSource adapts a read-only vector.Index to
// postgres.VectorPushSource, opening vectors.db lazily so a pg push whose
// target has vectors disabled never touches the file. Only a successful open is
// memoized so the three interface methods of one push share one connection; a
// missing file or open failure is never cached, so a daemon push that starts
// before embeddings exist picks up a build that lands afterward. Generation
// snapshots the active generation and SessionDocHashes/SessionDocs read the
// snapshotted ordinal, so a build that activates mid-push cannot pair this
// generation's fingerprint with a newer generation's docs; the next Generation
// call refreshes the snapshot, keeping successive daemon pushes current. The
// read-only handle stays open for the adapter's lifetime; the creator owns
// that lifetime and must release it via closeVectorPushSource — per push in
// PGPush, at loop exit in the watch path (which reuses one adapter across
// reconnects) — since postgres.Sync never closes its source.
type vectorPushSource struct {
	cfg config.Config

	mu sync.Mutex
	ix *vector.Index

	// snap/snapOK hold the active generation captured by the most recent
	// Generation call; SessionDocHashes and SessionDocs read snap.Ordinal.
	snap   vector.ActiveExport
	snapOK bool
}

// newVectorPushSource returns a lazy vectors.db push adapter, or nil when
// [vector] is disabled: there is then no vectors.db to open and nothing to
// push, and a nil source leaves postgres.Sync's vector phase skipped.
func newVectorPushSource(appCfg config.Config) postgres.VectorPushSource {
	if !appCfg.Vector.Enabled {
		return nil
	}
	return &vectorPushSource{cfg: appCfg}
}

// closeVectorPushSource releases a push source's memoized read-only
// vectors.db handle. postgres.Sync never closes its source — the creator
// owns the handle's lifetime — so every call site that builds a source must
// close it when the push (or watch loop) is done, or repeated pushes leak
// one SQLite handle each. Safe on a nil source and on sources holding no
// handle.
func closeVectorPushSource(src postgres.VectorPushSource) {
	c, ok := src.(io.Closer)
	if !ok {
		return
	}
	if err := c.Close(); err != nil {
		log.Printf("closing vectors.db push source: %v", err)
	}
}

// openIndex opens vectors.db read-only and memoizes only a successful open. A
// missing file (no build yet) and any open failure are NOT cached, so a later
// call — a daemon push that starts before embeddings exist — reopens and picks
// up a build that lands afterward. A nil index without error means the file is
// absent ("nothing to push"); the pointer's own nil checks gate every
// dereference.
func (s *vectorPushSource) openIndex(ctx context.Context) (*vector.Index, error) {
	s.mu.Lock()
	defer s.mu.Unlock()
	if s.ix != nil {
		return s.ix, nil
	}
	path := s.cfg.Vector.ResolvedDBPath(s.cfg.DataDir)
	if _, err := os.Stat(path); err != nil {
		if os.IsNotExist(err) {
			return nil, nil
		}
		return nil, fmt.Errorf("stat vectors.db: %w", err)
	}
	ix, err := vector.Open(ctx, path, true, s.cfg.Vector.Embeddings.MaxInputChars)
	if err != nil {
		return nil, fmt.Errorf("opening vectors.db: %w", err)
	}
	s.ix = ix
	return ix, nil
}

// Close releases the memoized read-only vectors.db handle, if one was opened.
// Safe to call multiple times; a later interface call reopens the file via
// openIndex's uncached-miss path.
func (s *vectorPushSource) Close() error {
	s.mu.Lock()
	defer s.mu.Unlock()
	if s.ix == nil {
		return nil
	}
	err := s.ix.Close()
	s.ix = nil
	return err
}

// setSnapshot records the active generation captured by Generation.
func (s *vectorPushSource) setSnapshot(exp vector.ActiveExport, ok bool) {
	s.mu.Lock()
	s.snap, s.snapOK = exp, ok
	s.mu.Unlock()
}

// snapshot returns the active generation captured by the most recent
// Generation call, and whether one is set.
func (s *vectorPushSource) snapshot() (vector.ActiveExport, bool) {
	s.mu.Lock()
	defer s.mu.Unlock()
	return s.snap, s.snapOK
}

// Generation implements postgres.VectorPushSource and snapshots the active
// generation for the rest of the push. Contract: Generation must be called
// before SessionDocHashes and SessionDocs, which read the snapshotted ordinal;
// the postgres push (pushVectors) calls Generation once first, then again as
// the pre-eviction safety re-check. Snapshotting here means a build that
// activates mid-push cannot mix this generation's fingerprint with a newer
// generation's docs, while the next Generation call refreshes the snapshot so
// successive pushes stay current.
//
// A generation with missing embeddings is refused with an error wrapping
// postgres.ErrVectorSourceNotReady rather than exported: the mirror only
// changes during builds, so missing coverage means a build is rewriting the
// generation right now (a same-fingerprint full rebuild clears and refills it
// in place) or one was interrupted. Exporting that partial view would evict
// or overwrite valid PG vectors; the push after the build completes sends
// everything that changed.
func (s *vectorPushSource) Generation(
	ctx context.Context,
) (postgres.VectorGenerationInfo, bool, error) {
	ix, err := s.openIndex(ctx)
	if err != nil || ix == nil {
		s.setSnapshot(vector.ActiveExport{}, false)
		return postgres.VectorGenerationInfo{}, false, err
	}
	exp, ok, err := ix.ActiveExport(ctx)
	if err != nil {
		s.setSnapshot(vector.ActiveExport{}, false)
		return postgres.VectorGenerationInfo{}, false,
			fmt.Errorf("reading active generation: %w", err)
	}
	if !ok {
		s.setSnapshot(vector.ActiveExport{}, false)
		return postgres.VectorGenerationInfo{}, false, nil
	}
	info, err := ix.GenerationByID(ctx, exp.Ordinal)
	if err != nil {
		s.setSnapshot(vector.ActiveExport{}, false)
		return postgres.VectorGenerationInfo{}, false,
			fmt.Errorf("reading active generation coverage: %w", err)
	}
	if info.Missing > 0 {
		s.setSnapshot(vector.ActiveExport{}, false)
		return postgres.VectorGenerationInfo{}, false, fmt.Errorf(
			"%w: %d document(s) pending",
			postgres.ErrVectorSourceNotReady, info.Missing)
	}
	s.setSnapshot(exp, true)
	return postgres.VectorGenerationInfo{
		Fingerprint: exp.Fingerprint,
		Model:       exp.Model,
		Dimension:   exp.Dimension,
	}, true, nil
}

// SessionDocHashes implements postgres.VectorPushSource, reading the ordinal
// snapshotted by Generation. An absent file or no active snapshot yields an
// empty (non-nil) map: no local sessions, so the push evicts any PG state this
// pusher previously owned.
func (s *vectorPushSource) SessionDocHashes(
	ctx context.Context,
) (map[string]string, error) {
	ix, err := s.openIndex(ctx)
	if err != nil {
		return nil, err
	}
	exp, ok := s.snapshot()
	if ix == nil || !ok {
		return map[string]string{}, nil
	}
	return ix.SessionEmbeddedDocHashes(ctx, exp.Ordinal)
}

// SessionDocs implements postgres.VectorPushSource, exporting one session's
// docs at the ordinal snapshotted by Generation and mapping each mirror doc and
// its chunk vectors field-by-field onto the push types. The returned hash is
// the aggregate of exactly the exported doc set, computed inside the export's
// own read snapshot; the push defers the session when it no longer matches the
// delta-scan hash.
func (s *vectorPushSource) SessionDocs(
	ctx context.Context, sessionID string,
) ([]postgres.VectorPushDoc, string, error) {
	ix, err := s.openIndex(ctx)
	if err != nil {
		return nil, "", err
	}
	exp, ok := s.snapshot()
	if ix == nil || !ok {
		return nil, "", nil
	}
	docs, aggHash, err := ix.ExportSessionDocs(ctx, exp.Ordinal, sessionID)
	if err != nil {
		return nil, "", err
	}
	return convertVectorPushDocs(docs), aggHash, nil
}

// convertVectorPushDocs copies exported mirror docs onto the backend-agnostic
// push types. It is a mechanical field-by-field copy, isolated here so the
// mapping is testable against the raw export API.
func convertVectorPushDocs(docs []vector.ExportDoc) []postgres.VectorPushDoc {
	out := make([]postgres.VectorPushDoc, len(docs))
	for i, d := range docs {
		chunks := make([]postgres.VectorPushChunk, len(d.Chunks))
		for j, c := range d.Chunks {
			chunks[j] = postgres.VectorPushChunk{
				ChunkIndex: c.ChunkIndex,
				Embedding:  c.Embedding,
			}
		}
		out[i] = postgres.VectorPushDoc{
			DocKey:      d.DocKey,
			SessionID:   d.SessionID,
			SourceUUID:  d.SourceUUID,
			Ordinal:     d.Ordinal,
			OrdinalEnd:  d.OrdinalEnd,
			Subordinate: d.Subordinate,
			OffsetsJSON: d.OffsetsJSON,
			Content:     d.Content,
			ContentHash: d.ContentHash,
			Chunks:      chunks,
		}
	}
	return out
}