kenn-io--agentsview/internal/postgres/search_content_hybrid.go

package postgres

import (
	"context"
	"fmt"

	"go.kenn.io/agentsview/internal/db"
)

// pgHybridDisplay carries what one fused unit needs for presentation: the
// anchor (session, ordinal) the match reports and enriches by, the unit's
// ordinal span and subordinate flag (structurally derived for a unit-less
// message-granularity keyword hit, see classifyUnitlessHybridHitsPG), plus the
// leg's raw (unredacted) approximate snippet text used only to center the
// redacted window. It is the PG twin of internal/db's hybridDisplay.
type pgHybridDisplay struct {
	sessionID    string
	ordinal      int
	ordinalStart int
	ordinalEnd   int
	subordinate  bool
	snippet      string
}

// pgHybridLeg is one rank-ordered fusion leg: entries for db.RRFMerge plus each
// key's display info.
type pgHybridLeg struct {
	ranked  []db.RankedUnit
	display map[string]pgHybridDisplay
}

// searchContentHybridPG runs mode "hybrid" on the PostgreSQL store, mirroring
// internal/db.searchContentHybrid with the keyword-leg substitution PG uses
// (ILIKE candidates ordered by recency instead of BM25 rank). The vector and
// keyword rankings are each over-fetched to k, the vector leg filtered down to
// sessions passing the filter's metadata scope, keyword message hits resolved
// to their containing units, and the two rank-ordered leg lists fused at unit
// granularity with db.RRFMerge. Returned matches are enriched with
// session/message metadata via the same lookup semantic search uses, ordered
// by fused score descending, truncated to f.Limit. The caller (SearchContent)
// has already run db.ValidateSemanticFilter and confirmed a searcher is wired.
func (s *Store) searchContentHybridPG(
	ctx context.Context, f db.ContentSearchFilter,
) (db.ContentSearchPage, error) {
	searcher := s.getVectorSearcher()
	if searcher == nil {
		return db.ContentSearchPage{}, s.semanticUnavailableError()
	}

	k := max(f.Limit*4, db.SemanticOverfetchMin)
	vecLeg, err := s.hybridVectorLegPG(ctx, f, searcher, k)
	if err != nil {
		return db.ContentSearchPage{}, err
	}
	kwLeg, err := s.hybridKeywordLegPG(ctx, f, searcher, k)
	if err != nil {
		return db.ContentSearchPage{}, err
	}
	if len(vecLeg.ranked) == 0 && len(kwLeg.ranked) == 0 {
		return db.ContentSearchPage{}, nil
	}

	merged := db.RRFMerge([][]db.RankedUnit{vecLeg.ranked, kwLeg.ranked}, f.Limit)
	return s.enrichHybridMatchesPG(ctx, f, merged, vecLeg.display, kwLeg.display)
}

// hybridVectorLegPG over-fetches k semantic unit hits, drops any whose session
// fails the filter's metadata scope or whose subordinate flag falls outside
// f.Scope (via survivingVectorHitsPG, shared with the semantic mode), and
// returns the survivors as a rank-ordered fusion leg keyed by unit. Both
// filters run before the merge so reciprocal-rank fusion only ranks eligible
// units and a scoped search can still fill the limit.
func (s *Store) hybridVectorLegPG(
	ctx context.Context, f db.ContentSearchFilter, searcher db.VectorSearcher, k int,
) (pgHybridLeg, error) {
	leg := pgHybridLeg{display: make(map[string]pgHybridDisplay)}
	surviving, err := s.survivingVectorHitsPG(ctx, f, searcher, k)
	if err != nil {
		return pgHybridLeg{}, err
	}
	for _, h := range surviving {
		key := db.UnitFusionKey(h.SessionID, h.OrdinalStart)
		if _, seen := leg.display[key]; seen {
			continue
		}
		leg.ranked = append(leg.ranked, db.RankedUnit{Key: key, Subordinate: h.Subordinate})
		leg.display[key] = pgHybridDisplay{
			sessionID: h.SessionID, ordinal: h.Ordinal, snippet: h.Snippet,
			ordinalStart: h.OrdinalStart, ordinalEnd: h.OrdinalEnd,
			subordinate: h.Subordinate,
		}
	}
	return leg, nil
}

