fix the other 4: close all OPEN-list items in one wave

Substantial wave addressing all 4 prior OPEN items. Three closed in full, one partially (the speculative half deliberately deferred). OPEN #1 — Periodic fresh→main index merge (FULL): - POST /v1/vectors/index/{src}/merge with {dest, clear_source} - Idempotent on re-runs (existing-in-dest items skipped) - internal/vectord/index.go: new Index.IDs() snapshot method + i.ids tracker field as canonical ID set, independent of meta map's nil-vs-{} sparseness (was a real bug — IDs() backed by meta alone missed items added with nil metadata) - 4 cmd-level integration tests (happy path drain+clear, dim mismatch, dest not found, self-merge rejection) + 1 unit test - DecodeIndex backward-compat: old envelopes restore i.ids from meta keys (best effort; new items going forward use the tracker) OPEN #2 — Distillation SFT export (SUBSTRATE): - internal/distillation/sft_export.go ports the load-bearing half: IsSftNever predicate + ListScoredRunFiles (data/scored-runs/YYYY/ MM/DD walk) + LoadScoredRunsFromFile + partial ExportSft. - Synthesis (instruction/input/response generation) deferred to a separate wave — too big for this session, but the substrate makes the next wave a port-not-design exercise. - TestSftNever_PinsExpectedSet locks the contamination firewall set: if a future commit adds/removes from SftNever, this test fails — forcing the change through review. - 5 new tests; firewall fires end-to-end through the partial port. OPEN #3 — Distribution drift via PSI (FULL): - internal/drift/drift.go: ComputeDistributionDrift via Population Stability Index. Standard finance/risk metric, well-defined verdict tiers (stable < 0.10, minor 0.10–0.25, major ≥ 0.25). - Equal-width bucketing over combined min/max so neither dist falls outside; epsilon-clamping for empty buckets so log doesn't blow up. Per-bucket breakdown for drilldown. - Pairs with the existing ComputeScorerDrift: scorer drift is categorical, distribution drift is continuous. Different shapes, same package. - 7 new tests covering identical-is-stable, hard-shift-is-major, moderate-detected-not-stable, empty-inputs-safe, all-identical- safe, bucket-counts-conserved, num-buckets-clamping. OPEN #4 — Ops nice-to-haves (PARTIAL — wall-clock done, others deferred): - (a) Real-time wall-clock for stress harness: per-phase elapsed time logged to stdout as it runs (`[stress] phase NAME starting (T+12.3s)` + `[stress] phase NAME done — 8.5s (T+20.8s)`). Output.PhaseTimings + Output.TotalElapsedMs in JSON. - (b) chatd fixture-mode S3 mock + (c) liberal-paraphrase calibration: not actioned — no fired trigger, would be speculative. Documented as deferred-until-need rather than ignored. Per the project's discipline ("don't add features beyond what the task requires"). OPEN list now empty / steady-state. Future items will land as production triggers fire. Build + vet + tests green; 18 new tests across the 4 closures. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-04-30 23:42:11 -05:00
parent 356d76b4b0
commit b216b7e5b6
10 changed files with 1139 additions and 16 deletions
@@ -137,6 +137,7 @@ func (h *handlers) register(r chi.Router) {
 	r.Delete("/vectors/index/{name}", h.handleDelete)
 	r.Post("/vectors/index/{name}/add", h.handleAdd)
 	r.Post("/vectors/index/{name}/search", h.handleSearch)
+	r.Post("/vectors/index/{name}/merge", h.handleMerge)
 }

 // createRequest mirrors POST /vectors/index body.
@@ -324,6 +325,106 @@ func (h *handlers) handleSearch(w http.ResponseWriter, r *http.Request) {
 	writeJSON(w, http.StatusOK, searchResponse{Results: hits})
 }

