wehub-resource-sync
f99010fae1
CI / lint (push) Failing after 1s
CI / frontend (push) Failing after 1s
CI / scripts (push) Failing after 1s
CI / Go Test (ubuntu-latest) (push) Failing after 0s
CI / frontend-node-25 (push) Failing after 1s
CI / docs (push) Failing after 0s
CI / coverage (push) Failing after 0s
CI / e2e (push) Failing after 0s
Docker / build-and-push (push) Failing after 1s
CI / integration (push) Failing after 4m43s
CI / Go Test (windows-latest) (push) Has been cancelled
CI / Desktop Unit Tests (Windows) (push) Has been cancelled
Desktop Artifacts / Desktop Build (Linux (arm64)) (push) Has been cancelled
Desktop Artifacts / Desktop Build (Linux) (push) Has been cancelled
Desktop Artifacts / Desktop Build (Windows) (push) Has been cancelled
Desktop Artifacts (macOS) / Desktop Build (macOS (aarch64)) (push) Has been cancelled
Desktop Artifacts (macOS) / Desktop Build (macOS (x86_64)) (push) Has been cancelled
1443 lines
53 KiB
Go
1443 lines
53 KiB
Go
//go:build pgtest
|
|
|
|
package postgres
|
|
|
|
import (
|
|
"context"
|
|
"database/sql"
|
|
"fmt"
|
|
"net/url"
|
|
"path/filepath"
|
|
"testing"
|
|
|
|
"github.com/stretchr/testify/assert"
|
|
"github.com/stretchr/testify/require"
|
|
|
|
"go.kenn.io/agentsview/internal/db"
|
|
)
|
|
|
|
// fakeVectorSource is an in-memory VectorPushSource whose generation, aggregate
|
|
// hashes, and per-session docs are mutated between pushes to drive the delta,
|
|
// eviction, and cross-generation cases. genOverride, when set, replaces the
|
|
// nth Generation answer (1-based call count) so a test can change the source's
|
|
// state between the push's initial Generation call and its mid-push
|
|
// re-checks. exportHashes, when set, overrides the per-session hash SessionDocs
|
|
// returns (simulating a local index that changed between the delta scan and
|
|
// the export); by default SessionDocs echoes the delta-scan hash so pushes
|
|
// proceed.
|
|
type fakeVectorSource struct {
|
|
gen VectorGenerationInfo
|
|
hasGen bool
|
|
hashes map[string]string
|
|
docs map[string][]VectorPushDoc
|
|
exportHashes map[string]string
|
|
docsCalls map[string]int // per-session SessionDocs invocation count
|
|
genCalls int
|
|
genOverride func(call int) (VectorGenerationInfo, bool, error)
|
|
}
|
|
|
|
func (f *fakeVectorSource) Generation(
|
|
ctx context.Context,
|
|
) (VectorGenerationInfo, bool, error) {
|
|
f.genCalls++
|
|
if f.genOverride != nil {
|
|
return f.genOverride(f.genCalls)
|
|
}
|
|
return f.gen, f.hasGen, nil
|
|
}
|
|
|
|
func (f *fakeVectorSource) SessionDocHashes(
|
|
ctx context.Context,
|
|
) (map[string]string, error) {
|
|
return f.hashes, nil
|
|
}
|
|
|
|
func (f *fakeVectorSource) SessionDocs(
|
|
ctx context.Context, id string,
|
|
) ([]VectorPushDoc, string, error) {
|
|
if f.docsCalls == nil {
|
|
f.docsCalls = make(map[string]int)
|
|
}
|
|
f.docsCalls[id]++
|
|
if hash, ok := f.exportHashes[id]; ok {
|
|
return f.docs[id], hash, nil
|
|
}
|
|
return f.docs[id], f.hashes[id], nil
|
|
}
|
|
|
|
// newVectorPushTestSync creates a fresh schema with the base + vector tables
|
|
// and returns a Sync wired to a local SQLite DB. It skips the test when the
|
|
// pgvector extension is unavailable, since the whole phase no-ops there.
|
|
func newVectorPushTestSync(
|
|
t *testing.T, pgURL, schema string,
|
|
) (*Sync, *db.DB, *sql.DB) {
|
|
t.Helper()
|
|
pg, err := Open(pgURL, schema, true)
|
|
require.NoError(t, err, "Open")
|
|
t.Cleanup(func() { _ = pg.Close() })
|
|
|
|
ctx := context.Background()
|
|
_, err = pg.Exec(`DROP SCHEMA IF EXISTS ` + schema + ` CASCADE`)
|
|
require.NoError(t, err, "drop schema")
|
|
require.NoError(t, EnsureSchema(ctx, pg, schema), "EnsureSchema")
|
|
|
|
available, err := VectorExtensionAvailable(ctx, pg)
|
|
require.NoError(t, err, "VectorExtensionAvailable")
|
|
if !available {
|
|
t.Skip("pgvector extension unavailable")
|
|
}
|
|
|
|
localDB, err := db.Open(filepath.Join(t.TempDir(), "local.db"))
|
|
require.NoError(t, err, "db.Open")
|
|
t.Cleanup(func() { _ = localDB.Close() })
|
|
|
|
sync := &Sync{
|
|
pg: pg,
|
|
local: localDB,
|
|
machine: "test-machine",
|
|
schema: schema,
|
|
schemaDone: true,
|
|
}
|
|
return sync, localDB, pg
|
|
}
|
|
|
|
// seedVectorSession inserts a minimal session with one message so that a Push
|
|
// creates a PG sessions row carrying this pusher's owner marker.
|
|
func seedVectorSession(t *testing.T, localDB *db.DB, id string) {
|
|
t.Helper()
|
|
require.NoError(t, localDB.UpsertSession(db.Session{
|
|
ID: id,
|
|
Project: "proj",
|
|
Machine: "test-machine",
|
|
Agent: "claude",
|
|
MessageCount: 1,
|
|
UserMessageCount: 1,
|
|
CreatedAt: "2026-01-01T00:00:00Z",
|
|
}), "UpsertSession "+id)
|
|
require.NoError(t, localDB.InsertMessages([]db.Message{{
|
|
SessionID: id,
|
|
Ordinal: 0,
|
|
Role: "user",
|
|
Content: id,
|
|
ContentLength: len(id),
|
|
}}), "InsertMessages "+id)
|
|
}
|
|
|
|
// seedVectorSessionProject inserts a minimal session under the given project so
|
|
// the PG sessions row carries a project the vector push filter can scope on.
|
|
func seedVectorSessionProject(t *testing.T, localDB *db.DB, id, project string) {
|
|
t.Helper()
|
|
require.NoError(t, localDB.UpsertSession(db.Session{
|
|
ID: id,
|
|
Project: project,
|
|
Machine: "test-machine",
|
|
Agent: "claude",
|
|
MessageCount: 1,
|
|
UserMessageCount: 1,
|
|
CreatedAt: "2026-01-01T00:00:00Z",
|
|
}), "UpsertSession "+id)
|
|
require.NoError(t, localDB.InsertMessages([]db.Message{{
|
|
SessionID: id,
|
|
Ordinal: 0,
|
|
Role: "user",
|
|
Content: id,
|
|
ContentLength: len(id),
|
|
}}), "InsertMessages "+id)
|
|
}
|
|
|
|
// vdoc builds a VectorPushDoc with the supplied embeddings as consecutive
|
|
// chunks (chunk_index 0..n-1).
|
|
func vdoc(
|
|
sessionID, docKey string, ordinal int,
|
|
content, hash string, chunks ...[]float32,
|
|
) VectorPushDoc {
|
|
d := VectorPushDoc{
|
|
DocKey: docKey,
|
|
SessionID: sessionID,
|
|
Ordinal: ordinal,
|
|
OrdinalEnd: ordinal,
|
|
OffsetsJSON: "[]",
|
|
Content: content,
|
|
ContentHash: hash,
|
|
}
|
|
for i, emb := range chunks {
|
|
d.Chunks = append(d.Chunks, VectorPushChunk{ChunkIndex: i, Embedding: emb})
|
|
}
|
|
return d
|
|
}
|
|
|
|
func countRows(t *testing.T, pg *sql.DB, query string, args ...any) int {
|
|
t.Helper()
|
|
var n int
|
|
require.NoError(t, pg.QueryRow(query, args...).Scan(&n), "count: "+query)
|
|
return n
|
|
}
|
|
|
|
// docOrdinal returns the current ordinal of the doc_key's vector_documents row.
|
|
func docOrdinal(t *testing.T, pg *sql.DB, docKey string) int {
|
|
t.Helper()
|
|
var ord int
|
|
require.NoError(t, pg.QueryRow(
|
|
`SELECT ordinal FROM vector_documents WHERE doc_key = $1`, docKey,
|
|
).Scan(&ord), "ordinal of "+docKey)
|
|
return ord
|
|
}
|
|
|
|
func TestVectorPushRoundTrip(t *testing.T) {
|
|
pgURL := testPGURL(t)
|
|
sync, localDB, pg := newVectorPushTestSync(
|
|
t, pgURL, "agentsview_vector_push_roundtrip_test")
|
|
ctx := context.Background()
|
|
|
|
seedVectorSession(t, localDB, "A")
|
|
seedVectorSession(t, localDB, "B")
|
|
|
|
sync.vectorSource = &fakeVectorSource{
|
|
gen: VectorGenerationInfo{Fingerprint: "fp-rt", Model: "m", Dimension: 4},
|
|
hasGen: true,
|
|
hashes: map[string]string{"A": "ha", "B": "hb"},
|
|
docs: map[string][]VectorPushDoc{
|
|
"A": {
|
|
vdoc("A", "A#0", 0, "c1", "h1",
|
|
[]float32{1, 0, 0, 0}, []float32{0, 1, 0, 0}),
|
|
vdoc("A", "A#1", 1, "c2", "h2", []float32{0, 0, 1, 0}),
|
|
},
|
|
"B": {vdoc("B", "B#0", 0, "c3", "h3", []float32{0, 0, 0, 1})},
|
|
},
|
|
}
|
|
|
|
var vectorReports, prepareReports []PushProgress
|
|
res, err := sync.Push(ctx, false, func(p PushProgress) {
|
|
switch p.Phase {
|
|
case "vectors":
|
|
vectorReports = append(vectorReports, p)
|
|
case "preparing":
|
|
prepareReports = append(prepareReports, p)
|
|
}
|
|
})
|
|
require.NoError(t, err, "Push")
|
|
assert.False(t, res.Vectors.Skipped)
|
|
assert.Equal(t, 2, res.Vectors.SessionsPushed)
|
|
assert.Equal(t, 3, res.Vectors.DocsPushed)
|
|
assert.Equal(t, 4, res.Vectors.ChunksPushed)
|
|
|
|
require.NotEmpty(t, prepareReports, "fingerprint phase must report progress")
|
|
prepared := prepareReports[len(prepareReports)-1]
|
|
assert.Equal(t, 2, prepared.SessionsDone)
|
|
assert.Equal(t, 2, prepared.SessionsTotal)
|
|
|
|
require.NotEmpty(t, vectorReports, "vector phase must report progress")
|
|
final := vectorReports[len(vectorReports)-1]
|
|
assert.Equal(t, 2, final.VectorSessionsDone)
|
|
assert.Equal(t, 2, final.VectorSessionsTotal)
|
|
assert.Equal(t, 4, final.VectorChunksPushed)
|
|
|
|
genID, dim, ok, err := LookupVectorGeneration(ctx, pg, "fp-rt")
|
|
require.NoError(t, err, "LookupVectorGeneration")
|
|
require.True(t, ok, "generation registered")
|
|
assert.Equal(t, 4, dim)
|
|
|
|
assert.Equal(t, 3,
|
|
countRows(t, pg, `SELECT COUNT(*) FROM vector_documents`))
|
|
assert.Equal(t, 4,
|
|
countRows(t, pg, `SELECT COUNT(*) FROM `+vectorChunkTable(genID)))
|
|
assert.Equal(t, 2, countRows(t, pg,
|
|
`SELECT COUNT(*) FROM vector_push_state WHERE generation_id = $1`, genID))
|
|
assert.Equal(t, 1, countRows(t, pg,
|
|
`SELECT COUNT(*) FROM vector_generation_machines
|
|
WHERE generation_id = $1 AND machine = $2`, genID, "test-machine"))
|
|
}
|
|
|
|
func TestVectorPushDeltaNoop(t *testing.T) {
|
|
pgURL := testPGURL(t)
|
|
sync, localDB, _ := newVectorPushTestSync(
|
|
t, pgURL, "agentsview_vector_push_noop_test")
|
|
ctx := context.Background()
|
|
|
|
seedVectorSession(t, localDB, "A")
|
|
seedVectorSession(t, localDB, "B")
|
|
|
|
sync.vectorSource = &fakeVectorSource{
|
|
gen: VectorGenerationInfo{Fingerprint: "fp-noop", Model: "m", Dimension: 4},
|
|
hasGen: true,
|
|
hashes: map[string]string{"A": "ha", "B": "hb"},
|
|
docs: map[string][]VectorPushDoc{
|
|
"A": {vdoc("A", "A#0", 0, "c1", "h1", []float32{1, 0, 0, 0})},
|
|
"B": {vdoc("B", "B#0", 0, "c3", "h3", []float32{0, 0, 0, 1})},
|
|
},
|
|
}
|
|
|
|
_, err := sync.Push(ctx, false, nil)
|
|
require.NoError(t, err, "first Push")
|
|
|
|
res, err := sync.Push(ctx, false, nil)
|
|
require.NoError(t, err, "second Push")
|
|
assert.Equal(t, 2, res.Vectors.SessionsUnchanged)
|
|
assert.Equal(t, 0, res.Vectors.SessionsPushed)
|
|
assert.Equal(t, 0, res.Vectors.DocsPushed)
|
|
}
|
|
|
|
// TestVectorPushFullRepairsSilentCorruption pins --full's repair contract for
|
|
// the vector phase: a chunk row missing from PG while vector_push_state still
|
|
// reports the session current is invisible to a normal delta push, but a full
|
|
// push bypasses the unchanged-hash skip and restores it.
|
|
func TestVectorPushFullRepairsSilentCorruption(t *testing.T) {
|
|
pgURL := testPGURL(t)
|
|
sync, localDB, pg := newVectorPushTestSync(
|
|
t, pgURL, "agentsview_vector_push_full_test")
|
|
ctx := context.Background()
|
|
|
|
seedVectorSession(t, localDB, "A")
|
|
|
|
sync.vectorSource = &fakeVectorSource{
|
|
gen: VectorGenerationInfo{Fingerprint: "fp-full", Model: "m", Dimension: 4},
|
|
hasGen: true,
|
|
hashes: map[string]string{"A": "ha"},
|
|
docs: map[string][]VectorPushDoc{
|
|
"A": {vdoc("A", "A#0", 0, "c1", "h1", []float32{1, 0, 0, 0})},
|
|
},
|
|
}
|
|
|
|
_, err := sync.Push(ctx, false, nil)
|
|
require.NoError(t, err, "seed Push")
|
|
genID, _, ok, err := LookupVectorGeneration(ctx, pg, "fp-full")
|
|
require.NoError(t, err)
|
|
require.True(t, ok)
|
|
|
|
// Silently lose the chunk while push state still says "current".
|
|
_, err = pg.Exec(`DELETE FROM ` + vectorChunkTable(genID))
|
|
require.NoError(t, err, "corrupting chunk table")
|
|
|
|
res, err := sync.pushVectors(ctx, false, nil, nil)
|
|
require.NoError(t, err, "delta pushVectors")
|
|
assert.Equal(t, 1, res.SessionsUnchanged, "delta push cannot see the loss")
|
|
assert.Equal(t, 0,
|
|
countRows(t, pg, `SELECT COUNT(*) FROM `+vectorChunkTable(genID)))
|
|
|
|
res, err = sync.pushVectors(ctx, true, nil, nil)
|
|
require.NoError(t, err, "full pushVectors")
|
|
assert.Equal(t, 1, res.SessionsPushed)
|
|
assert.Equal(t, 0, res.SessionsUnchanged)
|
|
assert.Equal(t, 1,
|
|
countRows(t, pg, `SELECT COUNT(*) FROM `+vectorChunkTable(genID)),
|
|
"full push restores the missing chunk")
|
|
}
|
|
|
|
// TestVectorPushDeferredOnSessionError pins that a changed session named in
|
|
// failedSessions is deferred — its newer local vectors must not land ahead of
|
|
// sessions/messages rows its session-phase push failed to write — and that
|
|
// the untouched delta state sends it on the next successful push.
|
|
func TestVectorPushDeferredOnSessionError(t *testing.T) {
|
|
pgURL := testPGURL(t)
|
|
sync, localDB, pg := newVectorPushTestSync(
|
|
t, pgURL, "agentsview_vector_push_deferred_test")
|
|
ctx := context.Background()
|
|
|
|
seedVectorSession(t, localDB, "A")
|
|
seedVectorSession(t, localDB, "B")
|
|
|
|
fake := &fakeVectorSource{
|
|
gen: VectorGenerationInfo{Fingerprint: "fp-def", Model: "m", Dimension: 4},
|
|
hasGen: true,
|
|
hashes: map[string]string{"A": "ha1", "B": "hb1"},
|
|
docs: map[string][]VectorPushDoc{
|
|
"A": {vdoc("A", "A#0", 0, "old", "h-old", []float32{1, 0, 0, 0})},
|
|
"B": {vdoc("B", "B#0", 0, "cb", "hb", []float32{0, 0, 0, 1})},
|
|
},
|
|
}
|
|
sync.vectorSource = fake
|
|
|
|
_, err := sync.Push(ctx, false, nil)
|
|
require.NoError(t, err, "seed Push")
|
|
|
|
// A's docs change locally, but its session push failed this run.
|
|
fake.hashes["A"] = "ha2"
|
|
fake.docs["A"] = []VectorPushDoc{
|
|
vdoc("A", "A#0", 0, "new", "h-new", []float32{0, 1, 0, 0}),
|
|
}
|
|
res, err := sync.pushVectors(ctx, false, map[string]struct{}{"A": {}}, nil)
|
|
require.NoError(t, err, "deferred pushVectors")
|
|
assert.Equal(t, 1, res.SessionsDeferred)
|
|
assert.Equal(t, 0, res.SessionsPushed)
|
|
assert.Equal(t, 1, res.SessionsUnchanged, "B is unchanged")
|
|
|
|
var content string
|
|
require.NoError(t, pg.QueryRow(
|
|
`SELECT content FROM vector_documents WHERE doc_key = $1`, "A#0",
|
|
).Scan(&content))
|
|
assert.Equal(t, "old", content, "deferred session keeps its prior vectors")
|
|
|
|
// Next push with no failures sends the deferred session.
|
|
res, err = sync.pushVectors(ctx, false, nil, nil)
|
|
require.NoError(t, err, "healing pushVectors")
|
|
assert.Equal(t, 0, res.SessionsDeferred)
|
|
assert.Equal(t, 1, res.SessionsPushed)
|
|
require.NoError(t, pg.QueryRow(
|
|
`SELECT content FROM vector_documents WHERE doc_key = $1`, "A#0",
|
|
).Scan(&content))
|
|
assert.Equal(t, "new", content)
|
|
}
|
|
|
|
func TestVectorPushContentChange(t *testing.T) {
|
|
pgURL := testPGURL(t)
|
|
sync, localDB, pg := newVectorPushTestSync(
|
|
t, pgURL, "agentsview_vector_push_content_test")
|
|
ctx := context.Background()
|
|
|
|
seedVectorSession(t, localDB, "A")
|
|
seedVectorSession(t, localDB, "B")
|
|
|
|
fake := &fakeVectorSource{
|
|
gen: VectorGenerationInfo{Fingerprint: "fp-cc", Model: "m", Dimension: 4},
|
|
hasGen: true,
|
|
hashes: map[string]string{"A": "ha", "B": "hb"},
|
|
docs: map[string][]VectorPushDoc{
|
|
"A": {vdoc("A", "A#0", 0, "c1", "h1", []float32{1, 0, 0, 0})},
|
|
"B": {vdoc("B", "B#0", 0, "c3", "h3", []float32{0, 0, 0, 1})},
|
|
},
|
|
}
|
|
sync.vectorSource = fake
|
|
|
|
_, err := sync.Push(ctx, false, nil)
|
|
require.NoError(t, err, "first Push")
|
|
|
|
fake.hashes["A"] = "ha2"
|
|
fake.docs["A"] = []VectorPushDoc{
|
|
vdoc("A", "A#0", 0, "c1-updated", "h1b", []float32{0, 1, 0, 0}),
|
|
}
|
|
|
|
res, err := sync.Push(ctx, false, nil)
|
|
require.NoError(t, err, "second Push")
|
|
assert.Equal(t, 1, res.Vectors.SessionsPushed)
|
|
assert.Equal(t, 1, res.Vectors.SessionsUnchanged)
|
|
|
|
var content, hash string
|
|
require.NoError(t, pg.QueryRow(
|
|
`SELECT content, content_hash FROM vector_documents WHERE doc_key = $1`,
|
|
"A#0").Scan(&content, &hash), "read updated doc")
|
|
assert.Equal(t, "c1-updated", content)
|
|
assert.Equal(t, "h1b", hash)
|
|
}
|
|
|
|
func TestVectorPushEviction(t *testing.T) {
|
|
pgURL := testPGURL(t)
|
|
sync, localDB, pg := newVectorPushTestSync(
|
|
t, pgURL, "agentsview_vector_push_evict_test")
|
|
ctx := context.Background()
|
|
|
|
seedVectorSession(t, localDB, "A")
|
|
seedVectorSession(t, localDB, "B")
|
|
|
|
fake := &fakeVectorSource{
|
|
gen: VectorGenerationInfo{Fingerprint: "fp-ev", Model: "m", Dimension: 4},
|
|
hasGen: true,
|
|
hashes: map[string]string{"A": "ha", "B": "hb"},
|
|
docs: map[string][]VectorPushDoc{
|
|
"A": {vdoc("A", "A#0", 0, "c1", "h1", []float32{1, 0, 0, 0})},
|
|
"B": {vdoc("B", "B#0", 0, "c3", "h3", []float32{0, 0, 0, 1})},
|
|
},
|
|
}
|
|
sync.vectorSource = fake
|
|
|
|
_, err := sync.Push(ctx, false, nil)
|
|
require.NoError(t, err, "first Push")
|
|
genID, _, ok, err := LookupVectorGeneration(ctx, pg, "fp-ev")
|
|
require.NoError(t, err)
|
|
require.True(t, ok)
|
|
|
|
delete(fake.hashes, "B")
|
|
delete(fake.docs, "B")
|
|
|
|
res, err := sync.Push(ctx, false, nil)
|
|
require.NoError(t, err, "second Push")
|
|
assert.Equal(t, 1, res.Vectors.SessionsEvicted)
|
|
assert.Equal(t, 1, res.Vectors.DocsDeleted)
|
|
|
|
assert.Equal(t, 0, countRows(t, pg,
|
|
`SELECT COUNT(*) FROM `+vectorChunkTable(genID)+` WHERE doc_key = $1`,
|
|
"B#0"))
|
|
assert.Equal(t, 0, countRows(t, pg,
|
|
`SELECT COUNT(*) FROM vector_push_state
|
|
WHERE generation_id = $1 AND session_id = $2`, genID, "B"))
|
|
assert.Equal(t, 0, countRows(t, pg,
|
|
`SELECT COUNT(*) FROM vector_documents WHERE session_id = $1`, "B"))
|
|
assert.Equal(t, 1, countRows(t, pg,
|
|
`SELECT COUNT(*) FROM vector_documents WHERE session_id = $1`, "A"))
|
|
}
|
|
|
|
func TestVectorPushNoActiveGeneration(t *testing.T) {
|
|
pgURL := testPGURL(t)
|
|
sync, localDB, pg := newVectorPushTestSync(
|
|
t, pgURL, "agentsview_vector_push_nogen_test")
|
|
ctx := context.Background()
|
|
|
|
seedVectorSession(t, localDB, "A")
|
|
sync.vectorSource = &fakeVectorSource{hasGen: false}
|
|
|
|
res, err := sync.Push(ctx, false, nil)
|
|
require.NoError(t, err, "Push")
|
|
assert.True(t, res.Vectors.Skipped)
|
|
assert.NotEmpty(t, res.Vectors.SkippedReason)
|
|
assert.Equal(t, 0,
|
|
countRows(t, pg, `SELECT COUNT(*) FROM vector_generations`))
|
|
}
|
|
|
|
func TestVectorPushSharedDocAcrossGenerations(t *testing.T) {
|
|
pgURL := testPGURL(t)
|
|
sync, localDB, pg := newVectorPushTestSync(
|
|
t, pgURL, "agentsview_vector_push_sharedgen_test")
|
|
ctx := context.Background()
|
|
|
|
seedVectorSession(t, localDB, "X")
|
|
seedVectorSession(t, localDB, "Y")
|
|
|
|
fake := &fakeVectorSource{
|
|
gen: VectorGenerationInfo{Fingerprint: "fp-a", Model: "m", Dimension: 4},
|
|
hasGen: true,
|
|
hashes: map[string]string{"X": "hx", "Y": "hy"},
|
|
docs: map[string][]VectorPushDoc{
|
|
"X": {vdoc("X", "X#0", 0, "cx", "hx1", []float32{1, 0, 0, 0})},
|
|
"Y": {vdoc("Y", "Y#0", 0, "cy", "hy1", []float32{0, 1, 0, 0})},
|
|
},
|
|
}
|
|
sync.vectorSource = fake
|
|
|
|
_, err := sync.Push(ctx, false, nil)
|
|
require.NoError(t, err, "push gen A")
|
|
genA, _, ok, err := LookupVectorGeneration(ctx, pg, "fp-a")
|
|
require.NoError(t, err)
|
|
require.True(t, ok)
|
|
|
|
// Switch to a second generation (new fingerprint) sharing the same docs.
|
|
fake.gen.Fingerprint = "fp-b"
|
|
_, err = sync.Push(ctx, false, nil)
|
|
require.NoError(t, err, "push gen B")
|
|
genB, _, ok, err := LookupVectorGeneration(ctx, pg, "fp-b")
|
|
require.NoError(t, err)
|
|
require.True(t, ok)
|
|
|
|
// Drop Y from gen B's source, triggering a gen-B eviction of Y.
|
|
delete(fake.hashes, "Y")
|
|
delete(fake.docs, "Y")
|
|
|
|
res, err := sync.Push(ctx, false, nil)
|
|
require.NoError(t, err, "push gen B without Y")
|
|
assert.Equal(t, 1, res.Vectors.SessionsEvicted)
|
|
assert.Equal(t, 0, res.Vectors.DocsDeleted,
|
|
"Y's doc is still referenced by gen A, so it must survive")
|
|
|
|
assert.Equal(t, 1, countRows(t, pg,
|
|
`SELECT COUNT(*) FROM `+vectorChunkTable(genA)+` WHERE doc_key = $1`,
|
|
"Y#0"), "gen A chunks for Y survive")
|
|
assert.Equal(t, 0, countRows(t, pg,
|
|
`SELECT COUNT(*) FROM `+vectorChunkTable(genB)+` WHERE doc_key = $1`,
|
|
"Y#0"), "gen B chunks for Y evicted")
|
|
assert.Equal(t, 1, countRows(t, pg,
|
|
`SELECT COUNT(*) FROM vector_documents WHERE doc_key = $1`, "Y#0"),
|
|
"shared doc row survives")
|
|
assert.Equal(t, 0, countRows(t, pg,
|
|
`SELECT COUNT(*) FROM vector_push_state
|
|
WHERE generation_id = $1 AND session_id = $2`, genB, "Y"))
|
|
}
|
|
|
|
// TestVectorPushOrdinalShift pins the slot-safe replacement: stable doc_keys
|
|
// whose ordinals shift up (0/1 -> 1/2 with a new doc taking slot 0) re-push
|
|
// without a UNIQUE (session_id, ordinal) violation and land on their final
|
|
// ordinals.
|
|
func TestVectorPushOrdinalShift(t *testing.T) {
|
|
pgURL := testPGURL(t)
|
|
sync, localDB, pg := newVectorPushTestSync(
|
|
t, pgURL, "agentsview_vector_push_shift_test")
|
|
ctx := context.Background()
|
|
|
|
seedVectorSession(t, localDB, "A")
|
|
|
|
fake := &fakeVectorSource{
|
|
gen: VectorGenerationInfo{Fingerprint: "fp-sh", Model: "m", Dimension: 4},
|
|
hasGen: true,
|
|
hashes: map[string]string{"A": "h1"},
|
|
docs: map[string][]VectorPushDoc{
|
|
"A": {
|
|
vdoc("A", "k0", 0, "c0", "hc0", []float32{1, 0, 0, 0}),
|
|
vdoc("A", "k1", 1, "c1", "hc1", []float32{0, 1, 0, 0}),
|
|
},
|
|
},
|
|
}
|
|
sync.vectorSource = fake
|
|
|
|
_, err := sync.Push(ctx, false, nil)
|
|
require.NoError(t, err, "first Push")
|
|
|
|
// Shift k0/k1 up to ordinals 1/2 and insert kNew at ordinal 0.
|
|
fake.hashes["A"] = "h2"
|
|
fake.docs["A"] = []VectorPushDoc{
|
|
vdoc("A", "kNew", 0, "cn", "hcn", []float32{0, 0, 1, 0}),
|
|
vdoc("A", "k0", 1, "c0", "hc0", []float32{1, 0, 0, 0}),
|
|
vdoc("A", "k1", 2, "c1", "hc1", []float32{0, 1, 0, 0}),
|
|
}
|
|
|
|
res, err := sync.Push(ctx, false, nil)
|
|
require.NoError(t, err, "shift Push")
|
|
assert.Equal(t, 1, res.Vectors.SessionsPushed)
|
|
assert.Equal(t, 0, res.Vectors.DocsDeleted)
|
|
|
|
assert.Equal(t, 0, docOrdinal(t, pg, "kNew"))
|
|
assert.Equal(t, 1, docOrdinal(t, pg, "k0"))
|
|
assert.Equal(t, 2, docOrdinal(t, pg, "k1"))
|
|
assert.Equal(t, 3,
|
|
countRows(t, pg, `SELECT COUNT(*) FROM vector_documents WHERE session_id = $1`, "A"))
|
|
}
|
|
|
|
// TestVectorPushDocRemovedAndShared pins that a doc removed from a re-pushed
|
|
// session disappears from vector_documents in the same push, cascading its
|
|
// chunks out of every generation — including one that still embedded it.
|
|
// Regression: preserving the shared row parked at a negative ordinal left the
|
|
// older generation with chunks the read path could never hydrate (the
|
|
// ordinal >= 0 tombstone guard), silently shrinking its results. Local kit
|
|
// removal deletes a vanished doc's vectors from all generations, and PG must
|
|
// match.
|
|
func TestVectorPushDocRemovedAndShared(t *testing.T) {
|
|
pgURL := testPGURL(t)
|
|
sync, localDB, pg := newVectorPushTestSync(
|
|
t, pgURL, "agentsview_vector_push_shrink_test")
|
|
ctx := context.Background()
|
|
|
|
seedVectorSession(t, localDB, "A")
|
|
|
|
// Gen A embeds kShared + kDropShared (not kDropSolo).
|
|
fake := &fakeVectorSource{
|
|
gen: VectorGenerationInfo{Fingerprint: "fp-ga", Model: "m", Dimension: 4},
|
|
hasGen: true,
|
|
hashes: map[string]string{"A": "ha"},
|
|
docs: map[string][]VectorPushDoc{
|
|
"A": {
|
|
vdoc("A", "kShared", 0, "cs", "hcs", []float32{1, 0, 0, 0}),
|
|
vdoc("A", "kDropShared", 1, "cds", "hcds", []float32{0, 1, 0, 0}),
|
|
},
|
|
},
|
|
}
|
|
sync.vectorSource = fake
|
|
_, err := sync.Push(ctx, false, nil)
|
|
require.NoError(t, err, "push gen A")
|
|
genA, _, ok, err := LookupVectorGeneration(ctx, pg, "fp-ga")
|
|
require.NoError(t, err)
|
|
require.True(t, ok)
|
|
|
|
// Gen B embeds all three docs for A.
|
|
fake.gen.Fingerprint = "fp-gb"
|
|
fake.hashes["A"] = "hb1"
|
|
fake.docs["A"] = []VectorPushDoc{
|
|
vdoc("A", "kShared", 0, "cs", "hcs", []float32{1, 0, 0, 0}),
|
|
vdoc("A", "kDropShared", 1, "cds", "hcds", []float32{0, 1, 0, 0}),
|
|
vdoc("A", "kDropSolo", 2, "cdo", "hcdo", []float32{0, 0, 1, 0}),
|
|
}
|
|
_, err = sync.Push(ctx, false, nil)
|
|
require.NoError(t, err, "push gen B")
|
|
genB, _, ok, err := LookupVectorGeneration(ctx, pg, "fp-gb")
|
|
require.NoError(t, err)
|
|
require.True(t, ok)
|
|
|
|
// Shrink gen B's A to only kShared: kDropShared + kDropSolo vanish locally.
|
|
fake.hashes["A"] = "hb2"
|
|
fake.docs["A"] = []VectorPushDoc{
|
|
vdoc("A", "kShared", 0, "cs", "hcs", []float32{1, 0, 0, 0}),
|
|
}
|
|
res, err := sync.Push(ctx, false, nil)
|
|
require.NoError(t, err, "shrink push gen B")
|
|
assert.Equal(t, 1, res.Vectors.SessionsPushed)
|
|
assert.Equal(t, 2, res.Vectors.DocsDeleted,
|
|
"kDropSolo and kDropShared both vanished locally")
|
|
|
|
// Both vanished docs are gone, whether or not another generation embedded
|
|
// them.
|
|
assert.Equal(t, 0, countRows(t, pg,
|
|
`SELECT COUNT(*) FROM vector_documents WHERE doc_key = $1`, "kDropSolo"))
|
|
assert.Equal(t, 0, countRows(t, pg,
|
|
`SELECT COUNT(*) FROM vector_documents WHERE doc_key = $1`, "kDropShared"))
|
|
assert.Equal(t, 0, countRows(t, pg,
|
|
`SELECT COUNT(*) FROM `+vectorChunkTable(genA)+` WHERE doc_key = $1`,
|
|
"kDropShared"), "gen A chunks for kDropShared cascade-deleted")
|
|
assert.Equal(t, 0, countRows(t, pg,
|
|
`SELECT COUNT(*) FROM `+vectorChunkTable(genB)+` WHERE doc_key = $1`,
|
|
"kDropShared"), "gen B chunks for kDropShared cleared")
|
|
// No parked tombstones survive the push, and every remaining gen A chunk
|
|
// hydrates: the older generation must not retain chunks pointing at rows
|
|
// the read path's ordinal >= 0 guard hides.
|
|
assert.Equal(t, 0, countRows(t, pg,
|
|
`SELECT COUNT(*) FROM vector_documents WHERE ordinal < 0`))
|
|
assert.Equal(t, 0, countRows(t, pg, `
|
|
SELECT COUNT(*) FROM `+vectorChunkTable(genA)+` c
|
|
LEFT JOIN vector_documents d
|
|
ON d.doc_key = c.doc_key AND d.ordinal >= 0
|
|
WHERE d.doc_key IS NULL`),
|
|
"gen A has no chunks that cannot hydrate")
|
|
// kShared stays live at its positive ordinal in both generations.
|
|
assert.Equal(t, 0, docOrdinal(t, pg, "kShared"))
|
|
assert.Equal(t, 1, countRows(t, pg,
|
|
`SELECT COUNT(*) FROM `+vectorChunkTable(genA)+` WHERE doc_key = $1`,
|
|
"kShared"), "gen A keeps its chunks for the still-present doc")
|
|
}
|
|
|
|
// TestVectorPushConflictSkipped pins finding 3: a session whose PG owner marker
|
|
// names another machine is skipped on push (counted in Conflicts) and left
|
|
// untouched on evict (also counted).
|
|
func TestVectorPushConflictSkipped(t *testing.T) {
|
|
pgURL := testPGURL(t)
|
|
sync, localDB, pg := newVectorPushTestSync(
|
|
t, pgURL, "agentsview_vector_push_conflict_test")
|
|
ctx := context.Background()
|
|
|
|
seedVectorSession(t, localDB, "A")
|
|
seedVectorSession(t, localDB, "B")
|
|
|
|
fake := &fakeVectorSource{
|
|
gen: VectorGenerationInfo{Fingerprint: "fp-cf", Model: "m", Dimension: 4},
|
|
hasGen: true,
|
|
hashes: map[string]string{"A": "ha", "B": "hb"},
|
|
docs: map[string][]VectorPushDoc{
|
|
"A": {vdoc("A", "cA0", 0, "ca", "hca", []float32{1, 0, 0, 0})},
|
|
"B": {vdoc("B", "cB0", 0, "cb", "hcb", []float32{0, 0, 0, 1})},
|
|
},
|
|
}
|
|
sync.vectorSource = fake
|
|
_, err := sync.Push(ctx, false, nil)
|
|
require.NoError(t, err, "first Push")
|
|
|
|
// Hand A to another machine on the PG side.
|
|
_, err = pg.Exec(
|
|
`UPDATE sessions SET owner_marker = 'other-machine' WHERE id = $1`, "A")
|
|
require.NoError(t, err, "reassign owner")
|
|
|
|
// A changed locally: the vector phase must skip it as a conflict, leaving
|
|
// its pushed content intact.
|
|
fake.hashes["A"] = "ha2"
|
|
fake.docs["A"] = []VectorPushDoc{
|
|
vdoc("A", "cA0", 0, "ca-new", "hca2", []float32{0, 1, 0, 0}),
|
|
}
|
|
res, err := sync.pushVectors(ctx, false, nil, nil)
|
|
require.NoError(t, err, "push conflict")
|
|
assert.Equal(t, 1, res.Conflicts)
|
|
assert.Equal(t, 0, res.SessionsPushed)
|
|
var content string
|
|
require.NoError(t, pg.QueryRow(
|
|
`SELECT content FROM vector_documents WHERE doc_key = $1`, "cA0",
|
|
).Scan(&content))
|
|
assert.Equal(t, "ca", content, "conflicted doc left untouched")
|
|
|
|
// A now vanishes from local: the evict path must leave the other machine's
|
|
// session in place and count the conflict.
|
|
delete(fake.hashes, "A")
|
|
delete(fake.docs, "A")
|
|
res, err = sync.pushVectors(ctx, false, nil, nil)
|
|
require.NoError(t, err, "evict conflict")
|
|
assert.Equal(t, 1, res.Conflicts)
|
|
assert.Equal(t, 0, res.SessionsEvicted)
|
|
genID, _, ok, err := LookupVectorGeneration(ctx, pg, "fp-cf")
|
|
require.NoError(t, err)
|
|
require.True(t, ok)
|
|
assert.Equal(t, 1, countRows(t, pg,
|
|
`SELECT COUNT(*) FROM vector_push_state
|
|
WHERE generation_id = $1 AND session_id = $2`, genID, "A"))
|
|
assert.Equal(t, 1, countRows(t, pg,
|
|
`SELECT COUNT(*) FROM vector_documents WHERE session_id = $1`, "A"))
|
|
}
|
|
|
|
// TestVectorPushLegacyMachineOwnership pins machine-aware ownership for
|
|
// legacy PG rows whose owner_marker is empty: a foreign machine name blocks
|
|
// both the push and evict paths (counted as conflicts), while this pusher's
|
|
// own machine name self-heals — its vectors push and evict normally.
|
|
func TestVectorPushLegacyMachineOwnership(t *testing.T) {
|
|
pgURL := testPGURL(t)
|
|
sync, localDB, pg := newVectorPushTestSync(
|
|
t, pgURL, "agentsview_vector_push_legacy_test")
|
|
ctx := context.Background()
|
|
|
|
seedVectorSession(t, localDB, "A")
|
|
seedVectorSession(t, localDB, "B")
|
|
|
|
fake := &fakeVectorSource{
|
|
gen: VectorGenerationInfo{Fingerprint: "fp-lg", Model: "m", Dimension: 4},
|
|
hasGen: true,
|
|
hashes: map[string]string{"A": "ha", "B": "hb"},
|
|
docs: map[string][]VectorPushDoc{
|
|
"A": {vdoc("A", "lA0", 0, "ca", "hca", []float32{1, 0, 0, 0})},
|
|
"B": {vdoc("B", "lB0", 0, "cb", "hcb", []float32{0, 0, 0, 1})},
|
|
},
|
|
}
|
|
sync.vectorSource = fake
|
|
_, err := sync.Push(ctx, false, nil)
|
|
require.NoError(t, err, "first Push")
|
|
|
|
// Rewrite both PG rows as legacy (empty marker): A owned by a foreign
|
|
// machine, B carrying this pusher's own machine name.
|
|
_, err = pg.Exec(`UPDATE sessions
|
|
SET owner_marker = '', machine = 'other-machine' WHERE id = $1`, "A")
|
|
require.NoError(t, err, "make A a foreign legacy row")
|
|
_, err = pg.Exec(`UPDATE sessions
|
|
SET owner_marker = '', machine = 'test-machine' WHERE id = $1`, "B")
|
|
require.NoError(t, err, "make B an owned legacy row")
|
|
|
|
// Both change locally: only B may push.
|
|
fake.hashes["A"], fake.hashes["B"] = "ha2", "hb2"
|
|
fake.docs["A"] = []VectorPushDoc{
|
|
vdoc("A", "lA0", 0, "ca-new", "hca2", []float32{0, 1, 0, 0}),
|
|
}
|
|
fake.docs["B"] = []VectorPushDoc{
|
|
vdoc("B", "lB0", 0, "cb-new", "hcb2", []float32{0, 0, 1, 0}),
|
|
}
|
|
res, err := sync.pushVectors(ctx, false, nil, nil)
|
|
require.NoError(t, err, "legacy push")
|
|
assert.Equal(t, 1, res.Conflicts, "foreign legacy row is a conflict")
|
|
assert.Equal(t, 1, res.SessionsPushed, "owned legacy row self-heals")
|
|
var content string
|
|
require.NoError(t, pg.QueryRow(
|
|
`SELECT content FROM vector_documents WHERE doc_key = $1`, "lA0",
|
|
).Scan(&content))
|
|
assert.Equal(t, "ca", content, "foreign legacy doc left untouched")
|
|
require.NoError(t, pg.QueryRow(
|
|
`SELECT content FROM vector_documents WHERE doc_key = $1`, "lB0",
|
|
).Scan(&content))
|
|
assert.Equal(t, "cb-new", content)
|
|
|
|
// Both vanish locally: only B may be evicted.
|
|
fake.hashes = map[string]string{}
|
|
fake.docs = map[string][]VectorPushDoc{}
|
|
res, err = sync.pushVectors(ctx, false, nil, nil)
|
|
require.NoError(t, err, "legacy evict")
|
|
assert.Equal(t, 1, res.Conflicts, "foreign legacy row is not evicted")
|
|
assert.Equal(t, 1, res.SessionsEvicted)
|
|
assert.Equal(t, 1, countRows(t, pg,
|
|
`SELECT COUNT(*) FROM vector_documents WHERE session_id = $1`, "A"))
|
|
assert.Equal(t, 0, countRows(t, pg,
|
|
`SELECT COUNT(*) FROM vector_documents WHERE session_id = $1`, "B"))
|
|
}
|
|
|
|
// TestVectorPushEvictionSkippedWhenSourceTurnsUnready pins the pre-eviction
|
|
// re-check: when the local source becomes unready between the delta scan and
|
|
// the eviction pass (an embeddings rebuild started mid-push), the eviction
|
|
// list — computed from a now-partial local view — must be dropped, and the
|
|
// next healthy push re-derives and applies it.
|
|
func TestVectorPushEvictionSkippedWhenSourceTurnsUnready(t *testing.T) {
|
|
pgURL := testPGURL(t)
|
|
sync, localDB, pg := newVectorPushTestSync(
|
|
t, pgURL, "agentsview_vector_push_unready_test")
|
|
ctx := context.Background()
|
|
|
|
seedVectorSession(t, localDB, "A")
|
|
seedVectorSession(t, localDB, "B")
|
|
|
|
fake := &fakeVectorSource{
|
|
gen: VectorGenerationInfo{Fingerprint: "fp-ur", Model: "m", Dimension: 4},
|
|
hasGen: true,
|
|
hashes: map[string]string{"A": "ha", "B": "hb"},
|
|
docs: map[string][]VectorPushDoc{
|
|
"A": {vdoc("A", "uA0", 0, "ca", "hca", []float32{1, 0, 0, 0})},
|
|
"B": {vdoc("B", "uB0", 0, "cb", "hcb", []float32{0, 0, 0, 1})},
|
|
},
|
|
}
|
|
sync.vectorSource = fake
|
|
_, err := sync.Push(ctx, false, nil)
|
|
require.NoError(t, err, "first Push")
|
|
|
|
// B vanishes locally, but the source turns unready before the evict pass.
|
|
delete(fake.hashes, "B")
|
|
delete(fake.docs, "B")
|
|
fake.genCalls = 0
|
|
fake.genOverride = func(call int) (VectorGenerationInfo, bool, error) {
|
|
if call == 1 {
|
|
return fake.gen, true, nil
|
|
}
|
|
return VectorGenerationInfo{}, false, fmt.Errorf(
|
|
"%w: rebuild started", ErrVectorSourceNotReady)
|
|
}
|
|
res, err := sync.pushVectors(ctx, false, nil, nil)
|
|
require.NoError(t, err, "push with mid-push rebuild")
|
|
assert.Equal(t, 0, res.SessionsEvicted, "eviction must be dropped")
|
|
assert.Equal(t, 1, countRows(t, pg,
|
|
`SELECT COUNT(*) FROM vector_documents WHERE session_id = $1`, "B"),
|
|
"B's docs survive the aborted eviction")
|
|
|
|
// The next healthy push evicts B.
|
|
fake.genOverride = nil
|
|
res, err = sync.pushVectors(ctx, false, nil, nil)
|
|
require.NoError(t, err, "healthy push")
|
|
assert.Equal(t, 1, res.SessionsEvicted)
|
|
assert.Equal(t, 0, countRows(t, pg,
|
|
`SELECT COUNT(*) FROM vector_documents WHERE session_id = $1`, "B"))
|
|
}
|
|
|
|
// TestVectorPushProjectScope pins the project-scope gate: a filtered push
|
|
// (projects=[alpha]) must leave vector state for an out-of-scope beta session
|
|
// already in PG untouched, while still pushing the in-scope alpha session's
|
|
// changes. The vectors.db source carries every session regardless of scope, so
|
|
// without the gate the beta doc would be re-pushed under a filtered push.
|
|
func TestVectorPushProjectScope(t *testing.T) {
|
|
pgURL := testPGURL(t)
|
|
sync, localDB, pg := newVectorPushTestSync(
|
|
t, pgURL, "agentsview_vector_push_scope_test")
|
|
ctx := context.Background()
|
|
|
|
seedVectorSessionProject(t, localDB, "alpha", "alpha")
|
|
seedVectorSessionProject(t, localDB, "beta", "beta")
|
|
|
|
fake := &fakeVectorSource{
|
|
gen: VectorGenerationInfo{Fingerprint: "fp-scope", Model: "m", Dimension: 4},
|
|
hasGen: true,
|
|
hashes: map[string]string{"alpha": "ha", "beta": "hb"},
|
|
docs: map[string][]VectorPushDoc{
|
|
"alpha": {vdoc("alpha", "alpha#0", 0, "a-orig", "ha1", []float32{1, 0, 0, 0})},
|
|
"beta": {vdoc("beta", "beta#0", 0, "b-orig", "hb1", []float32{0, 0, 0, 1})},
|
|
},
|
|
}
|
|
sync.vectorSource = fake
|
|
|
|
// Unfiltered push lands both alpha and beta vectors in PG.
|
|
_, err := sync.Push(ctx, false, nil)
|
|
require.NoError(t, err, "unfiltered push")
|
|
|
|
// Restrict the push to project alpha and change both sessions locally.
|
|
sync.projects = []string{"alpha"}
|
|
fake.hashes["alpha"] = "ha2"
|
|
fake.docs["alpha"] = []VectorPushDoc{
|
|
vdoc("alpha", "alpha#0", 0, "a-updated", "ha2", []float32{0, 1, 0, 0}),
|
|
}
|
|
fake.hashes["beta"] = "hb2"
|
|
fake.docs["beta"] = []VectorPushDoc{
|
|
vdoc("beta", "beta#0", 0, "b-updated", "hb2", []float32{0, 1, 0, 0}),
|
|
}
|
|
|
|
res, err := sync.pushVectors(ctx, false, nil, nil)
|
|
require.NoError(t, err, "filtered vector push")
|
|
assert.Equal(t, 1, res.SessionsPushed, "only in-scope alpha pushed")
|
|
assert.Equal(t, 0, res.Conflicts, "beta is out of scope, not a conflict")
|
|
assert.Equal(t, 0, res.SessionsEvicted, "beta must not be evicted")
|
|
|
|
var alphaContent, betaContent string
|
|
require.NoError(t, pg.QueryRow(
|
|
`SELECT content FROM vector_documents WHERE doc_key = $1`, "alpha#0",
|
|
).Scan(&alphaContent))
|
|
require.NoError(t, pg.QueryRow(
|
|
`SELECT content FROM vector_documents WHERE doc_key = $1`, "beta#0",
|
|
).Scan(&betaContent))
|
|
assert.Equal(t, "a-updated", alphaContent, "in-scope alpha updated")
|
|
assert.Equal(t, "b-orig", betaContent, "out-of-scope beta untouched")
|
|
|
|
genID, _, ok, err := LookupVectorGeneration(ctx, pg, "fp-scope")
|
|
require.NoError(t, err)
|
|
require.True(t, ok)
|
|
var betaHash string
|
|
require.NoError(t, pg.QueryRow(
|
|
`SELECT doc_agg_hash FROM vector_push_state
|
|
WHERE generation_id = $1 AND session_id = $2`, genID, "beta",
|
|
).Scan(&betaHash))
|
|
assert.Equal(t, "hb", betaHash, "beta push state left at its original hash")
|
|
}
|
|
|
|
// TestVectorPushLocalProjectMoveOutOfScope pins that filtered pushes scope
|
|
// local candidates by their LIVE local project, not the (possibly stale) PG
|
|
// project. A session pushed in scope (project alpha) then moved out of scope
|
|
// locally (alpha->beta) keeps its old PG project because the filtered session
|
|
// push skips it; scoping by PG would let the next `pg push --projects alpha`
|
|
// re-push or evict its vectors. Scoping by the local project must leave them
|
|
// untouched.
|
|
func TestVectorPushLocalProjectMoveOutOfScope(t *testing.T) {
|
|
pgURL := testPGURL(t)
|
|
sync, localDB, pg := newVectorPushTestSync(
|
|
t, pgURL, "agentsview_vector_push_localmove_test")
|
|
ctx := context.Background()
|
|
|
|
seedVectorSessionProject(t, localDB, "mover", "alpha")
|
|
|
|
fake := &fakeVectorSource{
|
|
gen: VectorGenerationInfo{Fingerprint: "fp-move", Model: "m", Dimension: 4},
|
|
hasGen: true,
|
|
hashes: map[string]string{"mover": "h1"},
|
|
docs: map[string][]VectorPushDoc{
|
|
"mover": {vdoc("mover", "mover#0", 0, "orig", "hc1", []float32{1, 0, 0, 0})},
|
|
},
|
|
}
|
|
sync.vectorSource = fake
|
|
|
|
// Unfiltered push lands the alpha vector in PG (PG project = alpha).
|
|
_, err := sync.Push(ctx, false, nil)
|
|
require.NoError(t, err, "unfiltered push")
|
|
|
|
// The session moves to beta locally, but its PG sessions.project stays alpha
|
|
// (a filtered session push skips it). The local doc content also changes.
|
|
require.NoError(t, localDB.UpsertSession(db.Session{
|
|
ID: "mover",
|
|
Project: "beta",
|
|
Machine: "test-machine",
|
|
Agent: "claude",
|
|
MessageCount: 1,
|
|
UserMessageCount: 1,
|
|
CreatedAt: "2026-01-01T00:00:00Z",
|
|
}), "move mover to beta")
|
|
fake.hashes["mover"] = "h2"
|
|
fake.docs["mover"] = []VectorPushDoc{
|
|
vdoc("mover", "mover#0", 0, "updated", "hc2", []float32{0, 1, 0, 0}),
|
|
}
|
|
|
|
var pgProject string
|
|
require.NoError(t, pg.QueryRow(
|
|
`SELECT project FROM sessions WHERE id = $1`, "mover",
|
|
).Scan(&pgProject))
|
|
require.Equal(t, "alpha", pgProject, "PG project is still stale alpha")
|
|
|
|
// A push filtered to alpha must treat mover (now local beta) as out of scope.
|
|
sync.projects = []string{"alpha"}
|
|
res, err := sync.pushVectors(ctx, false, nil, nil)
|
|
require.NoError(t, err, "filtered vector push after local move")
|
|
assert.Equal(t, 0, res.SessionsPushed, "moved-out session not pushed")
|
|
assert.Equal(t, 0, res.SessionsEvicted, "moved-out session not evicted")
|
|
assert.Equal(t, 0, res.Conflicts, "moved-out session is not a conflict")
|
|
|
|
var content string
|
|
require.NoError(t, pg.QueryRow(
|
|
`SELECT content FROM vector_documents WHERE doc_key = $1`, "mover#0",
|
|
).Scan(&content))
|
|
assert.Equal(t, "orig", content, "out-of-scope vector left untouched")
|
|
|
|
genID, _, ok, err := LookupVectorGeneration(ctx, pg, "fp-move")
|
|
require.NoError(t, err)
|
|
require.True(t, ok)
|
|
var stateHash string
|
|
require.NoError(t, pg.QueryRow(
|
|
`SELECT doc_agg_hash FROM vector_push_state
|
|
WHERE generation_id = $1 AND session_id = $2`, genID, "mover",
|
|
).Scan(&stateHash))
|
|
assert.Equal(t, "h1", stateHash, "push state left at its original hash")
|
|
}
|
|
|
|
// TestVectorPushSkipsDocExportForMissingSession pins that a local-only session
|
|
// with no PG sessions row is skipped without exporting its docs. Doc export
|
|
// (content + chunk decode) must run only after the in-tx existence probe
|
|
// passes, so a session that always skips never re-exports docs each push.
|
|
func TestVectorPushSkipsDocExportForMissingSession(t *testing.T) {
|
|
pgURL := testPGURL(t)
|
|
sync, localDB, _ := newVectorPushTestSync(
|
|
t, pgURL, "agentsview_vector_push_skipexport_test")
|
|
ctx := context.Background()
|
|
|
|
// "A" is a real session (pushed to PG). "ghost" is local-only: it never gets
|
|
// a PG sessions row, so the vector push must skip it.
|
|
seedVectorSession(t, localDB, "A")
|
|
|
|
fake := &fakeVectorSource{
|
|
gen: VectorGenerationInfo{Fingerprint: "fp-skip", Model: "m", Dimension: 4},
|
|
hasGen: true,
|
|
hashes: map[string]string{"A": "ha", "ghost": "hg"},
|
|
docs: map[string][]VectorPushDoc{
|
|
"A": {vdoc("A", "A#0", 0, "ca", "hca", []float32{1, 0, 0, 0})},
|
|
"ghost": {vdoc("ghost", "g#0", 0, "cg", "hcg", []float32{0, 1, 0, 0})},
|
|
},
|
|
}
|
|
sync.vectorSource = fake
|
|
|
|
// Push seeds A's PG sessions row (session phase) then runs the vector phase.
|
|
// ghost is absent from localDB, so it never gets a PG sessions row.
|
|
res, err := sync.Push(ctx, false, nil)
|
|
require.NoError(t, err, "Push")
|
|
assert.Equal(t, 1, res.Vectors.SessionsPushed, "only A lands")
|
|
assert.Equal(t, 0, res.Vectors.Conflicts, "missing session is not a conflict")
|
|
|
|
assert.Equal(t, 1, fake.docsCalls["A"], "A's docs exported once")
|
|
assert.Zero(t, fake.docsCalls["ghost"],
|
|
"missing session's docs never exported")
|
|
}
|
|
|
|
// TestVectorPushOrphanStateEvicted pins finding-3 rule d: a vector_push_state
|
|
// row whose sessions row has vanished is evicted regardless of owner marker.
|
|
func TestVectorPushOrphanStateEvicted(t *testing.T) {
|
|
pgURL := testPGURL(t)
|
|
sync, localDB, pg := newVectorPushTestSync(
|
|
t, pgURL, "agentsview_vector_push_orphan_test")
|
|
ctx := context.Background()
|
|
|
|
seedVectorSession(t, localDB, "A")
|
|
seedVectorSession(t, localDB, "B")
|
|
|
|
fake := &fakeVectorSource{
|
|
gen: VectorGenerationInfo{Fingerprint: "fp-or", Model: "m", Dimension: 4},
|
|
hasGen: true,
|
|
hashes: map[string]string{"A": "ha", "B": "hb"},
|
|
docs: map[string][]VectorPushDoc{
|
|
"A": {vdoc("A", "oA0", 0, "ca", "hca", []float32{1, 0, 0, 0})},
|
|
"B": {vdoc("B", "oB0", 0, "cb", "hcb", []float32{0, 0, 0, 1})},
|
|
},
|
|
}
|
|
sync.vectorSource = fake
|
|
_, err := sync.Push(ctx, false, nil)
|
|
require.NoError(t, err, "first Push")
|
|
genID, _, ok, err := LookupVectorGeneration(ctx, pg, "fp-or")
|
|
require.NoError(t, err)
|
|
require.True(t, ok)
|
|
|
|
// Session A's row disappears (e.g. deleted) but its vector state lingers,
|
|
// and A leaves local so it is an eviction candidate.
|
|
_, err = pg.Exec(`DELETE FROM sessions WHERE id = $1`, "A")
|
|
require.NoError(t, err, "delete session row")
|
|
delete(fake.hashes, "A")
|
|
delete(fake.docs, "A")
|
|
|
|
res, err := sync.pushVectors(ctx, false, nil, nil)
|
|
require.NoError(t, err, "evict orphan")
|
|
assert.Equal(t, 1, res.SessionsEvicted)
|
|
assert.Equal(t, 0, res.Conflicts, "orphaned state is not a conflict")
|
|
|
|
assert.Equal(t, 0, countRows(t, pg,
|
|
`SELECT COUNT(*) FROM vector_push_state
|
|
WHERE generation_id = $1 AND session_id = $2`, genID, "A"))
|
|
assert.Equal(t, 0, countRows(t, pg,
|
|
`SELECT COUNT(*) FROM `+vectorChunkTable(genID)+` WHERE doc_key = $1`, "oA0"))
|
|
assert.Equal(t, 0, countRows(t, pg,
|
|
`SELECT COUNT(*) FROM vector_documents WHERE session_id = $1`, "A"))
|
|
}
|
|
|
|
// TestVectorPushDefersSessionWhenExportDiverges pins the replacement-path
|
|
// guard against concurrent local rebuilds: when a session's export hash no
|
|
// longer matches the hash the delta scan read (an embeddings build rewrote
|
|
// the local index between the two reads), the session is deferred with
|
|
// nothing written — the previously pushed PG chunks and push state survive
|
|
// untouched. Without the guard, the push would replace valid PG vectors with
|
|
// the partial view and record the stale scan hash as current, which a
|
|
// same-fingerprint rebuild (same content, same hash) would never repair.
|
|
func TestVectorPushDefersSessionWhenExportDiverges(t *testing.T) {
|
|
pgURL := testPGURL(t)
|
|
sync, localDB, pg := newVectorPushTestSync(
|
|
t, pgURL, "agentsview_vector_push_diverge_test")
|
|
ctx := context.Background()
|
|
|
|
seedVectorSession(t, localDB, "A")
|
|
|
|
source := &fakeVectorSource{
|
|
gen: VectorGenerationInfo{Fingerprint: "fp-div", Model: "m", Dimension: 4},
|
|
hasGen: true,
|
|
hashes: map[string]string{"A": "ha"},
|
|
docs: map[string][]VectorPushDoc{
|
|
"A": {vdoc("A", "A#0", 0, "c1", "h1",
|
|
[]float32{1, 0, 0, 0}, []float32{0, 1, 0, 0})},
|
|
},
|
|
}
|
|
sync.vectorSource = source
|
|
|
|
_, err := sync.Push(ctx, false, nil)
|
|
require.NoError(t, err, "seed Push")
|
|
genID, _, ok, err := LookupVectorGeneration(ctx, pg, "fp-div")
|
|
require.NoError(t, err)
|
|
require.True(t, ok)
|
|
require.Equal(t, 2,
|
|
countRows(t, pg, `SELECT COUNT(*) FROM `+vectorChunkTable(genID)))
|
|
|
|
// The delta scan sees a changed session, but by export time the local
|
|
// index has moved again: SessionDocs returns a shrunken doc set whose
|
|
// hash differs from the scan's.
|
|
source.hashes = map[string]string{"A": "ha-changed"}
|
|
source.exportHashes = map[string]string{"A": "ha-mid-rebuild"}
|
|
source.docs = map[string][]VectorPushDoc{
|
|
"A": {vdoc("A", "A#0", 0, "c1", "h1", []float32{9, 9, 9, 9})},
|
|
}
|
|
|
|
res, err := sync.pushVectors(ctx, false, nil, nil)
|
|
require.NoError(t, err, "diverged pushVectors")
|
|
assert.Equal(t, 1, res.SessionsDeferred, "diverged session must defer")
|
|
assert.Equal(t, 0, res.SessionsPushed)
|
|
|
|
assert.Equal(t, 2,
|
|
countRows(t, pg, `SELECT COUNT(*) FROM `+vectorChunkTable(genID)),
|
|
"previously pushed chunks must survive the deferral")
|
|
assert.Equal(t, 1, countRows(t, pg, `
|
|
SELECT COUNT(*) FROM vector_push_state
|
|
WHERE generation_id = $1 AND session_id = 'A' AND doc_agg_hash = 'ha'`,
|
|
genID), "push state must keep the last successfully exported hash")
|
|
|
|
// Once the local index settles (export hash matches the scan again), the
|
|
// deferred session is re-derived and pushed.
|
|
source.exportHashes = nil
|
|
res, err = sync.pushVectors(ctx, false, nil, nil)
|
|
require.NoError(t, err, "settled pushVectors")
|
|
assert.Equal(t, 1, res.SessionsPushed)
|
|
assert.Equal(t, 1,
|
|
countRows(t, pg, `SELECT COUNT(*) FROM `+vectorChunkTable(genID)))
|
|
}
|
|
|
|
// TestVectorEvictionRechecksOwnershipInTx pins the eviction-path TOCTOU
|
|
// guard: ownership read by the delta scan can be minutes stale, so each
|
|
// eviction transaction re-probes the sessions row FOR UPDATE. A session
|
|
// another pusher claimed after the scan is skipped as a conflict — its new
|
|
// owner's chunks and push state survive — while a session whose row vanished
|
|
// is still evicted.
|
|
func TestVectorEvictionRechecksOwnershipInTx(t *testing.T) {
|
|
pgURL := testPGURL(t)
|
|
sync, localDB, pg := newVectorPushTestSync(
|
|
t, pgURL, "agentsview_vector_evict_recheck_test")
|
|
ctx := context.Background()
|
|
|
|
seedVectorSession(t, localDB, "A")
|
|
seedVectorSession(t, localDB, "B")
|
|
|
|
sync.vectorSource = &fakeVectorSource{
|
|
gen: VectorGenerationInfo{Fingerprint: "fp-ev", Model: "m", Dimension: 4},
|
|
hasGen: true,
|
|
hashes: map[string]string{"A": "ha", "B": "hb"},
|
|
docs: map[string][]VectorPushDoc{
|
|
"A": {vdoc("A", "A#0", 0, "c1", "h1", []float32{1, 0, 0, 0})},
|
|
"B": {vdoc("B", "B#0", 0, "c2", "h2", []float32{0, 1, 0, 0})},
|
|
},
|
|
}
|
|
|
|
_, err := sync.Push(ctx, false, nil)
|
|
require.NoError(t, err, "seed Push")
|
|
genID, _, ok, err := LookupVectorGeneration(ctx, pg, "fp-ev")
|
|
require.NoError(t, err)
|
|
require.True(t, ok)
|
|
|
|
// Simulate a concurrent pusher claiming A after this push's delta scan
|
|
// (which would have seen it as ours), and B's session row vanishing.
|
|
_, err = pg.ExecContext(ctx, `
|
|
UPDATE sessions SET owner_marker = 'other-marker', machine = 'other-machine'
|
|
WHERE id = 'A'`)
|
|
require.NoError(t, err, "reassign A")
|
|
_, err = pg.ExecContext(ctx, `DELETE FROM sessions WHERE id = 'B'`)
|
|
require.NoError(t, err, "delete B")
|
|
|
|
owner, err := sync.vectorOwnerIdentity(ctx)
|
|
require.NoError(t, err)
|
|
allGenIDs, err := sync.allVectorGenerationIDs(ctx)
|
|
require.NoError(t, err)
|
|
genIDs, err := sync.existingChunkGenerations(ctx, allGenIDs)
|
|
require.NoError(t, err)
|
|
scope := vectorPushScope{
|
|
gen: vectorGeneration{id: genID},
|
|
fingerprint: "fp-ev",
|
|
genIDs: genIDs,
|
|
owner: owner,
|
|
}
|
|
|
|
var res VectorPushResult
|
|
require.NoError(t,
|
|
sync.evictVectorSessions(ctx, scope, []string{"A", "B"}, &res))
|
|
|
|
assert.Equal(t, 1, res.Conflicts, "reclaimed session is a conflict")
|
|
assert.Equal(t, 1, res.SessionsEvicted, "vanished session is evicted")
|
|
assert.Equal(t, 1, countRows(t, pg, `
|
|
SELECT COUNT(*) FROM vector_push_state
|
|
WHERE generation_id = $1 AND session_id = 'A'`, genID),
|
|
"the new owner's push state survives")
|
|
assert.Equal(t, 1, countRows(t, pg, `
|
|
SELECT COUNT(*) FROM `+vectorChunkTable(genID)+` c
|
|
WHERE c.doc_key IN (
|
|
SELECT doc_key FROM vector_documents WHERE session_id = 'A')`),
|
|
"the new owner's chunks survive")
|
|
assert.Equal(t, 0, countRows(t, pg, `
|
|
SELECT COUNT(*) FROM vector_push_state
|
|
WHERE generation_id = $1 AND session_id = 'B'`, genID))
|
|
}
|
|
|
|
// TestVectorPushEvictionDefersFailedSession pins that failedSessions gates
|
|
// eviction, not just replacement: a failed session with zero embedded docs is
|
|
// absent from the local hash map and would otherwise be evicted — vector
|
|
// state running ahead of the sessions/messages rows its session-phase push
|
|
// failed to write. It must defer instead, and evict on the next clean push.
|
|
func TestVectorPushEvictionDefersFailedSession(t *testing.T) {
|
|
pgURL := testPGURL(t)
|
|
sync, localDB, pg := newVectorPushTestSync(
|
|
t, pgURL, "agentsview_vector_evict_failed_test")
|
|
ctx := context.Background()
|
|
|
|
seedVectorSession(t, localDB, "A")
|
|
|
|
fake := &fakeVectorSource{
|
|
gen: VectorGenerationInfo{Fingerprint: "fp-evf", Model: "m", Dimension: 4},
|
|
hasGen: true,
|
|
hashes: map[string]string{"A": "ha"},
|
|
docs: map[string][]VectorPushDoc{
|
|
"A": {vdoc("A", "A#0", 0, "c1", "h1", []float32{1, 0, 0, 0})},
|
|
},
|
|
}
|
|
sync.vectorSource = fake
|
|
|
|
_, err := sync.Push(ctx, false, nil)
|
|
require.NoError(t, err, "seed Push")
|
|
genID, _, ok, err := LookupVectorGeneration(ctx, pg, "fp-evf")
|
|
require.NoError(t, err)
|
|
require.True(t, ok)
|
|
|
|
// A's embedded docs all vanish locally while its session push fails.
|
|
fake.hashes = map[string]string{}
|
|
fake.docs = map[string][]VectorPushDoc{}
|
|
|
|
res, err := sync.pushVectors(ctx, false, map[string]struct{}{"A": {}}, nil)
|
|
require.NoError(t, err, "failed-session pushVectors")
|
|
assert.Equal(t, 0, res.SessionsEvicted, "failed session must not be evicted")
|
|
assert.Equal(t, 1, res.SessionsDeferred)
|
|
assert.Equal(t, 1,
|
|
countRows(t, pg, `SELECT COUNT(*) FROM `+vectorChunkTable(genID)),
|
|
"failed session keeps its vectors")
|
|
assert.Equal(t, 1, countRows(t, pg, `
|
|
SELECT COUNT(*) FROM vector_push_state
|
|
WHERE generation_id = $1 AND session_id = 'A'`, genID))
|
|
|
|
res, err = sync.pushVectors(ctx, false, nil, nil)
|
|
require.NoError(t, err, "clean pushVectors")
|
|
assert.Equal(t, 1, res.SessionsEvicted, "next clean push evicts")
|
|
assert.Equal(t, 0,
|
|
countRows(t, pg, `SELECT COUNT(*) FROM `+vectorChunkTable(genID)))
|
|
}
|
|
|
|
// TestVectorPushFilteredEvictionScopesByLocalProject pins the eviction-scope
|
|
// partition for filtered pushes: an eviction candidate absent from the vector
|
|
// hash map can still have a live local sessions row (zero embedded docs), and
|
|
// that row's LIVE project — not the stale PG project — decides scope. Only
|
|
// candidates genuinely absent from the local sessions table fall back to PG
|
|
// project scoping.
|
|
func TestVectorPushFilteredEvictionScopesByLocalProject(t *testing.T) {
|
|
pgURL := testPGURL(t)
|
|
sync, localDB, pg := newVectorPushTestSync(
|
|
t, pgURL, "agentsview_vector_evict_scope_test")
|
|
ctx := context.Background()
|
|
|
|
seedVectorSessionProject(t, localDB, "moved", "alpha")
|
|
seedVectorSessionProject(t, localDB, "gone-out", "beta")
|
|
seedVectorSessionProject(t, localDB, "gone-in", "alpha")
|
|
|
|
fake := &fakeVectorSource{
|
|
gen: VectorGenerationInfo{Fingerprint: "fp-evs", Model: "m", Dimension: 4},
|
|
hasGen: true,
|
|
hashes: map[string]string{"moved": "hm", "gone-out": "ho", "gone-in": "hi"},
|
|
docs: map[string][]VectorPushDoc{
|
|
"moved": {vdoc("moved", "m#0", 0, "cm", "hm1", []float32{1, 0, 0, 0})},
|
|
"gone-out": {vdoc("gone-out", "o#0", 0, "co", "ho1", []float32{0, 1, 0, 0})},
|
|
"gone-in": {vdoc("gone-in", "i#0", 0, "ci", "hi1", []float32{0, 0, 1, 0})},
|
|
},
|
|
}
|
|
sync.vectorSource = fake
|
|
|
|
// Unfiltered push lands all three sessions' vectors and PG rows.
|
|
_, err := sync.Push(ctx, false, nil)
|
|
require.NoError(t, err, "unfiltered push")
|
|
genID, _, ok, err := LookupVectorGeneration(ctx, pg, "fp-evs")
|
|
require.NoError(t, err)
|
|
require.True(t, ok)
|
|
|
|
// "moved" changes project locally (alpha -> beta) — its PG project stays
|
|
// alpha because a filtered session push skips it. The others vanish from
|
|
// the local archive entirely. All three drop out of the embedded set.
|
|
require.NoError(t, localDB.UpsertSession(db.Session{
|
|
ID: "moved", Project: "beta", Machine: "test-machine",
|
|
Agent: "claude", MessageCount: 1, UserMessageCount: 1,
|
|
CreatedAt: "2026-01-01T00:00:00Z",
|
|
}), "move session to beta")
|
|
require.NoError(t, localDB.DeleteSession("gone-out"))
|
|
require.NoError(t, localDB.DeleteSession("gone-in"))
|
|
fake.hashes = map[string]string{}
|
|
fake.docs = map[string][]VectorPushDoc{}
|
|
|
|
sync.projects = []string{"alpha"}
|
|
res, err := sync.pushVectors(ctx, false, nil, nil)
|
|
require.NoError(t, err, "filtered pushVectors")
|
|
|
|
assert.Equal(t, 1, res.SessionsEvicted,
|
|
"only the locally-absent alpha session is evicted")
|
|
chunkCount := func(sessionID string) int {
|
|
return countRows(t, pg, `
|
|
SELECT COUNT(*) FROM `+vectorChunkTable(genID)+` c
|
|
WHERE c.doc_key IN (
|
|
SELECT doc_key FROM vector_documents WHERE session_id = $1)`,
|
|
sessionID)
|
|
}
|
|
assert.Equal(t, 1, chunkCount("moved"),
|
|
"session that moved out of the filter locally is untouched")
|
|
assert.Equal(t, 1, chunkCount("gone-out"),
|
|
"locally-absent session with out-of-filter PG project is untouched")
|
|
assert.Equal(t, 0, chunkCount("gone-in"),
|
|
"locally-absent session with in-filter PG project is evicted")
|
|
}
|
|
|
|
// TestVectorPushSkipsOnInsufficientPrivilege pins the restricted-role degrade
|
|
// path: a schema provisioned by a privileged role (core tables exist) whose
|
|
// push role cannot CREATE means the vector base DDL fails with SQLSTATE
|
|
// 42501. That must skip the vector phase with a reason — matching a database
|
|
// without pgvector — not fail a push whose session phase succeeded.
|
|
func TestVectorPushSkipsOnInsufficientPrivilege(t *testing.T) {
|
|
pgURL := testPGURL(t)
|
|
const schema = "agentsview_vector_privilege_test"
|
|
const role = "agentsview_vector_restricted"
|
|
const rolePassword = "agentsview_vector_restricted_pw"
|
|
|
|
admin, err := Open(pgURL, schema, true)
|
|
require.NoError(t, err, "Open admin")
|
|
t.Cleanup(func() { _ = admin.Close() })
|
|
|
|
ctx := context.Background()
|
|
_, err = admin.Exec(`DROP SCHEMA IF EXISTS ` + schema + ` CASCADE`)
|
|
require.NoError(t, err, "drop schema")
|
|
require.NoError(t, EnsureSchema(ctx, admin, schema))
|
|
unavailable, err := ensureVectorBaseSchemaPG(ctx, admin)
|
|
require.NoError(t, err)
|
|
if unavailable != "" {
|
|
t.Skip(unavailable)
|
|
}
|
|
|
|
// Simulate a schema provisioned by an older, pre-vector agentsview:
|
|
// core tables exist but the vector tables do not, so the restricted
|
|
// role's own setup must CREATE them — and cannot.
|
|
for _, tbl := range []string{
|
|
"vector_push_state", "vector_documents",
|
|
"vector_generation_machines", "vector_generations",
|
|
} {
|
|
_, err = admin.Exec(`DROP TABLE IF EXISTS ` + tbl + ` CASCADE`)
|
|
require.NoError(t, err, tbl)
|
|
}
|
|
|
|
_, _ = admin.Exec(`DROP OWNED BY ` + role) // clear grants from prior runs
|
|
_, _ = admin.Exec(`DROP ROLE IF EXISTS ` + role)
|
|
_, err = admin.Exec(
|
|
`CREATE ROLE ` + role + ` LOGIN PASSWORD '` + rolePassword + `'`)
|
|
require.NoError(t, err, "create restricted role")
|
|
t.Cleanup(func() {
|
|
_, _ = admin.Exec(`DROP SCHEMA IF EXISTS ` + schema + ` CASCADE`)
|
|
_, _ = admin.Exec(`DROP OWNED BY ` + role)
|
|
_, _ = admin.Exec(`DROP ROLE IF EXISTS ` + role)
|
|
})
|
|
for _, grant := range []string{
|
|
`GRANT USAGE ON SCHEMA ` + schema + ` TO ` + role,
|
|
`GRANT SELECT, INSERT, UPDATE, DELETE ON ALL TABLES IN SCHEMA ` +
|
|
schema + ` TO ` + role,
|
|
} {
|
|
_, err = admin.Exec(grant)
|
|
require.NoError(t, err, grant)
|
|
}
|
|
|
|
restrictedURL, err := url.Parse(pgURL)
|
|
require.NoError(t, err)
|
|
restrictedURL.User = url.UserPassword(role, rolePassword)
|
|
restricted, err := Open(restrictedURL.String(), schema, true)
|
|
require.NoError(t, err, "Open restricted")
|
|
t.Cleanup(func() { _ = restricted.Close() })
|
|
|
|
localDB, err := db.Open(filepath.Join(t.TempDir(), "local.db"))
|
|
require.NoError(t, err, "db.Open")
|
|
t.Cleanup(func() { _ = localDB.Close() })
|
|
|
|
sync := &Sync{
|
|
pg: restricted,
|
|
local: localDB,
|
|
machine: "test-machine",
|
|
schema: schema,
|
|
schemaDone: true,
|
|
}
|
|
sync.vectorSource = &fakeVectorSource{
|
|
gen: VectorGenerationInfo{
|
|
Fingerprint: "fp-priv", Model: "m", Dimension: 4,
|
|
},
|
|
hasGen: true,
|
|
hashes: map[string]string{},
|
|
}
|
|
|
|
res, err := sync.pushVectors(ctx, false, nil, nil)
|
|
require.NoError(t, err, "privilege failure must skip, not fail the push")
|
|
assert.True(t, res.Skipped)
|
|
assert.Contains(t, res.SkippedReason, "privileges")
|
|
}
|