Consolidate reports refused merges as errors, and a rebound memory store rereads its vault

cmd/logos/memory.go

@@ -147,6 +147,11 @@ func memoryCmd(args []string) error {
 		}
 		m, s, err := memory.Consolidate(ix.DB, rt)
 		if err != nil {
+			// What went through before the failure is real and already in the
+			// vault; an error alone would read as "nothing changed".
+			if m+s > 0 {
+				fmt.Printf("merged %d duplicates · superseded %d outdated before stopping\n", m, s)
+			}
 			return err
 		}
 		fmt.Printf("merged %d duplicates · superseded %d outdated · %d faded from disuse (ranked lower, not removed)\n", m, s, faded)

internal/memory/consolidate.go

@@ -3,6 +3,7 @@ package memory
 import (
 	"database/sql"
 	"encoding/json"
+	"errors"
 	"math"
 	"strings"
 	"time"
@@ -190,6 +191,13 @@ func Consolidate(db *sql.DB, rt *router.Router) (merged int, superseded int, err
 	gone := map[int64]bool{}
 	const gate = 0.80
 
+	// A refused write stops the pass rather than being counted: the counts were
+	// reported as "merged N" whether or not the UPDATE went through, so a
+	// database that refused every write produced a success receipt over a store
+	// nothing had changed. What did go through is still flushed below, and the
+	// error comes back with those counts.
+	var werr error
+pairs:
 	for i := 0; i < len(mems); i++ {
 		if gone[mems[i].ID] {
 			continue
@@ -223,17 +231,19 @@ func Consolidate(db *sql.DB, rt *router.Router) (merged int, superseded int, err
 				if older.Salience >= newer.Salience {
 					keep, drop = older, newer
 				}
-				db.Exec("UPDATE memories SET salience = ?, confidence = MIN(1.0, confidence + 0.05), uses = uses + ? WHERE id = ?",
-					math.Min(1, keep.Salience+0.1), drop.Uses, keep.ID)
-				db.Exec("UPDATE memories SET superseded = 1, superseded_by = ? WHERE id = ?", keep.ID, drop.ID)
+				if werr = mergeInto(db, keep, drop); werr != nil {
+					break pairs
+				}
 				logEvent(db, keep.ID, EvMerged, keep.Text, drop.ID)
 				logEvent(db, drop.ID, EvSuperseded, drop.Text, keep.ID)
 				gone[drop.ID] = true
 				merged++
 			case "update":
 				// The newer fact wins; the older is superseded but retained, with a
 				// pointer to what replaced it so the timeline can show the change.
-				db.Exec("UPDATE memories SET superseded = 1, superseded_by = ? WHERE id = ?", newer.ID, older.ID)
+				if _, werr = db.Exec("UPDATE memories SET superseded = 1, superseded_by = ? WHERE id = ?", newer.ID, older.ID); werr != nil {
+					break pairs
+				}
 				logEvent(db, older.ID, EvSuperseded, older.Text, newer.ID)
 				gone[older.ID] = true
 				superseded++
@@ -245,11 +255,30 @@ func Consolidate(db *sql.DB, rt *router.Router) (merged int, superseded int, err
 	if merged+superseded > 0 {
 		for _, k := range kinds {
 			if err := flush(db, k); err != nil {
-				return merged, superseded, err
+				return merged, superseded, errors.Join(werr, err)
 			}
 		}
 	}
-	return merged, superseded, nil
+	return merged, superseded, werr
+}
+
+// mergeInto folds drop into keep in one transaction. As two separate writes, a
+// failure between them left keep's salience raised for a merge that never
+// happened, with drop still active beside it.
+func mergeInto(db *sql.DB, keep, drop Memory) error {
+	tx, err := db.Begin()
+	if err != nil {
+		return err
+	}
+	defer tx.Rollback()
+	if _, err := tx.Exec("UPDATE memories SET salience = ?, confidence = MIN(1.0, confidence + 0.05), uses = uses + ? WHERE id = ?",
+		math.Min(1, keep.Salience+0.1), drop.Uses, keep.ID); err != nil {
+		return err
+	}
+	if _, err := tx.Exec("UPDATE memories SET superseded = 1, superseded_by = ? WHERE id = ?", keep.ID, drop.ID); err != nil {
+		return err
+	}
+	return tx.Commit()
 }
 
 func classify(rt *router.Router, model, older, newer string) string {

internal/memory/consolidate_test.go

@@ -2,6 +2,9 @@ package memory
 
 import (
 	"errors"
+	"fmt"
+	"net/http"
+	"net/http/httptest"
 	"testing"
 
 	"github.com/Coder8124/logos/internal/router"
@@ -20,3 +23,53 @@ func TestConsolidateWithNilRouterReturnsErrNoRuntimeInsteadOfPanicking(t *testin
 		t.Fatalf("Consolidate(db, nil) = %v, want an error wrapping router.ErrNoRuntime", err)
 	}
 }
+
+// chatRuntime is a configured runtime that lists the default T1 model and
+// answers every chat with the same JSON: ok for the router's capability probe,
+// relation for Consolidate's classifier.
+func chatRuntime(t *testing.T, relation string) {
+	t.Helper()
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		switch r.URL.Path {
+		case "/models":
+			w.Write([]byte(`{"data":[{"id":"gemma3:4b"}]}`))
+		case "/chat/completions":
+			fmt.Fprintf(w, `{"choices":[{"message":{"content":"{\"ok\":true,\"relation\":\"%s\"}"}}]}`, relation)
+		default:
+			http.NotFound(w, r)
+		}
+	}))
+	t.Cleanup(srv.Close)
+	t.Setenv("LOGOS_RUNTIME", srv.URL)
+}
+
+// Consolidate counted a merge or a supersession whether or not the UPDATE
+// that performed it went through, and never looked at the error. With the
+// database refusing writes, `logos memory consolidate` reported "merged 1"
+// over a store it had not changed — a failure returned in the shape of a
+// success, which is the one outcome this codebase does not allow.
+func TestAConsolidationTheDatabaseRefusedIsReportedAsAnErrorNotAsAMerge(t *testing.T) {
+	for _, relation := range []string{"duplicate", "update"} {
+		t.Run(relation, func(t *testing.T) {
+			chatRuntime(t, relation)
+			db := testDB(t)
+			storeVec(t, db, "deploys go out on Tuesdays", Fact, 0.5, []float32{1, 0, 0})
+			storeVec(t, db, "deploys go out on Tuesday", Fact, 0.5, []float32{1, 0, 0})
+			if _, err := db.Exec(`CREATE TRIGGER refuse BEFORE UPDATE ON memories BEGIN SELECT RAISE(ABORT, 'disk I/O error'); END`); err != nil {
+				t.Fatal(err)
+			}
+			rt, err := router.New(nil, "")
+			if err != nil {
+				t.Fatal(err)
+			}
+
+			merged, superseded, err := Consolidate(db, rt)
+			if err == nil {
+				t.Error("Consolidate returned no error although every UPDATE was refused")
+			}
+			if merged+superseded != 0 {
+				t.Errorf("Consolidate reported merged=%d superseded=%d although nothing was written", merged, superseded)
+			}
+		})
+	}
+}

internal/memory/vaultstore.go

@@ -63,11 +63,15 @@ var (
 func SetVault(db *sql.DB, dir string) {
 	forgetLoggedHigh(db)
 	pendingStamps.Forget(db)
+	// On every bind, not only on unbind: a stamp describes a file in the vault
+	// it was taken in. Carried across a rebind, a copied vault's file matched
+	// it byte for byte, the next write skipped adopting it, and the rewrite
+	// deleted every line the handle's rows did not already hold.
+	dropStamps(db)
 	vaultMu.Lock()
 	defer vaultMu.Unlock()
 	if dir == "" {
 		delete(vaults, db)
-		dropStamps(db)
 		return
 	}
 	vaults[db] = dir

internal/memory/vaultstore_test.go

@@ -442,3 +442,47 @@ func TestTheUsageSignalSurvivesDeletingTheIndex(t *testing.T) {
 		t.Errorf("last_used came back as %d, not %d — decay restarts from created", all[0].LastUsed, used)
 	}
 }
+
+// A stamp says "this file's bytes are ones this handle wrote, so its lines are
+// already rows". SetVault forgot that claim on unbind but kept it on a rebind,
+// so a handle moved to a second vault carried claims about the first. When the
+// second vault's file had the same bytes — a copied vault — the next write
+// skipped adopting it and rewrote the file from rows that never held its
+// lines, deleting them from the vault without a word.
+func TestRebindingToAnotherVaultDoesNotTrustTheFirstVaultsStamps(t *testing.T) {
+	db, first := vaultDB(t)
+	m := Memory{Text: "The staging cluster has no rollback", Kind: Fact, Source: "manual"}
+	if _, err := Store(db, nil, "", &m); err != nil {
+		t.Fatal(err)
+	}
+
+	second := t.TempDir()
+	if err := os.MkdirAll(filepath.Join(second, Dir), 0o755); err != nil {
+		t.Fatal(err)
+	}
+	raw, err := os.ReadFile(filepath.Join(first, Dir, "fact.md"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := os.WriteFile(filepath.Join(second, Dir, "fact.md"), raw, 0o644); err != nil {
+		t.Fatal(err)
+	}
+	// The handle's rows are not the second vault's: it has never read it.
+	if _, err := db.Exec(`DELETE FROM memories`); err != nil {
+		t.Fatal(err)
+	}
+
+	SetVault(db, second)
+	next := Memory{Text: "Deploys go out on Tuesdays", Kind: Fact, Source: "manual"}
+	if _, err := Store(db, nil, "", &next); err != nil {
+		t.Fatal(err)
+	}
+
+	got, err := os.ReadFile(filepath.Join(second, Dir, "fact.md"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	if !strings.Contains(string(got), "staging cluster has no rollback") {
+		t.Errorf("the second vault's own line was deleted by a write trusting the first vault's stamp:\n%s", got)
+	}
+}