diff --git a/cmd/logos/memory.go b/cmd/logos/memory.go index f25b661..ecf6953 100644 --- a/cmd/logos/memory.go +++ b/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) diff --git a/internal/memory/consolidate.go b/internal/memory/consolidate.go index 172ebb7..119c42f 100644 --- a/internal/memory/consolidate.go +++ b/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,9 +231,9 @@ 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 @@ -233,7 +241,9 @@ func Consolidate(db *sql.DB, rt *router.Router) (merged int, superseded int, err 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 { diff --git a/internal/memory/consolidate_test.go b/internal/memory/consolidate_test.go index d80683c..c11d77f 100644 --- a/internal/memory/consolidate_test.go +++ b/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) + } + }) + } +} diff --git a/internal/memory/vaultstore.go b/internal/memory/vaultstore.go index 9cff1ed..c69f0ee 100644 --- a/internal/memory/vaultstore.go +++ b/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 diff --git a/internal/memory/vaultstore_test.go b/internal/memory/vaultstore_test.go index cbb1db4..f9520cf 100644 --- a/internal/memory/vaultstore_test.go +++ b/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) + } +}