// maxHybridKeywordBatches caps how many k-row ILIKE batches the keyword leg
// fetches. It is the PG twin of internal/db.maxHybridFTSBatches (unexported
// there); defined locally so this diff stays self-contained. It bounds the
// worst-case work when discard dominates — many rows collapsing into one unit,
// or a narrow f.Scope dropping most rows — while letting the leg keep paging
// past discarded rows instead of under-filling after the first batch.
const maxHybridKeywordBatches = 4

// hybridKeywordLegPG runs a recency-ordered ILIKE query over the embedded
// universe (role user/assistant, is_system = FALSE, system-prefix excluded),
// scoped in SQL to sessions passing the child-exclusion-lifted filter (so both
// hybrid legs see the same universe), resolves each message hit to its
// containing unit, drops units outside f.Scope, and returns up to k hits as a
// rank-ordered fusion leg. A hit inside a unit adopts the unit's fusion key and
// subordinate flag while keeping its own message ordinal as the anchor and its
// keyword snippet for display; several hits in one unit collapse to the
// best-ranked one. A hit with no containing unit keeps a message-granularity
// key and survives fusion on its own, with its range and subordinate flag
// structurally derived before the scope filter and merge. Rows are fetched in
// recency-ordered batches of k with OFFSET continuation: collapse and scope
// filtering can discard most of a batch, so the leg keeps fetching until it
// holds k entries, the stream is exhausted, or maxHybridKeywordBatches is hit.
func (s *Store) hybridKeywordLegPG(
	ctx context.Context, f db.ContentSearchFilter, searcher db.VectorSearcher, k int,
) (pgHybridLeg, error) {
	leg := pgHybridLeg{display: make(map[string]pgHybridDisplay, k)}
	for batch := range maxHybridKeywordBatches {
		hits, err := s.fetchHybridKeywordBatchPG(ctx, f, k, batch*k)
		if err != nil {
			return pgHybridLeg{}, err
		}
		if err := s.appendHybridKeywordHitsPG(ctx, searcher, f, hits, &leg); err != nil {
			return pgHybridLeg{}, err
		}
		if len(hits) < k || len(leg.ranked) >= k {
			break
		}
	}
	return leg, nil
}

