fix: preserve ochre teardown metadata for auto-reconcile (#863)

Teardown no longer purges the index registry or KB/DataSource records by
default, mirroring how they already survive a restart; a --purge-metadata
flag restores the old full wipe for an intentional destroy. On the next
appliance connect, a new reconcile pass re-creates any registry index whose
pg table is missing and re-enqueues ingest from its persisted SourceSpecs,
so a torn-down appliance heals without operator action and keeps its KB id.

cmd/spinifex/cmd/admin_ochre_appliance.go

@@ -29,10 +29,13 @@ var ochreApplianceTeardownCmd = &cobra.Command{
 instance, this node's VPC host port, and its KV record.
 
 This is a destructive, cluster-wide operator action -- every account's vector
-index data stored in the appliance is lost with it, and it is not
-automatically rebuilt. A future daemon startup re-provisions a fresh
-appliance (a new instance, a new password) only if Ochre remains enabled;
-existing index registry records are not restored.
+index data stored in the appliance is lost with it. By default the index
+registry and KB/DataSource metadata survive, exactly as they survive a
+restart: a future daemon startup re-provisions a fresh appliance (a new
+instance, a new password) and reconciles each surviving index by re-creating
+it and re-ingesting from its recorded source, with no operator action and no
+change to its id. --purge-metadata additionally wipes that metadata for an
+intentional full destroy, after which nothing is auto-restored.
 
 Requires --confirm.`,
 	Run: runOchreApplianceTeardown,
@@ -44,40 +47,47 @@ func init() {
 
 	ochreApplianceTeardownCmd.Flags().Bool("confirm", false,
 		"Required: confirms you intend to destroy the shared platform appliance and every account's vector index data with it")
+	ochreApplianceTeardownCmd.Flags().Bool("purge-metadata", false,
+		"Also wipe the index registry and KB/DataSource records, disabling auto-restore on the next re-provision")
 }
 
 // applianceTeardownFn indirects the NATS call so runApplianceTeardownGuarded
 // is testable without a live daemon, mirroring vectorServiceFn/
 // endpointServiceFn.
-var applianceTeardownFn = func(ctx context.Context) error {
+var applianceTeardownFn = func(ctx context.Context, purgeMetadata bool) error {
 	_, nc, err := loadConfigAndConnectFn()
 	if err != nil {
 		return err
 	}
 	defer nc.Close()
 	_, err = utils.NATSRequest[handlers_ochrevector.TeardownApplianceResponse](ctx, nc,
-		handlers_ochrevector.SubjectTeardownAppliance, &handlers_ochrevector.TeardownApplianceRequest{}, applianceTeardownTimeout, utils.GlobalAccountID)
+		handlers_ochrevector.SubjectTeardownAppliance, &handlers_ochrevector.TeardownApplianceRequest{PurgeMetadata: purgeMetadata},
+		applianceTeardownTimeout, utils.GlobalAccountID)
 	return err
 }
 
 // runApplianceTeardownGuarded is the testable core of 'ochre appliance
 // teardown': it refuses without --confirm before making any NATS call, so a
 // missing flag never even opens a connection to the cluster.
-func runApplianceTeardownGuarded(ctx context.Context, confirmed bool, teardown func(context.Context) error) (string, error) {
+func runApplianceTeardownGuarded(ctx context.Context, confirmed, purgeMetadata bool, teardown func(context.Context, bool) error) (string, error) {
 	if !confirmed {
 		return "", errors.New("refusing to tear down the shared platform appliance without --confirm: " +
 			"this destroys every account's vector index data and is not automatically rebuilt")
 	}
-	if err := teardown(ctx); err != nil {
+	if err := teardown(ctx, purgeMetadata); err != nil {
 		return "", err
 	}
+	if purgeMetadata {
+		return "✅ Platform appliance torn down, index/KB metadata purged.", nil
+	}
 	return "✅ Platform appliance torn down.", nil
 }
 
 func runOchreApplianceTeardown(cmd *cobra.Command, _ []string) {
 	confirmed, _ := cmd.Flags().GetBool("confirm")
+	purgeMetadata, _ := cmd.Flags().GetBool("purge-metadata")
 
-	msg, err := runApplianceTeardownGuarded(context.Background(), confirmed, applianceTeardownFn)
+	msg, err := runApplianceTeardownGuarded(context.Background(), confirmed, purgeMetadata, applianceTeardownFn)
 	if err != nil {
 		fmt.Fprintf(os.Stderr, "%v\n", err)
 		ochreExit(1)

cmd/spinifex/cmd/admin_ochre_appliance_test.go

@@ -9,6 +9,7 @@ import (
 	"errors"
 	"testing"
 
+	"github.com/mulgadc/spinifex/spinifex/testutil"
 	"github.com/stretchr/testify/require"
 )
 
@@ -17,12 +18,12 @@ import (
 // reach a NATS call, let alone destroy the appliance.
 func TestRunApplianceTeardownGuarded_RefusesWithoutConfirm(t *testing.T) {
 	called := false
-	teardown := func(context.Context) error {
+	teardown := func(context.Context, bool) error {
 		called = true
 		return nil
 	}
 
-	_, err := runApplianceTeardownGuarded(context.Background(), false, teardown)
+	_, err := runApplianceTeardownGuarded(context.Background(), false, false, teardown)
 	require.Error(t, err)
 	require.ErrorContains(t, err, "--confirm")
 	require.False(t, called, "teardown must not run without --confirm")
@@ -32,12 +33,12 @@ func TestRunApplianceTeardownGuarded_RefusesWithoutConfirm(t *testing.T) {
 // drives the teardown call through and reports success.
 func TestRunApplianceTeardownGuarded_ConfirmedInvokesTeardown(t *testing.T) {
 	called := false
-	teardown := func(context.Context) error {
+	teardown := func(context.Context, bool) error {
 		called = true
 		return nil
 	}
 
-	msg, err := runApplianceTeardownGuarded(context.Background(), true, teardown)
+	msg, err := runApplianceTeardownGuarded(context.Background(), true, false, teardown)
 	require.NoError(t, err)
 	require.True(t, called)
 	require.Contains(t, msg, "torn down")
@@ -47,19 +48,49 @@ func TestRunApplianceTeardownGuarded_ConfirmedInvokesTeardown(t *testing.T) {
 // failure (e.g. the daemon reporting no appliance up) surfaces to the caller
 // rather than being swallowed behind a generic success message.
 func TestRunApplianceTeardownGuarded_TeardownErrorSurfaces(t *testing.T) {
-	teardown := func(context.Context) error {
+	teardown := func(context.Context, bool) error {
 		return errors.New("ochrevector: platform appliance is not enabled or not up on this node")
 	}
 
-	_, err := runApplianceTeardownGuarded(context.Background(), true, teardown)
+	_, err := runApplianceTeardownGuarded(context.Background(), true, false, teardown)
 	require.Error(t, err)
 	require.ErrorContains(t, err, "not enabled or not up")
 }
 
+// TestRunApplianceTeardownGuarded_PurgeMetadataThreadsThrough proves the
+// --purge-metadata flag reaches the teardown call and is reflected in the
+// success message, so an operator can tell which mode ran.
+func TestRunApplianceTeardownGuarded_PurgeMetadataThreadsThrough(t *testing.T) {
+	var gotPurge bool
+	teardown := func(_ context.Context, purgeMetadata bool) error {
+		gotPurge = purgeMetadata
+		return nil
+	}
+
+	msg, err := runApplianceTeardownGuarded(context.Background(), true, true, teardown)
+	require.NoError(t, err)
+	require.True(t, gotPurge, "purgeMetadata must reach the teardown call")
+	require.Contains(t, msg, "metadata purged")
+}
+
+// TestApplianceTeardownFn_ThreadsPurgeMetadataOverNATS drives the real
+// (unstubbed) applianceTeardownFn through runApplianceTeardownGuarded, so the
+// actual NATSRequest call -- including the PurgeMetadata field on the wire
+// request -- executes rather than a test double standing in for it. No
+// daemon is subscribed to the teardown subject in this test, so the call
+// fails fast; the point is that it's reached and built with PurgeMetadata set.
+func TestApplianceTeardownFn_ThreadsPurgeMetadataOverNATS(t *testing.T) {
+	_, nc, _ := testutil.StartTestJetStream(t)
+	stubConnect(t, fakeClusterConfig(), nc, nil)
+
+	_, err := runApplianceTeardownGuarded(context.Background(), true, true, applianceTeardownFn)
+	require.Error(t, err, "no daemon is subscribed to the teardown subject in this test")
+}
+
 // withApplianceTeardown swaps applianceTeardownFn for one that never dials
 // NATS, so the Run wrapper's real flag/exit control flow is exercised
 // without a live daemon.
-func withApplianceTeardown(t *testing.T, fn func(context.Context) error) {
+func withApplianceTeardown(t *testing.T, fn func(context.Context, bool) error) {
 	t.Helper()
 	orig := applianceTeardownFn
 	t.Cleanup(func() { applianceTeardownFn = orig })
@@ -67,7 +98,7 @@ func withApplianceTeardown(t *testing.T, fn func(context.Context) error) {
 }
 
 func TestRunOchreApplianceTeardown_MissingConfirmExits1(t *testing.T) {
-	withApplianceTeardown(t, func(context.Context) error {
+	withApplianceTeardown(t, func(context.Context, bool) error {
 		t.Fatal("teardown must not be called without --confirm")
 		return nil
 	})
@@ -78,7 +109,7 @@ func TestRunOchreApplianceTeardown_MissingConfirmExits1(t *testing.T) {
 }
 
 func TestRunOchreApplianceTeardown_ConfirmedPrintsSuccess(t *testing.T) {
-	withApplianceTeardown(t, func(context.Context) error { return nil })
+	withApplianceTeardown(t, func(context.Context, bool) error { return nil })
 
 	cmd := *ochreApplianceTeardownCmd
 	require.NoError(t, cmd.Flags().Set("confirm", "true"))
@@ -92,7 +123,7 @@ func TestRunOchreApplianceTeardown_ConfirmedPrintsSuccess(t *testing.T) {
 }
 
 func TestRunOchreApplianceTeardown_TeardownErrorExits1(t *testing.T) {
-	withApplianceTeardown(t, func(context.Context) error {
+	withApplianceTeardown(t, func(context.Context, bool) error {
 		return errors.New("ochrevector: platform appliance is not enabled or not up on this node")
 	})
 
@@ -102,3 +133,21 @@ func TestRunOchreApplianceTeardown_TeardownErrorExits1(t *testing.T) {
 	code := withOchreExitCapture(t, func() { runOchreApplianceTeardown(&cmd, nil) })
 	require.Equal(t, 1, code)
 }
+
+// TestRunOchreApplianceTeardown_PurgeMetadataFlagReachesTeardownFn proves
+// the --purge-metadata flag is read and passed through the Run wrapper.
+func TestRunOchreApplianceTeardown_PurgeMetadataFlagReachesTeardownFn(t *testing.T) {
+	var gotPurge bool
+	withApplianceTeardown(t, func(_ context.Context, purgeMetadata bool) error {
+		gotPurge = purgeMetadata
+		return nil
+	})
+
+	cmd := *ochreApplianceTeardownCmd
+	require.NoError(t, cmd.Flags().Set("confirm", "true"))
+	require.NoError(t, cmd.Flags().Set("purge-metadata", "true"))
+
+	code := withOchreExitCapture(t, func() { runOchreApplianceTeardown(&cmd, nil) })
+	require.Equal(t, -1, code)
+	require.True(t, gotPurge, "--purge-metadata must reach applianceTeardownFn")
+}

spinifex/daemon/ochre_deps.go

@@ -189,10 +189,62 @@ func (d *Daemon) startOchreVector() {
 	// (there is none today beyond observability) must never see a non-nil
 	// service whose subjects are not yet actually serving.
 	d.ochreVectorService = vectorService
+
+	// Reconcile before the scheduler starts sweeping, so a job this pass
+	// enqueues for a rebuilt index is already PENDING for the scheduler's
+	// first Sweep rather than waiting for its next tick.
+	d.reconcileOchreIndexes(d.ctx, registry, backend, ingest)
+
 	go d.runOchreIngestScheduler(ingest)
 	slog.Info("Ochre vector store enabled")
 }
 
+// reconcileOchreIndexes re-creates and re-ingests every registry index whose
+// backing Postgres table is missing -- the state left by a teardown that
+// preserved metadata (registry/KB/DataSource) but destroyed the RDS
+// instance, once Ensure has connected to a freshly-provisioned replacement.
+// An index whose table survived (a plain restart) is left alone: no
+// EnsureAccount/CreateIndex call and no ingest job for it. Best-effort: a
+// failure on one record is logged and the pass continues to the rest rather
+// than failing the appliance connect that already succeeded.
+func (d *Daemon) reconcileOchreIndexes(ctx context.Context, registry *handlers_ochrevector.Registry, backend handlers_ochrevector.VectorBackend, ingest *handlers_ochrevector.IngestService) {
+	recs, err := registry.ListAll(ctx)
+	if err != nil {
+		slog.Warn("Ochre vector store: reconcile indexes: list registry failed", "err", err)
+		return
+	}
+	for _, rec := range recs {
+		exists, err := backend.IndexExists(ctx, rec.AccountID, rec.ID)
+		if err != nil {
+			slog.Warn("Ochre vector store: reconcile indexes: check exists failed",
+				"account", rec.AccountID, "index", rec.ID, "err", err)
+			continue
+		}
+		if exists {
+			continue
+		}
+		if err := backend.EnsureAccount(ctx, rec.AccountID); err != nil {
+			slog.Warn("Ochre vector store: reconcile indexes: ensure account failed",
+				"account", rec.AccountID, "index", rec.ID, "err", err)
+			continue
+		}
+		spec := handlers_ochrevector.IndexSpec{ID: rec.ID, Dimension: rec.Dimension}
+		if err := backend.CreateIndex(ctx, rec.AccountID, spec); err != nil {
+			slog.Warn("Ochre vector store: reconcile indexes: create index failed",
+				"account", rec.AccountID, "index", rec.ID, "err", err)
+			continue
+		}
+		for _, source := range rec.SourceSpecs {
+			if _, err := ingest.ReconcileFromSource(ctx, rec.AccountID, rec.ID, source); err != nil {
+				slog.Warn("Ochre vector store: reconcile indexes: enqueue ingest failed",
+					"account", rec.AccountID, "index", rec.ID, "err", err)
+			}
+		}
+		slog.Info("Ochre vector store: reconcile indexes: re-created index absent from fresh appliance",
+			"account", rec.AccountID, "index", rec.ID, "sources", len(rec.SourceSpecs))
+	}
+}
+
 // ochreIngestSweepInterval paces the scheduler tick that drives PENDING ingest
 // jobs and re-drives stale RUNNING ones.
 const ochreIngestSweepInterval = 15 * time.Second
@@ -225,7 +277,7 @@ func (d *Daemon) runOchreIngestScheduler(ingest *handlers_ochrevector.IngestServ
 // an earlier call) has nothing to act on, which is reported as an error
 // rather than silently accepted -- an operator asking to tear down expects
 // one to have existed.
-func (d *Daemon) handleOchreApplianceTeardown(ctx context.Context, _ *handlers_ochrevector.TeardownApplianceRequest, _ string) (*handlers_ochrevector.TeardownApplianceResponse, error) {
+func (d *Daemon) handleOchreApplianceTeardown(ctx context.Context, req *handlers_ochrevector.TeardownApplianceRequest, _ string) (*handlers_ochrevector.TeardownApplianceResponse, error) {
 	d.mu.Lock()
 	appliance := d.ochreAppliance
 	d.mu.Unlock()
@@ -233,7 +285,8 @@ func (d *Daemon) handleOchreApplianceTeardown(ctx context.Context, _ *handlers_o
 		return nil, errors.New("ochrevector: platform appliance is not enabled or not up on this node")
 	}
 
-	if err := appliance.Teardown(ctx); err != nil {
+	purgeMetadata := req != nil && req.PurgeMetadata
+	if err := appliance.Teardown(ctx, purgeMetadata); err != nil {
 		return nil, fmt.Errorf("ochrevector: teardown platform appliance: %w", err)
 	}
 

spinifex/daemon/ochre_deps_test.go

@@ -7,15 +7,104 @@ package daemon
 import (
 	"context"
 	"errors"
+	"sync"
 	"testing"
 	"time"
 
 	"github.com/mulgadc/spinifex/spinifex/config"
 	handlers_ochrevector "github.com/mulgadc/spinifex/spinifex/handlers/ochrevector"
+	"github.com/mulgadc/spinifex/spinifex/testutil"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
 
+// reconcileFakeBackend is a minimal handlers_ochrevector.VectorBackend
+// double for reconcileOchreIndexes tests: only IndexExists/EnsureAccount/
+// CreateIndex are exercised, so every other method is an unused stub.
+type reconcileFakeBackend struct {
+	mu      sync.Mutex
+	indexes map[string]bool // key: accountID+"/"+indexID
+
+	ensureAccountCalls []string
+	createIndexCalls   []handlers_ochrevector.IndexSpec
+
+	// Error-injection fields for the reconcile error-branch tests below: each
+	// makes the matching backend method fail on every call, so a test that
+	// sets one exercises exactly the corresponding continue+slog.Warn branch.
+	indexExistsErr   error
+	ensureAccountErr error
+	createIndexErr   error
+
+	// onCreateIndex runs after a successful CreateIndex, letting a test mutate
+	// shared state (e.g. delete the registry record) without any production
+	// code change, so a later ReconcileFromSource call in the same pass fails.
+	onCreateIndex func(ctx context.Context, accountID string, spec handlers_ochrevector.IndexSpec)
+}
+
+var _ handlers_ochrevector.VectorBackend = (*reconcileFakeBackend)(nil)
+
+func newReconcileFakeBackend() *reconcileFakeBackend {
+	return &reconcileFakeBackend{indexes: map[string]bool{}}
+}
+
+func (f *reconcileFakeBackend) Init(_ context.Context) error { return nil }
+
+func (f *reconcileFakeBackend) EnsureAccount(_ context.Context, accountID string) error {
+	f.mu.Lock()
+	f.ensureAccountCalls = append(f.ensureAccountCalls, accountID)
+	err := f.ensureAccountErr
+	f.mu.Unlock()
+	return err
+}
+
+func (f *reconcileFakeBackend) CreateIndex(ctx context.Context, accountID string, spec handlers_ochrevector.IndexSpec) error {
+	f.mu.Lock()
+	f.createIndexCalls = append(f.createIndexCalls, spec)
+	err := f.createIndexErr
+	if err == nil {
+		f.indexes[accountID+"/"+spec.ID] = true
+	}
+	onCreateIndex := f.onCreateIndex
+	f.mu.Unlock()
+	if err != nil {
+		return err
+	}
+	if onCreateIndex != nil {
+		onCreateIndex(ctx, accountID, spec)
+	}
+	return nil
+}
+
+func (f *reconcileFakeBackend) IndexExists(_ context.Context, accountID, indexID string) (bool, error) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	if f.indexExistsErr != nil {
+		return false, f.indexExistsErr
+	}
+	return f.indexes[accountID+"/"+indexID], nil
+}
+
+func (f *reconcileFakeBackend) DropIndex(_ context.Context, accountID, indexID string) error {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	delete(f.indexes, accountID+"/"+indexID)
+	return nil
+}
+
+func (f *reconcileFakeBackend) ReplaceDocument(_ context.Context, _, _, _ string, _ []handlers_ochrevector.VectorRow) error {
+	return nil
+}
+
+func (f *reconcileFakeBackend) Query(_ context.Context, _, _ string, _ []float32, _ int, _ *handlers_ochrevector.Filter) ([]handlers_ochrevector.QueryResult, error) {
+	return nil, nil
+}
+
+func (f *reconcileFakeBackend) ensureAccountCallCount() int {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	return len(f.ensureAccountCalls)
+}
+
 // TestStartOchreVector_DisabledSkipsConstruction pins the config gate's
 // default-off behavior (D-series Stage 5b): with OchreVector.Enabled false
 // (the zero value, i.e. every daemon that has not opted in), startOchreVector
@@ -90,3 +179,175 @@ func TestHandleOchreApplianceTeardown_NilApplianceRefuses(t *testing.T) {
 	assert.Nil(t, out)
 	assert.Contains(t, err.Error(), "not enabled or not up")
 }
+
+const reconcileTestAccount = "111111111111"
+
+// TestReconcileOchreIndexes_RecreatesMissingIndexAndEnqueuesJob proves the
+// reconcile pass re-creates a registry index whose pg table is absent (the
+// state a fresh, teardown-and-rebuilt appliance leaves) and enqueues exactly
+// one PENDING ingest job per persisted SourceSpec.
+func TestReconcileOchreIndexes_RecreatesMissingIndexAndEnqueuesJob(t *testing.T) {
+	_, _, js := testutil.StartTestJetStream(t)
+	registry := handlers_ochrevector.NewRegistry(js)
+	jobs := handlers_ochrevector.NewJobStore(js)
+	backend := newReconcileFakeBackend()
+	ingest := handlers_ochrevector.NewIngestService(jobs, registry, backend, nil, nil)
+
+	ctx := context.Background()
+	now := time.Now().UTC()
+	source := handlers_ochrevector.SourceSpec{Bucket: "docs", Prefix: "kb/", ChunkSize: 100, ChunkOverlap: 10, EmbeddingModel: "stub-embed", Dimension: 2}
+	require.NoError(t, registry.Reserve(ctx, reconcileTestAccount, handlers_ochrevector.Record{
+		ID: "idx-one", Dimension: 2, State: handlers_ochrevector.StateReady,
+		SourceSpecs: []handlers_ochrevector.SourceSpec{source}, CreatedAt: now, UpdatedAt: now,
+	}))
+
+	d := &Daemon{}
+	d.reconcileOchreIndexes(ctx, registry, backend, ingest)
+
+	assert.Equal(t, 1, backend.ensureAccountCallCount(), "must EnsureAccount before CreateIndex")
+	require.Len(t, backend.createIndexCalls, 1)
+	assert.Equal(t, "idx-one", backend.createIndexCalls[0].ID)
+
+	jobRecs, err := jobs.ListAll(ctx)
+	require.NoError(t, err)
+	require.Len(t, jobRecs, 1, "must enqueue exactly one job for the one persisted SourceSpec")
+	assert.Equal(t, handlers_ochrevector.JobStatePending, jobRecs[0].State)
+	assert.Equal(t, source, jobRecs[0].Source)
+}
+
+// TestReconcileOchreIndexes_ExistingIndexIsNoOp proves an index whose pg
+// table already exists (a plain restart, not a teardown) is left alone: no
+// EnsureAccount/CreateIndex call and no ingest job enqueued for it.
+func TestReconcileOchreIndexes_ExistingIndexIsNoOp(t *testing.T) {
+	_, _, js := testutil.StartTestJetStream(t)
+	registry := handlers_ochrevector.NewRegistry(js)
+	jobs := handlers_ochrevector.NewJobStore(js)
+	backend := newReconcileFakeBackend()
+	ingest := handlers_ochrevector.NewIngestService(jobs, registry, backend, nil, nil)
+
+	ctx := context.Background()
+	now := time.Now().UTC()
+	source := handlers_ochrevector.SourceSpec{Bucket: "docs", Prefix: "kb/", Dimension: 2}
+	require.NoError(t, registry.Reserve(ctx, reconcileTestAccount, handlers_ochrevector.Record{
+		ID: "idx-one", Dimension: 2, State: handlers_ochrevector.StateReady,
+		SourceSpecs: []handlers_ochrevector.SourceSpec{source}, CreatedAt: now, UpdatedAt: now,
+	}))
+	// Pre-seed the backend as if the table already survived a restart.
+	require.NoError(t, backend.CreateIndex(ctx, reconcileTestAccount, handlers_ochrevector.IndexSpec{ID: "idx-one", Dimension: 2}))
+	backend.mu.Lock()
+	backend.createIndexCalls = nil // reset so the assertion below is reconcile's own calls, not the seed
+	backend.mu.Unlock()
+
+	d := &Daemon{}
+	d.reconcileOchreIndexes(ctx, registry, backend, ingest)
+
+	assert.Zero(t, backend.ensureAccountCallCount(), "an already-present index must not be re-provisioned")
+	assert.Empty(t, backend.createIndexCalls)
+
+	jobRecs, err := jobs.ListAll(ctx)
+	require.NoError(t, err)
+	assert.Empty(t, jobRecs, "an already-present index must not get a re-ingest job")
+}
+
+// TestReconcileOchreIndexes_ErrorsAreLoggedAndSkipped proves each best-effort
+// error branch (IndexExists, EnsureAccount, CreateIndex, ReconcileFromSource)
+// logs and skips its own record without panicking, rather than failing the
+// whole reconcile pass.
+func TestReconcileOchreIndexes_ErrorsAreLoggedAndSkipped(t *testing.T) {
+	newRecord := func(id string) handlers_ochrevector.Record {
+		now := time.Now().UTC()
+		return handlers_ochrevector.Record{
+			ID: id, Dimension: 2, State: handlers_ochrevector.StateReady,
+			SourceSpecs: []handlers_ochrevector.SourceSpec{{Bucket: "docs", Dimension: 2}},
+			CreatedAt:   now, UpdatedAt: now,
+		}
+	}
+
+	t.Run("IndexExists error skips the record", func(t *testing.T) {
+		_, _, js := testutil.StartTestJetStream(t)
+		registry := handlers_ochrevector.NewRegistry(js)
+		jobs := handlers_ochrevector.NewJobStore(js)
+		backend := newReconcileFakeBackend()
+		backend.indexExistsErr = errors.New("index exists boom")
+		ingest := handlers_ochrevector.NewIngestService(jobs, registry, backend, nil, nil)
+
+		ctx := context.Background()
+		require.NoError(t, registry.Reserve(ctx, reconcileTestAccount, newRecord("idx-exists-err")))
+
+		d := &Daemon{}
+		assert.NotPanics(t, func() { d.reconcileOchreIndexes(ctx, registry, backend, ingest) })
+
+		assert.Zero(t, backend.ensureAccountCallCount(), "IndexExists error must skip EnsureAccount")
+		assert.Empty(t, backend.createIndexCalls, "IndexExists error must skip CreateIndex")
+		jobRecs, err := jobs.ListAll(ctx)
+		require.NoError(t, err)
+		assert.Empty(t, jobRecs, "IndexExists error must skip the ingest enqueue")
+	})
+
+	t.Run("EnsureAccount error skips the record", func(t *testing.T) {
+		_, _, js := testutil.StartTestJetStream(t)
+		registry := handlers_ochrevector.NewRegistry(js)
+		jobs := handlers_ochrevector.NewJobStore(js)
+		backend := newReconcileFakeBackend()
+		backend.ensureAccountErr = errors.New("ensure account boom")
+		ingest := handlers_ochrevector.NewIngestService(jobs, registry, backend, nil, nil)
+
+		ctx := context.Background()
+		require.NoError(t, registry.Reserve(ctx, reconcileTestAccount, newRecord("idx-ensure-err")))
+
+		d := &Daemon{}
+		assert.NotPanics(t, func() { d.reconcileOchreIndexes(ctx, registry, backend, ingest) })
+
+		assert.Equal(t, 1, backend.ensureAccountCallCount(), "EnsureAccount must still be attempted")
+		assert.Empty(t, backend.createIndexCalls, "EnsureAccount error must skip CreateIndex")
+		jobRecs, err := jobs.ListAll(ctx)
+		require.NoError(t, err)
+		assert.Empty(t, jobRecs, "EnsureAccount error must skip the ingest enqueue")
+	})
+
+	t.Run("CreateIndex error skips the record", func(t *testing.T) {
+		_, _, js := testutil.StartTestJetStream(t)
+		registry := handlers_ochrevector.NewRegistry(js)
+		jobs := handlers_ochrevector.NewJobStore(js)
+		backend := newReconcileFakeBackend()
+		backend.createIndexErr = errors.New("create index boom")
+		ingest := handlers_ochrevector.NewIngestService(jobs, registry, backend, nil, nil)
+
+		ctx := context.Background()
+		require.NoError(t, registry.Reserve(ctx, reconcileTestAccount, newRecord("idx-create-err")))
+
+		d := &Daemon{}
+		assert.NotPanics(t, func() { d.reconcileOchreIndexes(ctx, registry, backend, ingest) })
+
+		require.Len(t, backend.createIndexCalls, 1, "CreateIndex must still be attempted")
+		jobRecs, err := jobs.ListAll(ctx)
+		require.NoError(t, err)
+		assert.Empty(t, jobRecs, "CreateIndex error must skip the ingest enqueue")
+	})
+
+	t.Run("ReconcileFromSource error is logged and does not panic", func(t *testing.T) {
+		_, _, js := testutil.StartTestJetStream(t)
+		registry := handlers_ochrevector.NewRegistry(js)
+		jobs := handlers_ochrevector.NewJobStore(js)
+		backend := newReconcileFakeBackend()
+		ctx := context.Background()
+		// Delete the just-created record as a side effect of CreateIndex
+		// succeeding, so the ReconcileFromSource call right after it misses
+		// the registry and returns ErrIndexNotFound -- no production code
+		// change needed to force that failure.
+		backend.onCreateIndex = func(ctx context.Context, accountID string, spec handlers_ochrevector.IndexSpec) {
+			require.NoError(t, registry.Delete(ctx, accountID, spec.ID))
+		}
+		ingest := handlers_ochrevector.NewIngestService(jobs, registry, backend, nil, nil)
+
+		require.NoError(t, registry.Reserve(ctx, reconcileTestAccount, newRecord("idx-ingest-err")))
+
+		d := &Daemon{}
+		assert.NotPanics(t, func() { d.reconcileOchreIndexes(ctx, registry, backend, ingest) })
+
+		require.Len(t, backend.createIndexCalls, 1, "CreateIndex must still run before the ingest enqueue fails")
+		jobRecs, err := jobs.ListAll(ctx)
+		require.NoError(t, err)
+		assert.Empty(t, jobRecs, "ReconcileFromSource error must not leave a job behind")
+	})
+}

spinifex/handlers/ochrevector/appliance.go

@@ -92,9 +92,14 @@ type ApplianceRecord struct {
 	UpdatedAt         time.Time `json:"updatedAt"`
 }
 
-// TeardownApplianceRequest has no fields: the appliance is a fixed,
-// deployment-wide singleton (D2), never identified by a caller-supplied ID.
-type TeardownApplianceRequest struct{}
+// TeardownApplianceRequest identifies no resource -- the appliance is a
+// fixed, deployment-wide singleton (D2). PurgeMetadata opts into the old
+// full wipe (index registry, KB and DataSource records alongside the RDS
+// instance); the default leaves that metadata intact so a later Ensure can
+// reconcile and re-ingest from it without operator input.
+type TeardownApplianceRequest struct {
+	PurgeMetadata bool `json:"purgeMetadata,omitempty"`
+}
 
 type TeardownApplianceResponse struct{}
 
@@ -215,14 +220,17 @@ func (a *Appliance) TeardownHostPort() error {
 }
 
 // Teardown removes the singleton platform appliance -- RDS instance, host
-// port, then (if WithStores/WithKBStores were called) index/job and
-// KB/DataSource metadata, then the KV record -- so a rebuilt appliance starts
-// coherent. Every step runs
-// regardless of an earlier one's failure, and every failure is joined into
-// the returned error. Idempotent overall: tearing down an already-absent
-// appliance is a no-op success. Does not re-provision -- the daemon's own
-// Ensure does that on next startup, not this call.
-func (a *Appliance) Teardown(ctx context.Context) error {
+// port, then the appliance KV record -- so a rebuilt appliance starts
+// coherent. Index/job registry and KB/DataSource metadata (if
+// WithStores/WithKBStores were called) survive by default exactly as they
+// survive a restart, so a later Ensure can reconcile and re-ingest from them
+// without operator input; purgeMetadata=true restores the old full wipe of
+// all of it for an intentional destroy. Every step runs regardless of an
+// earlier one's failure, and every failure is joined into the returned
+// error. Idempotent overall: tearing down an already-absent appliance is a
+// no-op success. Does not re-provision -- the daemon's own Ensure does that
+// on next startup, not this call.
+func (a *Appliance) Teardown(ctx context.Context, purgeMetadata bool) error {
 	var errs []error
 
 	if err := a.launcher.Delete(ctx, ApplianceIdentifier); err != nil {
@@ -237,30 +245,34 @@ func (a *Appliance) Teardown(ctx context.Context) error {
 	a.mu.Lock()
 	registry, jobs, kb, ds := a.registry, a.jobs, a.kb, a.ds
 	a.mu.Unlock()
-	// Purged after the data-bearing RDS instance is gone but before the
-	// appliance record: a crash here still leaves a record an operator can
-	// retry teardown against, re-attempting the (idempotent) purge too.
-	if registry != nil {
-		if err := registry.PurgeAll(ctx); err != nil {
-			errs = append(errs, fmt.Errorf("ochrevector: purge index registry: %w", err))
-		}
-	}
+	// Job history is scoped to the now-destroyed RDS instance, so it is
+	// always cleared. Registry/kb/ds instead carry the identity and
+	// SourceSpecs a fresh Ensure needs to reconcile and re-ingest without
+	// operator input, so they purge only when purgeMetadata asks for the old
+	// full wipe.
 	if jobs != nil {
 		if err := jobs.PurgeAll(ctx); err != nil {
 			errs = append(errs, fmt.Errorf("ochrevector: purge ingestion jobs: %w", err))
 		}
 	}
-	// kb/ds are gateway-owned metadata, not daemon-owned like registry/jobs
-	// above, but they still point at this appliance's index/backend, so a
-	// torn-down appliance must not leave them dangling either.
-	if kb != nil {
-		if err := kb.PurgeAll(ctx); err != nil {
-			errs = append(errs, fmt.Errorf("ochrevector: purge knowledge bases: %w", err))
+	if purgeMetadata {
+		if registry != nil {
+			if err := registry.PurgeAll(ctx); err != nil {
+				errs = append(errs, fmt.Errorf("ochrevector: purge index registry: %w", err))
+			}
 		}
-	}
-	if ds != nil {
-		if err := ds.PurgeAll(ctx); err != nil {
-			errs = append(errs, fmt.Errorf("ochrevector: purge data sources: %w", err))
+		// kb/ds are gateway-owned metadata, not daemon-owned like
+		// registry/jobs above, but they still point at this appliance's
+		// index/backend, so a full wipe must not leave them dangling either.
+		if kb != nil {
+			if err := kb.PurgeAll(ctx); err != nil {
+				errs = append(errs, fmt.Errorf("ochrevector: purge knowledge bases: %w", err))
+			}
+		}
+		if ds != nil {
+			if err := ds.PurgeAll(ctx); err != nil {
+				errs = append(errs, fmt.Errorf("ochrevector: purge data sources: %w", err))
+			}
 		}
 	}
 

spinifex/handlers/ochrevector/appliance_test.go

@@ -519,7 +519,7 @@ func TestTeardown_DeletesRecordLauncherAndHostPort(t *testing.T) {
 	appliance.eniID = "eni-installed"
 	appliance.mu.Unlock()
 
-	require.NoError(t, appliance.Teardown(context.Background()))
+	require.NoError(t, appliance.Teardown(context.Background(), false))
 
 	assert.Equal(t, []string{ApplianceIdentifier}, launcher.deleteCalls)
 	assert.Equal(t, []string{"eni-installed"}, h.hostPort.removals())
@@ -540,11 +540,11 @@ func TestTeardown_NoExistingApplianceIsANoOp(t *testing.T) {
 	appliance, err := NewAppliance(js, testMasterKey(t), launcher)
 	require.NoError(t, err)
 
-	assert.NoError(t, appliance.Teardown(context.Background()))
+	assert.NoError(t, appliance.Teardown(context.Background(), false))
 	assert.Equal(t, 1, launcher.deleteCallCount())
 
 	// Tearing down again must still be a no-op success.
-	assert.NoError(t, appliance.Teardown(context.Background()))
+	assert.NoError(t, appliance.Teardown(context.Background(), false))
 }
 
 // TestTeardown_StillDeletesRecordWhenHostPortRemovalFails proves a host-port
@@ -565,7 +565,7 @@ func TestTeardown_StillDeletesRecordWhenHostPortRemovalFails(t *testing.T) {
 	appliance.eniID = "eni-installed"
 	appliance.mu.Unlock()
 
-	err = appliance.Teardown(context.Background())
+	err = appliance.Teardown(context.Background(), false)
 	require.Error(t, err)
 	assert.Contains(t, err.Error(), "ovs-vsctl exploded")
 
@@ -587,7 +587,7 @@ func TestTeardown_JoinsLauncherAndRecordErrors(t *testing.T) {
 	require.NoError(t, err)
 	seedAvailableAppliance(t, appliance, masterKey, "10.0.0.32", 5432)
 
-	err = appliance.Teardown(context.Background())
+	err = appliance.Teardown(context.Background(), false)
 	require.Error(t, err)
 	assert.Contains(t, err.Error(), "rds unreachable")
 
@@ -598,11 +598,11 @@ func TestTeardown_JoinsLauncherAndRecordErrors(t *testing.T) {
 	assert.Nil(t, rec, "the KV record must still be deleted despite the launcher failure")
 }
 
-// TestTeardown_WithStoresPurgesRegistryAndJobs proves an Appliance wired via
-// WithStores purges every index and job record on Teardown, so a
-// reprovisioned appliance never advertises metadata pointing at data that no
-// longer exists.
-func TestTeardown_WithStoresPurgesRegistryAndJobs(t *testing.T) {
+// TestTeardown_DefaultPreservesRegistryButPurgesJobs proves a default
+// Teardown (purgeMetadata=false) leaves the index registry intact -- so a
+// later Ensure can reconcile from it -- while still clearing job history,
+// which is scoped to the now-destroyed RDS instance.
+func TestTeardown_DefaultPreservesRegistryButPurgesJobs(t *testing.T) {
 	_, _, js := testutil.StartTestJetStream(t)
 	masterKey := testMasterKey(t)
 	launcher := &fakeLauncher{endpoint: "10.0.0.33", port: 5432}
@@ -618,22 +618,51 @@ func TestTeardown_WithStoresPurgesRegistryAndJobs(t *testing.T) {
 	require.NoError(t, jobs.Reserve(ctx, "111111111111", JobRecord{ID: "job-one", CreatedAt: now, UpdatedAt: now}))
 
 	appliance.WithStores(registry, jobs)
-	require.NoError(t, appliance.Teardown(ctx))
+	require.NoError(t, appliance.Teardown(ctx, false))
 
 	remainingIndexes, err := registry.ListAll(ctx)
 	require.NoError(t, err)
-	assert.Empty(t, remainingIndexes, "Teardown must purge every index record")
+	assert.Len(t, remainingIndexes, 1, "default Teardown must leave the index registry intact")
 
 	remainingJobs, err := jobs.ListAll(ctx)
 	require.NoError(t, err)
-	assert.Empty(t, remainingJobs, "Teardown must purge every job record")
+	assert.Empty(t, remainingJobs, "Teardown must still purge job history")
 }
 
-// TestTeardown_WithKBStoresPurgesKnowledgeBasesAndDataSources proves an
-// Appliance wired via WithKBStores purges every gateway-owned knowledge-base
-// and data-source record on Teardown, so a reprovisioned appliance never
-// leaves KB/DataSource metadata pointing at an index that no longer exists.
-func TestTeardown_WithKBStoresPurgesKnowledgeBasesAndDataSources(t *testing.T) {
+// TestTeardown_PurgeMetadataPurgesRegistryAndJobs proves purgeMetadata=true
+// restores the old full wipe of the index registry (and job history)
+// alongside the RDS instance, for an intentional full destroy.
+func TestTeardown_PurgeMetadataPurgesRegistryAndJobs(t *testing.T) {
+	_, _, js := testutil.StartTestJetStream(t)
+	masterKey := testMasterKey(t)
+	launcher := &fakeLauncher{endpoint: "10.0.0.36", port: 5432}
+	appliance, err := NewAppliance(js, masterKey, launcher)
+	require.NoError(t, err)
+	seedAvailableAppliance(t, appliance, masterKey, "10.0.0.36", 5432)
+
+	registry := NewRegistry(js)
+	jobs := NewJobStore(js)
+	ctx := context.Background()
+	now := time.Now().UTC()
+	require.NoError(t, registry.Reserve(ctx, "111111111111", Record{ID: "idx-one", CreatedAt: now, UpdatedAt: now}))
+	require.NoError(t, jobs.Reserve(ctx, "111111111111", JobRecord{ID: "job-one", CreatedAt: now, UpdatedAt: now}))
+
+	appliance.WithStores(registry, jobs)
+	require.NoError(t, appliance.Teardown(ctx, true))
+
+	remainingIndexes, err := registry.ListAll(ctx)
+	require.NoError(t, err)
+	assert.Empty(t, remainingIndexes, "purgeMetadata=true must purge every index record")
+
+	remainingJobs, err := jobs.ListAll(ctx)
+	require.NoError(t, err)
+	assert.Empty(t, remainingJobs, "purgeMetadata=true must still purge every job record")
+}
+
+// TestTeardown_DefaultPreservesKnowledgeBasesAndDataSources proves a default
+// Teardown leaves gateway-owned KB/DataSource records intact, mirroring the
+// registry: their SourceSpec is what a later reconcile re-ingests from.
+func TestTeardown_DefaultPreservesKnowledgeBasesAndDataSources(t *testing.T) {
 	_, _, js := testutil.StartTestJetStream(t)
 	masterKey := testMasterKey(t)
 	launcher := &fakeLauncher{endpoint: "10.0.0.35", port: 5432}
@@ -649,15 +678,80 @@ func TestTeardown_WithKBStoresPurgesKnowledgeBasesAndDataSources(t *testing.T) {
 	require.NoError(t, ds.Create(ctx, "111111111111", DataSourceRecord{ID: "ds-one", KnowledgeBaseID: "kb-one", CreatedAt: now, UpdatedAt: now}))
 
 	appliance.WithKBStores(kb, ds)
-	require.NoError(t, appliance.Teardown(ctx))
+	require.NoError(t, appliance.Teardown(ctx, false))
 
 	remainingKBs, err := kb.List(ctx, "111111111111")
 	require.NoError(t, err)
-	assert.Empty(t, remainingKBs, "Teardown must purge every knowledge base record")
+	assert.Len(t, remainingKBs, 1, "default Teardown must leave knowledge base records intact")
 
 	remainingDS, err := ds.List(ctx, "111111111111")
 	require.NoError(t, err)
-	assert.Empty(t, remainingDS, "Teardown must purge every data source record")
+	assert.Len(t, remainingDS, 1, "default Teardown must leave data source records intact")
+}
+
+// TestTeardown_PurgeMetadataPurgesKnowledgeBasesAndDataSources proves
+// purgeMetadata=true restores the old full wipe of the gateway-owned
+// knowledge-base and data-source records.
+func TestTeardown_PurgeMetadataPurgesKnowledgeBasesAndDataSources(t *testing.T) {
+	_, _, js := testutil.StartTestJetStream(t)
+	masterKey := testMasterKey(t)
+	launcher := &fakeLauncher{endpoint: "10.0.0.37", port: 5432}
+	appliance, err := NewAppliance(js, masterKey, launcher)
+	require.NoError(t, err)
+	seedAvailableAppliance(t, appliance, masterKey, "10.0.0.37", 5432)
+
+	kb := NewKBStore(js)
+	ds := NewDataSourceStore(js)
+	ctx := context.Background()
+	now := time.Now().UTC()
+	require.NoError(t, kb.Create(ctx, "111111111111", KBRecord{ID: "kb-one", IndexID: "idx-one", CreatedAt: now, UpdatedAt: now}))
+	require.NoError(t, ds.Create(ctx, "111111111111", DataSourceRecord{ID: "ds-one", KnowledgeBaseID: "kb-one", CreatedAt: now, UpdatedAt: now}))
+
+	appliance.WithKBStores(kb, ds)
+	require.NoError(t, appliance.Teardown(ctx, true))
+
+	remainingKBs, err := kb.List(ctx, "111111111111")
+	require.NoError(t, err)
+	assert.Empty(t, remainingKBs, "purgeMetadata=true must purge every knowledge base record")
+
+	remainingDS, err := ds.List(ctx, "111111111111")
+	require.NoError(t, err)
+	assert.Empty(t, remainingDS, "purgeMetadata=true must purge every data source record")
+}
+
+// TestTeardown_PurgeMetadataSurfacesStorePurgeErrors proves a purgeMetadata=
+// true Teardown reports (rather than swallows) a failure from any of
+// registry/kb/ds PurgeAll, joined alongside every other step's own errors.
+func TestTeardown_PurgeMetadataSurfacesStorePurgeErrors(t *testing.T) {
+	_, nc, js := testutil.StartTestJetStream(t)
+	masterKey := testMasterKey(t)
+	launcher := &fakeLauncher{endpoint: "10.0.0.39", port: 5432}
+	appliance, err := NewAppliance(js, masterKey, launcher)
+	require.NoError(t, err)
+	seedAvailableAppliance(t, appliance, masterKey, "10.0.0.39", 5432)
+
+	registry := NewRegistry(js)
+	jobs := NewJobStore(js)
+	kb := NewKBStore(js)
+	ds := NewDataSourceStore(js)
+	ctx := context.Background()
+	now := time.Now().UTC()
+	require.NoError(t, registry.Reserve(ctx, "111111111111", Record{ID: "idx-one", CreatedAt: now, UpdatedAt: now}))
+	require.NoError(t, kb.Create(ctx, "111111111111", KBRecord{ID: "kb-one", IndexID: "idx-one", CreatedAt: now, UpdatedAt: now}))
+	require.NoError(t, ds.Create(ctx, "111111111111", DataSourceRecord{ID: "ds-one", KnowledgeBaseID: "kb-one", CreatedAt: now, UpdatedAt: now}))
+
+	appliance.WithStores(registry, jobs)
+	appliance.WithKBStores(kb, ds)
+
+	// Sever the shared NATS connection so every store's PurgeAll call fails,
+	// without any production code change.
+	nc.Close()
+
+	err = appliance.Teardown(ctx, true)
+	require.Error(t, err)
+	assert.Contains(t, err.Error(), "purge index registry")
+	assert.Contains(t, err.Error(), "purge knowledge bases")
+	assert.Contains(t, err.Error(), "purge data sources")
 }
 
 // TestTeardown_WithoutStoresStillWorks proves an Appliance that never calls
@@ -671,7 +765,7 @@ func TestTeardown_WithoutStoresStillWorks(t *testing.T) {
 	require.NoError(t, err)
 	seedAvailableAppliance(t, appliance, masterKey, "10.0.0.34", 5432)
 
-	require.NoError(t, appliance.Teardown(context.Background()))
+	require.NoError(t, appliance.Teardown(context.Background(), false))
 }
 
 // TestBucket_RequiresJetStream proves bucket fails fast on a zero-value

spinifex/handlers/ochrevector/backend.go

@@ -71,6 +71,12 @@ type VectorBackend interface {
 	// accountID's schema. Idempotent: safe to retry after a crash mid-create.
 	CreateIndex(ctx context.Context, accountID string, spec IndexSpec) error
 
+	// IndexExists reports whether indexID's backing table already exists
+	// under accountID's schema, so a caller can tell a survived-restart index
+	// from one whose Postgres was rebuilt out from under it without
+	// re-creating (and re-ingesting) one that is already there.
+	IndexExists(ctx context.Context, accountID, indexID string) (bool, error)
+
 	// DropIndex drops indexID's backing table under accountID's schema.
 	// Idempotent: dropping an already-absent index is a no-op success.
 	DropIndex(ctx context.Context, accountID, indexID string) error

spinifex/handlers/ochrevector/fake_backend_test.go

@@ -65,6 +65,13 @@ func (f *fakeBackend) CreateIndex(_ context.Context, accountID string, spec Inde
 	return nil
 }
 
+// IndexExists mirrors pgxBackend.IndexExists against the same indexes map
+// CreateIndex/DropIndex mutate, so it reflects exactly what a real backend
+// would report without a database.
+func (f *fakeBackend) IndexExists(_ context.Context, accountID, indexID string) (bool, error) {
+	return f.hasIndex(accountID, indexID), nil
+}
+
 func (f *fakeBackend) DropIndex(_ context.Context, accountID, indexID string) error {
 	f.mu.Lock()
 	defer f.mu.Unlock()

spinifex/handlers/ochrevector/ingest.go

@@ -107,6 +107,15 @@ func (s *IngestService) StartIngest(ctx context.Context, accountID, indexID stri
 	return &rec, nil
 }
 
+// ReconcileFromSource enqueues a PENDING ingest job for accountID/indexID
+// from a persisted SourceSpec -- the entry point a startup reconcile pass
+// calls to re-populate an index whose backing table survived only in the
+// registry, not the freshly-provisioned Postgres. StartIngest's own
+// reservation makes a repeat call safe: Sweep/RunJob never double-write.
+func (s *IngestService) ReconcileFromSource(ctx context.Context, accountID, indexID string, source SourceSpec) (*JobRecord, error) {
+	return s.StartIngest(ctx, accountID, indexID, source, "")
+}
+
 // RunJob is the synchronous ingestion worker: lists job.Source, and for each
 // object, reads/chunks/embeds/replaces its rows. A per-document failure
 // (unreadable object, oversize, exhausted embed retries) is skipped and

spinifex/handlers/ochrevector/ingest_test.go

@@ -218,6 +218,34 @@ func TestStartIngest_StampsDataSourceID(t *testing.T) {
 	assert.Empty(t, noDSJob.DataSourceID)
 }
 
+// TestReconcileFromSource_EnqueuesPendingJobWithNoDataSourceID proves the
+// startup-reconcile entry point enqueues a PENDING job from a bare
+// SourceSpec, tagged with no DataSourceID -- it re-populates the index's
+// table directly from the registry, not through a bedrock-agent DataSource.
+func TestReconcileFromSource_EnqueuesPendingJobWithNoDataSourceID(t *testing.T) {
+	svc, _, _, _, _ := newIngestTestSetup(t)
+	ctx := context.Background()
+
+	job, err := svc.ReconcileFromSource(ctx, ingestAccountA, "idx-one", testSource())
+	require.NoError(t, err)
+	require.NotNil(t, job)
+	assert.Equal(t, JobStatePending, job.State)
+	assert.Empty(t, job.DataSourceID)
+	assert.Equal(t, testSource(), job.Source)
+}
+
+// TestReconcileFromSource_MissingIndexErrors proves the entry point refuses
+// the same way StartIngest does for an index the registry does not have --
+// a reconcile pass must never enqueue a job against nothing.
+func TestReconcileFromSource_MissingIndexErrors(t *testing.T) {
+	svc, _, _, _, _ := newIngestTestSetup(t)
+	ctx := context.Background()
+
+	_, err := svc.ReconcileFromSource(ctx, ingestAccountA, "idx-does-not-exist", testSource())
+	require.Error(t, err)
+	assert.ErrorIs(t, err, ErrIndexNotFound)
+}
+
 func TestRunJob_HappyPath(t *testing.T) {
 	svc, registry, backend, store, embedder := newIngestTestSetup(t)
 	ctx := context.Background()

spinifex/handlers/ochrevector/pgx_backend.go

@@ -223,6 +223,25 @@ func (b *pgxBackend) CreateIndex(ctx context.Context, accountID string, spec Ind
 	})
 }
 
+// IndexExists reports whether indexID's backing table exists under
+// accountID's schema. to_regclass resolves to NULL for a schema or table
+// that is not there rather than erroring, so no account role or search_path
+// setup is needed for this read-only lookup.
+func (b *pgxBackend) IndexExists(ctx context.Context, accountID, indexID string) (bool, error) {
+	if err := validateAccountID(accountID); err != nil {
+		return false, err
+	}
+	if err := validateIndexID(indexID); err != nil {
+		return false, err
+	}
+	qualified := schemaName(accountID) + "." + tableName(indexID)
+	var regclass *string
+	if err := b.pool.QueryRow(ctx, `SELECT to_regclass($1)::text`, qualified).Scan(&regclass); err != nil {
+		return false, fmt.Errorf("ochrevector: check index %s exists for account %s: %w", indexID, accountID, err)
+	}
+	return regclass != nil, nil
+}
+
 // DropIndex drops indexID's backing table under accountID's schema. Idempotent:
 // dropping an already-absent index is a no-op success.
 func (b *pgxBackend) DropIndex(ctx context.Context, accountID, indexID string) error {

spinifex/handlers/ochrevector/pgx_backend_test.go

@@ -47,8 +47,16 @@ func TestPgxBackend_Integration(t *testing.T) {
 		_ = backend.DropIndex(context.Background(), pgTestAccountA, indexID)
 	})
 
+	exists, err := backend.IndexExists(ctx, pgTestAccountA, indexID)
+	require.NoError(t, err)
+	assert.False(t, exists, "index must not exist before CreateIndex")
+
 	require.NoError(t, backend.CreateIndex(ctx, pgTestAccountA, IndexSpec{ID: indexID, Dimension: 8}))
 
+	exists, err = backend.IndexExists(ctx, pgTestAccountA, indexID)
+	require.NoError(t, err)
+	assert.True(t, exists, "index must exist after CreateIndex")
+
 	// Re-running CreateIndex must not error (IF NOT EXISTS), the property
 	// Reconcile's forward-retry depends on.
 	require.NoError(t, backend.CreateIndex(ctx, pgTestAccountA, IndexSpec{ID: indexID, Dimension: 8}))
@@ -69,6 +77,10 @@ func TestPgxBackend_Integration(t *testing.T) {
 
 	require.NoError(t, backend.DropIndex(ctx, pgTestAccountA, indexID))
 
+	exists, err = backend.IndexExists(ctx, pgTestAccountA, indexID)
+	require.NoError(t, err)
+	assert.False(t, exists, "index must not exist after DropIndex")
+
 	// Dropping an already-absent index is a no-op success.
 	require.NoError(t, backend.DropIndex(ctx, pgTestAccountA, indexID))
 }