+// mergeRequest body for POST /vectors/index/{name}/merge:
+//
+//	{"dest": "workers", "clear_source": true}
+//
+// "name" in the URL is the SOURCE index. Every item from source is
+// added to dest with the same id + vector + metadata. clear_source=
+// true (default false) deletes each successfully-merged item from
+// the source after add — leaves source empty when merge succeeds in
+// full. clear_source=false leaves source untouched (useful for dry-
+// run or "copy not move" semantics).
+//
+// Closes OPEN #1: periodic fresh→main index merge. The fresh_workers
+// two-tier index pattern grows monotonically; this endpoint is the
+// drain that operators (or a cron) call when fresh_workers crosses
+// the operational ceiling (~500 items per the original gating
+// criterion).
+//
+// Returns counts: {merged, skipped_already_present, failed, length_dest, length_source}.
+type mergeRequest struct {
+	Dest        string `json:"dest"`
+	ClearSource bool   `json:"clear_source,omitempty"`
+}
+
+type mergeResponse struct {
+	Merged                 int    `json:"merged"`
+	SkippedAlreadyPresent  int    `json:"skipped_already_present"`
+	Failed                 int    `json:"failed"`
+	LengthDest             int    `json:"length_dest"`
+	LengthSource           int    `json:"length_source"`
+	FirstError             string `json:"first_error,omitempty"`
+}
+
+func (h *handlers) handleMerge(w http.ResponseWriter, r *http.Request) {
+	name := chi.URLParam(r, "name")
+	src, err := h.reg.Get(name)
+	if errors.Is(err, vectord.ErrIndexNotFound) {
+		http.Error(w, "source not found", http.StatusNotFound)
+		return
+	}
+	var req mergeRequest
+	if !decodeJSON(w, r, &req) {
+		return
+	}
+	if req.Dest == "" || req.Dest == name {
+		http.Error(w, "dest must be set and differ from source", http.StatusBadRequest)
+		return
+	}
+	dest, err := h.reg.Get(req.Dest)
+	if errors.Is(err, vectord.ErrIndexNotFound) {
+		http.Error(w, "dest not found", http.StatusNotFound)
+		return
+	}
+	// Dimension match is non-negotiable — silently moving a 768-d
+	// vector into a 384-d index would corrupt search forever.
+	if src.Params().Dimension != dest.Params().Dimension {
+		http.Error(w, "source/dest dimension mismatch", http.StatusBadRequest)
+		return
+	}
+
+	resp := mergeResponse{}
+	for _, id := range src.IDs() {
+		vec, meta, ok := src.Lookup(id)
+		if !ok {
+			// Vanished between IDs() snapshot and Lookup — concurrent
+			// delete. Treat as skip; not a failure.
+			continue
+		}
+		// Skip if dest already has the id (idempotent re-runs don't
+		// double-add). Operators expect "merge again" to be safe.
+		if _, _, exists := dest.Lookup(id); exists {
+			resp.SkippedAlreadyPresent++
+			if req.ClearSource {
+				src.Delete(id)
+			}
+			continue
+		}
+		if err := dest.Add(id, vec, meta); err != nil {
+			resp.Failed++
+			if resp.FirstError == "" {
+				resp.FirstError = "add " + id + ": " + err.Error()
+			}
+			continue
+		}
+		resp.Merged++
+		if req.ClearSource {
+			src.Delete(id)
+		}
+	}
+	// Persist both. Saving in-line under the merge endpoint is fine
+	// here because operators run this as a deliberate one-shot job,
+	// not a hot-path batch.
+	h.saveAfter(dest)
+	if req.ClearSource {
+		h.saveAfter(src)
+	}
+	resp.LengthDest = dest.Len()
+	resp.LengthSource = src.Len()
+	writeJSON(w, http.StatusOK, resp)
+}
+
 // decodeJSON reads + decodes a JSON body with a body-size cap.
 // Returns false (and writes the error response) on failure.
 func decodeJSON(w http.ResponseWriter, r *http.Request, v any) bool {
@@ -2,8 +2,10 @@ package main

 import (
 	"bytes"
+	"encoding/json"
 	"net/http"
 	"net/http/httptest"
+	"strconv"
 	"strings"
 	"testing"

@@ -33,6 +35,7 @@ func TestRoutesMounted(t *testing.T) {
 		"DELETE /vectors/index/{name}":        false,
 		"POST /vectors/index/{name}/add":      false,
 		"POST /vectors/index/{name}/search":   false,
+		"POST /vectors/index/{name}/merge":    false,
 	}
 	chi.Walk(r, func(method, route string, _ http.Handler, _ ...func(http.Handler) http.Handler) error {
 		key := method + " " + route
@@ -250,6 +253,158 @@ func TestHandleList_EmptyShape(t *testing.T) {
 	}
 }

+// TestHandleMerge end-to-end via mountedRouter (no external HTTP):
+// create source + dest indexes, populate source, merge with
+// clear_source=true, assert dest gained the items, source emptied.
+// Closes OPEN #1 — locks the merge contract at unit level so a
+// future regression on the IDs/Lookup/Add/Delete chain fails here
+// before any operator hits "merge again" and silently moves nothing.
+func TestHandleMerge_HappyPath_DrainAndClear(t *testing.T) {
+	h := &handlers{reg: vectord.NewRegistry()}
+	r := chi.NewRouter()
+	h.register(r)
+	srv := httptest.NewServer(r)
+	defer srv.Close()
+
+	// Create both indexes (4-d for test simplicity).
+	for _, name := range []string{"fresh_test", "main_test"} {
+		body := `{"name":"` + name + `","dimension":4,"distance":"cosine"}`
+		resp, err := http.Post(srv.URL+"/vectors/index", "application/json", strings.NewReader(body))
+		if err != nil {
+			t.Fatalf("create %s: %v", name, err)
+		}
+		resp.Body.Close()
+	}
+
+	// Populate fresh_test with 3 items.
+	addBody := `{"items":[
+		{"id":"f-1","vector":[1,0,0,0],"metadata":{"name":"fresh-001"}},
+		{"id":"f-2","vector":[0,1,0,0],"metadata":{"name":"fresh-002"}},
+		{"id":"f-3","vector":[0,0,1,0],"metadata":{"name":"fresh-003"}}
+	]}`
+	resp, err := http.Post(srv.URL+"/vectors/index/fresh_test/add", "application/json", strings.NewReader(addBody))
+	if err != nil || resp.StatusCode != http.StatusOK {
+		t.Fatalf("add to fresh_test: status=%d err=%v", resp.StatusCode, err)
+	}
+	resp.Body.Close()
+
+	// Pre-seed main_test with one item that ALSO exists in fresh
+	// (collision) so we exercise the skipped_already_present path.
+	preBody := `{"items":[{"id":"f-1","vector":[1,0,0,0],"metadata":{"name":"main-collision"}}]}`
+	resp, err = http.Post(srv.URL+"/vectors/index/main_test/add", "application/json", strings.NewReader(preBody))
+	if err != nil || resp.StatusCode != http.StatusOK {
+		t.Fatalf("add collision to main_test: status=%d err=%v", resp.StatusCode, err)
+	}
+	resp.Body.Close()
+
+	// Merge fresh_test → main_test, clearing source.
+	mergeBody := `{"dest":"main_test","clear_source":true}`
+	resp, err = http.Post(srv.URL+"/vectors/index/fresh_test/merge", "application/json", strings.NewReader(mergeBody))
+	if err != nil {
+		t.Fatalf("merge: %v", err)
+	}
+	defer resp.Body.Close()
+	if resp.StatusCode != http.StatusOK {
+		t.Errorf("expected 200 on merge, got %d", resp.StatusCode)
+	}
+	var out mergeResponse
+	if err := json.NewDecoder(resp.Body).Decode(&out); err != nil {
+		t.Fatalf("decode merge resp: %v", err)
+	}
+	if out.Merged != 2 {
+		t.Errorf("expected 2 merged (f-2 + f-3), got %d", out.Merged)
+	}
+	if out.SkippedAlreadyPresent != 1 {
+		t.Errorf("expected 1 skipped (f-1 collision), got %d", out.SkippedAlreadyPresent)
+	}
+	if out.LengthSource != 0 {
+		t.Errorf("expected source emptied, got len=%d", out.LengthSource)
+	}
+	if out.LengthDest != 3 {
+		t.Errorf("expected dest len=3 after merge, got %d", out.LengthDest)
+	}
+}
+
+func TestHandleMerge_DimensionMismatch_400(t *testing.T) {
+	h := &handlers{reg: vectord.NewRegistry()}
+	r := chi.NewRouter()
+	h.register(r)
+	srv := httptest.NewServer(r)
+	defer srv.Close()
+
+	for _, c := range []struct{ name string; dim int }{
+		{"src_4d", 4},
+		{"dst_8d", 8},
+	} {
+		body := `{"name":"` + c.name + `","dimension":` + strconv.Itoa(c.dim) + `,"distance":"cosine"}`
+		resp, err := http.Post(srv.URL+"/vectors/index", "application/json", strings.NewReader(body))
+		if err != nil {
+			t.Fatalf("create %s: %v", c.name, err)
+		}
+		resp.Body.Close()
+	}
+
+	resp, err := http.Post(srv.URL+"/vectors/index/src_4d/merge", "application/json",
+		strings.NewReader(`{"dest":"dst_8d"}`))
+	if err != nil {
+		t.Fatalf("merge: %v", err)
+	}
+	defer resp.Body.Close()
+	if resp.StatusCode != http.StatusBadRequest {
+		t.Errorf("expected 400 on dim mismatch, got %d", resp.StatusCode)
+	}
+}
+
+func TestHandleMerge_DestNotFound_404(t *testing.T) {
+	h := &handlers{reg: vectord.NewRegistry()}
+	r := chi.NewRouter()
+	h.register(r)
+	srv := httptest.NewServer(r)
+	defer srv.Close()
+
+	body := `{"name":"only_src","dimension":4}`
+	resp, err := http.Post(srv.URL+"/vectors/index", "application/json", strings.NewReader(body))
+	if err != nil {
+		t.Fatalf("create: %v", err)
+	}
+	resp.Body.Close()
+
+	resp, err = http.Post(srv.URL+"/vectors/index/only_src/merge", "application/json",
+		strings.NewReader(`{"dest":"missing_dest"}`))
+	if err != nil {
+		t.Fatalf("merge: %v", err)
+	}
+	defer resp.Body.Close()
+	if resp.StatusCode != http.StatusNotFound {
+		t.Errorf("expected 404 for missing dest, got %d", resp.StatusCode)
+	}
+}
+
+func TestHandleMerge_SameSourceDest_400(t *testing.T) {
+	h := &handlers{reg: vectord.NewRegistry()}
+	r := chi.NewRouter()
+	h.register(r)
+	srv := httptest.NewServer(r)
+	defer srv.Close()
+
+	body := `{"name":"self","dimension":4}`
+	resp, err := http.Post(srv.URL+"/vectors/index", "application/json", strings.NewReader(body))
+	if err != nil {
+		t.Fatalf("create: %v", err)
+	}
+	resp.Body.Close()
+
+	resp, err = http.Post(srv.URL+"/vectors/index/self/merge", "application/json",
+		strings.NewReader(`{"dest":"self"}`))
+	if err != nil {
+		t.Fatalf("merge: %v", err)
+	}
+	defer resp.Body.Close()
+	if resp.StatusCode != http.StatusBadRequest {
+		t.Errorf("expected 400 for self-merge, got %d", resp.StatusCode)
+	}
+}
+
 func TestSearchK_DefaultsAndMax(t *testing.T) {
 	if defaultK <= 0 {
 		t.Errorf("defaultK = %d, must be > 0", defaultK)