// fetchHybridKeywordBatchPG fetches one recency-ordered batch of at most k
// keyword message rows, starting at offset. The term-AND ILIKE predicate
// (pgContentSearchPredicate with Mode "fts") preserves SQLite's implicit-AND
// FTS-term semantics; recency ordering is the documented PG keyword ranking (no
// BM25), with (session_id, ordinal) as a deterministic tiebreak so OFFSET
// continuation is stable across batches. The display snippet is windowed in Go
// around the first case-insensitive pattern occurrence; it centers the redacted
// window only, since redaction (SemanticSnippet) always runs on full content.
func (s *Store) fetchHybridKeywordBatchPG(
	ctx context.Context, f db.ContentSearchFilter, k, offset int,
) ([]pgHybridDisplay, error) {
	scopeWhere, scopeArgs := buildPGSessionBaseFilter(semanticPGSessionFilter(f))
	pb := &paramBuilder{n: len(scopeArgs), args: append([]any{}, scopeArgs...)}

	kf := f
	kf.Mode = "fts"
	contentPred := pgContentSearchPredicate("m.content", kf, escapeLike(f.Pattern), pb)
	limitP := pb.add(k)
	offsetP := pb.add(offset)

	query := fmt.Sprintf(`
		SELECT m.session_id, m.ordinal, m.content
		FROM messages m
		JOIN sessions s ON s.id = m.session_id
		WHERE %s
		  AND m.role IN ('user', 'assistant') AND m.is_system = FALSE
		  AND %s
		  AND m.session_id IN (SELECT id FROM sessions WHERE %s)
		ORDER BY COALESCE(s.ended_at, s.started_at, s.created_at) DESC NULLS LAST,
		         m.session_id ASC, m.ordinal ASC
		LIMIT %s OFFSET %s`,
		contentPred, db.PostgresSystemPrefixSQL("m.content", "m.role"),
		scopeWhere, limitP, offsetP)

	rows, err := s.pg.QueryContext(ctx, query, pb.args...)
	if err != nil {
		return nil, fmt.Errorf("pg hybrid keyword leg: %w", err)
	}
	defer rows.Close()

	var hits []pgHybridDisplay
	for rows.Next() {
		var hit pgHybridDisplay
		var content string
		if err := rows.Scan(&hit.sessionID, &hit.ordinal, &content); err != nil {
			return nil, fmt.Errorf("scan hybrid keyword hit: %w", err)
		}
		hit.snippet = pgKeywordApproxSnippet(content, f.Pattern)
		hits = append(hits, hit)
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return hits, nil
}

// pgKeywordApproxSnippet windows content around the pattern's term-aware span
// (db.FTSSnippetRange, the same helper the fts snippet path uses via
// pgSubstringSnippet), rune-snapped via pgSnippetBounds. Centering on the raw
// pattern would fall back to the message start for quoted phrases or multi-term
// queries whose pattern is not a literal substring; FTSSnippetRange instead
// locates the de-quoted phrase, then the first matched term, then the start. The
// result is the raw, unredacted centering text SemanticSnippet later locates in
// the full content; redaction always runs on the full content, not this window.
func pgKeywordApproxSnippet(content, pattern string) string {
	start, end := db.FTSSnippetRange(pattern, content)
	lo, hi := pgSnippetBounds(content, start, end)
	return content[lo:hi]
}

// appendHybridKeywordHitsPG resolves one batch of keyword message hits to their
// containing units, classifies unit-less hits structurally (range and
// subordinate flag, so the scope filter and fusion penalty treat them exactly
// like lexical mode), and accumulates the survivors into leg: hits outside
// scope are dropped, and a unit already seen (within or across batches) keeps
// its earlier, better-ranked entry.
func (s *Store) appendHybridKeywordHitsPG(
	ctx context.Context, searcher db.VectorSearcher, f db.ContentSearchFilter,
	hits []pgHybridDisplay, leg *pgHybridLeg,
) error {
	if len(hits) == 0 {
		return nil
	}
	refs := make([]db.MessageRef, len(hits))
	for i, hit := range hits {
		refs[i] = db.MessageRef{SessionID: hit.sessionID, Ordinal: hit.ordinal}
	}
	units, err := searcher.ResolveMessageUnits(ctx, refs)
	if err != nil {
		return fmt.Errorf("resolving keyword hits to units: %w", err)
	}
	if len(units) != len(refs) {
		return fmt.Errorf(
			"resolving keyword hits to units: got %d units for %d refs", len(units), len(refs))
	}

	keys := make([]string, len(hits))
	var unitless []int
	for i := range hits {
		hit := &hits[i]
		keys[i] = db.MessageFusionKey(hit.sessionID, hit.ordinal)
		hit.ordinalStart, hit.ordinalEnd = hit.ordinal, hit.ordinal
		if units[i].DocKey == "" {
			unitless = append(unitless, i)
			continue
		}
		keys[i] = db.UnitFusionKey(units[i].SessionID, units[i].OrdinalStart)
		hit.ordinalStart = units[i].OrdinalStart
		hit.ordinalEnd = units[i].OrdinalEnd
		hit.subordinate = units[i].Subordinate
	}
	if err := s.classifyUnitlessHybridHitsPG(ctx, hits, unitless); err != nil {
		return err
	}

	for i, hit := range hits {
		if db.ScopeExcludes(f.Scope, hit.subordinate) {
			continue
		}
		if _, seen := leg.display[keys[i]]; seen {
			continue
		}
		leg.ranked = append(leg.ranked, db.RankedUnit{Key: keys[i], Subordinate: hit.subordinate})
		leg.display[keys[i]] = hit
	}
	return nil
}

// classifyUnitlessHybridHitsPG assigns each unit-less keyword hit (the resolver
// returned no mirror unit; hits[i] for i in idxs) its structurally derived
// conversation-unit range and subordinate flag BEFORE scope filtering and
// fusion, so a unit-less sidechain (or subagent/fork-session) hit is excluded,
// included, and penalized exactly like lexical mode classifies the same anchor.
// It reuses fillAnchorMetaPG and db.DeriveUnitRanges + db.SubordinateSession,
// mirroring internal/db.classifyUnitlessHybridHits.
func (s *Store) classifyUnitlessHybridHitsPG(
	ctx context.Context, hits []pgHybridDisplay, idxs []int,
) error {
	if len(idxs) == 0 {
		return nil
	}
	matches := make([]db.ContentMatch, len(idxs))
	for k, i := range idxs {
		matches[k] = db.ContentMatch{SessionID: hits[i].sessionID, Ordinal: hits[i].ordinal}
	}
	metas, err := s.fillAnchorMetaPG(ctx, matches)
	if err != nil {
		return err
	}
	anchors := make([]db.UnitAnchor, len(idxs))
	for k := range idxs {
		meta := metas[k]
		anchors[k] = db.UnitAnchor{
			SessionID:  matches[k].SessionID,
			Ordinal:    matches[k].Ordinal,
			Role:       meta.role.String,
			Sidechain:  meta.sidechain.Valid && meta.sidechain.Bool,
			Embeddable: meta.embeddable.Valid && meta.embeddable.Bool,
			Missing:    meta.missing,
		}
	}
	ranges, err := db.DeriveUnitRanges(ctx, s, anchors)
	if err != nil {
		return fmt.Errorf("deriving hybrid unit-less ranges: %w", err)
	}
	for k, i := range idxs {
		hits[i].ordinalStart, hits[i].ordinalEnd = ranges[k][0], ranges[k][1]
		hits[i].subordinate = anchors[k].Sidechain ||
			db.SubordinateSession(metas[k].relationship, metas[k].parentSessionID)
	}
	return nil
}

// enrichHybridMatchesPG looks up session/message metadata for the fused units
// (reusing enrichSemanticHitsPG) and assembles the final page in fused-score
// order. When the keyword leg contributed a unit, its display wins: the match
// anchors on the keyword-matched message's ordinal and centers on the keyword
// snippet (the vector leg's chunk anchor may be a different run member). Either
// way the returned snippet is built (and redacted) from the anchor message's
// full content via SemanticSnippet, the guarantee mode "semantic" gives.
// Unit-less keyword rows arrive with their derived range and subordinate flag
// already assigned pre-merge, so no derivation runs here.
func (s *Store) enrichHybridMatchesPG(
	ctx context.Context, f db.ContentSearchFilter, merged []db.FusedUnit,
	vecDisplay, kwDisplay map[string]pgHybridDisplay,
) (db.ContentSearchPage, error) {
	displays := make([]pgHybridDisplay, len(merged))
	asHits := make([]db.VectorHit, len(merged))
	for i, m := range merged {
		d, ok := kwDisplay[m.Unit.Key]
		if !ok {
			d = vecDisplay[m.Unit.Key]
		}
		displays[i] = d
		asHits[i] = db.VectorHit{SessionID: d.sessionID, Ordinal: d.ordinal}
	}
	meta, err := s.enrichSemanticHitsPG(ctx, asHits)
	if err != nil {
		return db.ContentSearchPage{}, err
	}

	out := make([]db.ContentMatch, 0, len(merged))
	for i, m := range merged {
		d := displays[i]
		info, ok := meta[db.MessageRef{SessionID: d.sessionID, Ordinal: d.ordinal}]
		if !ok {
			continue
		}
		score := m.Score
		out = append(out, db.ContentMatch{
			SessionID:       d.sessionID,
			Project:         info.project,
			Agent:           info.agent,
			Location:        "message",
			Role:            info.role,
			Ordinal:         d.ordinal,
			OrdinalRange:    [2]int{d.ordinalStart, d.ordinalEnd},
			Subordinate:     d.subordinate,
			Relationship:    info.relationshipType,
			ParentSessionID: info.parentSessionID,
			Sidechain:       info.isSidechain,
			Timestamp:       info.timestamp,
			Snippet:         f.SemanticSnippet(info.content, d.snippet),
			Score:           &score,
		})
	}
	return db.ContentSearchPage{Matches: out}, nil
}