From 3d0668f50dd2db8e6898585a09629b9565b91575 Mon Sep 17 00:00:00 2001 From: Bryan Woodruff Date: Tue, 25 Aug 2026 06:49:33 -0700 Subject: [PATCH 1/5] fix(kg): clear the review-follow-up backlog MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Four acknowledged-but-unfiled items from this week's reviews: - kg health gains ADR-009's observation-age stats: newest/oldest/median stored created_at over timestamped observations (zero/NULL rows excluded — they are the zero-timestamp count, and would report every legacy-bearing graph as centuries old), rendered as ages in the human report. Median is positional (SKIP (n-1)/2), pinned by a fixture with known 2020/2022 rows so a query grabbing first, last, or a legacy row fails the test. - kg stats shares resolveScopeDB with kg health: an explicitly named scope that cannot be loaded is now an error, never a silent fallback to the legacy database. - kglib's HNSW build accepts float64 embedding components (converted) and rejects vectors with any other component type instead of silently zeroing them — a zeroed component distorted every distance the vector participated in, invisibly. - The hub reconciles installs at construction: a hard kill between repointing `current` and writing the registry left search reading one commit's database with another commit's ProjectID — 200 with zero results, permanently. The registry write is the commit point, so an unregistered `current` target now rolls back to the registered commit; the orphaned directory is left for the next install's prune. All four are mutation-verified. Co-Authored-By: Claude Fable 5 --- src/kg/health.go | 39 +++++++++-- src/kg/health_test.go | 45 ++++++++++++- src/kg/internal/hub/install_test.go | 51 ++++++++++++++ src/kg/internal/hub/server.go | 47 +++++++++++++ src/kg/internal/knowledge/health.go | 100 ++++++++++++++++++++++++++++ src/kg/stats.go | 28 ++------ src/kglib/hnsw_index.go | 32 +++++++-- src/kglib/hnsw_index_test.go | 24 +++++++ 8 files changed, 327 insertions(+), 39 deletions(-) create mode 100644 src/kglib/hnsw_index_test.go diff --git a/src/kg/health.go b/src/kg/health.go index 565a477..0431587 100644 --- a/src/kg/health.go +++ b/src/kg/health.go @@ -7,6 +7,7 @@ import ( "os" "path/filepath" "sort" + "time" "github.com/cortexa-llc/mcp/kg/internal/knowledge" "github.com/spf13/cobra" @@ -72,7 +73,7 @@ func runHealth(root, scopeName string, jsonOut bool, out io.Writer) error { aiDir := filepath.Join(root, ".ai") projectID := projectIDFromCwd(root) - dbPath, scopeName, err := resolveHealthDB(aiDir, scopeName) + dbPath, scopeName, err := resolveScopeDB(aiDir, scopeName) if err != nil { return err } @@ -121,15 +122,14 @@ func runHealth(root, scopeName string, jsonOut bool, out io.Writer) error { return nil } -// resolveHealthDB resolves the database path: explicit scope, else the -// default scope, else the legacy knowledge.db. Returns the path and the scope -// name actually used ("" for legacy). This mirrors `kg stats` with one -// deliberate divergence: stats silently falls back to the legacy database -// when a named scope cannot be loaded, health errors (see below). +// resolveScopeDB resolves the database path for read-only report commands +// (kg health, kg stats): explicit scope, else the default scope, else the +// legacy knowledge.db. Returns the path and the scope name actually used +// ("" for legacy). // A named scope that cannot be loaded is an error, never a silent fallback — // reporting the legacy database's health under a scope the user asked for // would be a wrong answer, not a degraded one. -func resolveHealthDB(aiDir, scopeName string) (string, string, error) { +func resolveScopeDB(aiDir, scopeName string) (string, string, error) { if scopeName == "" { defaultScope, err := knowledge.GetDefaultScope(aiDir) if err != nil { @@ -220,6 +220,12 @@ func printHealth(out io.Writer, o healthOutput, snapPath string) { fmt.Fprintf(out, "\nNo previous snapshot — growth will be reported from the next run.\n") } + if oa := o.Current.ObservationAge; oa != nil { + gen := o.Current.GeneratedAt + fmt.Fprintf(out, "\nObservation age: newest %s, median %s, oldest %s\n", + humanAge(gen.Sub(oa.Newest)), humanAge(gen.Sub(oa.Median)), humanAge(gen.Sub(oa.Oldest))) + } + share := 0.0 if o.Current.Observations > 0 { share = float64(o.Current.ZeroTimestampObservations) / float64(o.Current.Observations) * 100 @@ -232,6 +238,25 @@ func printHealth(out io.Writer, o healthOutput, snapPath string) { fmt.Fprintf(out, "\nSnapshot written to %s\n", snapPath) } +// humanAge renders a duration as a coarse age ("3h", "2d", "5mo") for the +// observation-age line; precision beyond this is noise in a health report. +func humanAge(d time.Duration) string { + switch { + case d < time.Minute: + return "just now" + case d < time.Hour: + return fmt.Sprintf("%dm", int(d.Minutes())) + case d < 24*time.Hour: + return fmt.Sprintf("%dh", int(d.Hours())) + case d < 30*24*time.Hour: + return fmt.Sprintf("%dd", int(d.Hours()/24)) + case d < 365*24*time.Hour: + return fmt.Sprintf("%dmo", int(d.Hours()/(24*30))) + default: + return fmt.Sprintf("%dy", int(d.Hours()/(24*365))) + } +} + // formatEntityTypes renders the by-type counts sorted by count descending, // then name, so the report is stable between runs. func formatEntityTypes(byType map[string]int) []string { diff --git a/src/kg/health_test.go b/src/kg/health_test.go index 735a3d6..b3b8c20 100644 --- a/src/kg/health_test.go +++ b/src/kg/health_test.go @@ -7,6 +7,7 @@ import ( "path/filepath" "strings" "testing" + "time" "github.com/cortexa-llc/mcp/kg/internal/knowledge" ) @@ -63,6 +64,13 @@ func seedHealthFixture(t *testing.T, root string) { for _, mutation := range []string{ "SET o.created_at = NULL", `SET o.created_at = timestamp("0001-01-01 00:00:00")`, + // Two dated rows pin the age stats: with the two freshly-written + // observations above, the four timestamped rows sort as + // [2020, 2022, now, now], so oldest must be 2020 and the (lower) + // median must be 2022 — a median query that grabs the first, last, + // or a legacy row cannot pass. + `SET o.created_at = timestamp("2020-01-01 00:00:00")`, + `SET o.created_at = timestamp("2022-06-15 00:00:00")`, } { obs, err := store.CreateObservation(e2.ID, "legacy-era note", projectID) if err != nil { @@ -110,8 +118,8 @@ func TestHealthCommandReportsMetricsAndGrowth(t *testing.T) { if out.Current.Relations != 1 { t.Errorf("relations = %d, want 1", out.Current.Relations) } - if out.Current.Observations != 4 { - t.Errorf("observations = %d, want 4", out.Current.Observations) + if out.Current.Observations != 6 { + t.Errorf("observations = %d, want 6", out.Current.Observations) } // Exactly the two deliberately aged rows — not 0 (metric dead) and not 4 // (metric counting scan artifacts instead of stored values). @@ -122,6 +130,19 @@ func TestHealthCommandReportsMetricsAndGrowth(t *testing.T) { if out.Current.OrphanedEntities != 1 { t.Errorf("orphaned entities = %d, want 1", out.Current.OrphanedEntities) } + if oa := out.Current.ObservationAge; oa == nil { + t.Error("observation age missing with 4 timestamped observations") + } else { + if oa.Oldest.Year() != 2020 { + t.Errorf("oldest observation year = %d, want 2020", oa.Oldest.Year()) + } + if oa.Median.Year() != 2022 { + t.Errorf("median observation year = %d, want 2022 (lower median of [2020 2022 now now])", oa.Median.Year()) + } + if time.Since(oa.Newest) > time.Minute { + t.Errorf("newest observation = %v, want within the last minute", oa.Newest) + } + } if out.Current.ObsoleteObservations != 1 { t.Errorf("[OBSOLETE observations = %d, want 1", out.Current.ObsoleteObservations) } @@ -202,9 +223,27 @@ func TestHealthCommandHumanOutput(t *testing.T) { if err := runHealth(root, "", false, &buf); err != nil { t.Fatalf("runHealth: %v", err) } - for _, want := range []string{"legacy, age unknown", "Orphaned entities", "No previous snapshot"} { + for _, want := range []string{"legacy, age unknown", "Orphaned entities", "No previous snapshot", "Observation age: newest"} { if !strings.Contains(buf.String(), want) { t.Errorf("human output missing %q:\n%s", want, buf.String()) } } } + +// An explicitly named scope that cannot be loaded is an error for both +// report commands (kg health and kg stats share resolveScopeDB) — never a +// silent fallback to the legacy database, which answers the wrong question. +func TestResolveScopeDBErrorsOnUnloadableScope(t *testing.T) { + aiDir := filepath.Join(t.TempDir(), ".ai") + if err := os.MkdirAll(aiDir, 0o755); err != nil { + t.Fatalf("MkdirAll: %v", err) + } + if _, _, err := resolveScopeDB(aiDir, "no-such-scope"); err == nil { + t.Error("resolveScopeDB with an unloadable scope returned nil error, want failure") + } + // No scope named and none configured: the legacy database, no error. + dbPath, scopeName, err := resolveScopeDB(aiDir, "") + if err != nil || scopeName != "" || filepath.Base(dbPath) != "knowledge.db" { + t.Errorf("legacy resolution = (%q, %q, %v), want knowledge.db path, empty scope, nil", dbPath, scopeName, err) + } +} diff --git a/src/kg/internal/hub/install_test.go b/src/kg/internal/hub/install_test.go index 0d76a0f..fca6fb3 100644 --- a/src/kg/internal/hub/install_test.go +++ b/src/kg/internal/hub/install_test.go @@ -3,8 +3,10 @@ package hub import ( "encoding/json" "net/http" + "net/http/httptest" "os" "path/filepath" + "strings" "testing" "time" @@ -142,3 +144,52 @@ func TestRegistryWriteFailureLeavesGraphUnchanged(t *testing.T) { t.Errorf("retry did not install: commit = %q (want %q), results = %v", commit, commitB, names) } } + +// A hard kill between install's two renames — `current` repointed, registry +// not yet written — leaves search reading one commit's database with another +// commit's ProjectID: HTTP 200 with zero results, permanently. Construction +// reconciles: the registry write is the commit point, so `current` rolls +// back to the registered commit and the graph answers again. +func TestReconcileRollsBackInterruptedSeed(t *testing.T) { + dataDir := t.TempDir() + ts := httptest.NewServer(NewServer(dataDir, "", "s3cret", "dev").Handler()) + + commit := strings.Repeat("a", 40) + dbPath := buildFixtureDBFor(t, "ReconcileEntity", "proj-one", commit) + if err := pushFixtureFor(t, ts.URL, "recon", dbPath, commit, "proj-one"); err != nil { + t.Fatalf("push fixture: %v", err) + } + if names, _ := searchNames(t, ts.URL, "recon", "ReconcileEntity"); len(names) == 0 { + t.Fatal("fixture graph does not answer before the simulated crash") + } + ts.Close() + + // Simulate the crash: a new commit directory exists and `current` points + // at it, but the process died before the registry recorded it. The empty + // directory doubles as a tripwire — if reconciliation does not roll back, + // the search below fails against the missing database rather than + // silently passing. + gdir := filepath.Join(dataDir, "graphs", "recon") + orphan := strings.Repeat("b", 40) + if err := os.MkdirAll(filepath.Join(gdir, orphan), 0o755); err != nil { + t.Fatalf("create orphan commit dir: %v", err) + } + if err := replaceSymlink(gdir, filepath.Join(gdir, "current"), orphan); err != nil { + t.Fatalf("repoint current: %v", err) + } + + // Restart the hub over the same data dir. + ts2 := httptest.NewServer(NewServer(dataDir, "", "s3cret", "dev").Handler()) + defer ts2.Close() + + if target, err := os.Readlink(filepath.Join(gdir, "current")); err != nil || target != commit { + t.Errorf("current -> %q (err %v), want rolled back to %q", target, err, commit) + } + names, gotCommit := searchNames(t, ts2.URL, "recon", "ReconcileEntity") + if len(names) == 0 || names[0] != "ReconcileEntity" { + t.Errorf("search after reconcile = %v, want [ReconcileEntity]", names) + } + if gotCommit != commit { + t.Errorf("served commit = %q, want %q", gotCommit, commit) + } +} diff --git a/src/kg/internal/hub/server.go b/src/kg/internal/hub/server.go index 867c529..21b212f 100644 --- a/src/kg/internal/hub/server.go +++ b/src/kg/internal/hub/server.go @@ -68,6 +68,7 @@ func NewServerWithAuth(dataDir string, readVerifier, seedVerifier Verifier, kgVe kgVersion: kgVersion, } s.sweepStaging() + s.reconcileInstalls() return s } @@ -98,6 +99,52 @@ func (s *Server) sweepStaging() { } } +// reconcileInstalls restores the invariant that `current` and the registry +// name the same commit for every graph. +// +// install's rollback covers error returns, but a hard kill between its two +// renames — repointing `current`, then writing the registry — leaves +// `current` on a commit the registry never recorded. That split is silent +// and permanent: a search reads the database from `current` but queries it +// with the registry's ProjectID, so the graph answers HTTP 200 with zero +// results whenever the project ID changed, and no later read corrects it. +// +// The registry write is install's commit point, so the registry is the +// authority: an unregistered `current` target means the seed never +// committed, and the symlink is rolled back to the registered commit. The +// orphaned commit directory is deliberately left in place — the next +// successful install's prune removes it, and deleting data is not a job for +// a constructor-time repair pass. +func (s *Server) reconcileInstalls() { + reg, err := loadRegistry(s.dataDir) + if err != nil { + log.Printf("reconcile installs: load registry: %v", err) + return + } + for name, info := range reg.Graphs { + // registry.json is hand-editable: validate everything joined into a + // path, as everywhere else. + if !validPathComponent(name) || !validPathComponent(info.Commit) { + continue + } + gdir := s.graphDir(name) + currentLink := filepath.Join(gdir, "current") + target, err := os.Readlink(currentLink) + if err != nil || target == info.Commit { + continue + } + if _, err := os.Stat(filepath.Join(gdir, info.Commit)); err != nil { + log.Printf("reconcile %s: current -> %s but registry records %s, whose directory is missing — leaving as is", name, target, info.Commit) + continue + } + if err := replaceSymlink(gdir, currentLink, info.Commit); err != nil { + log.Printf("reconcile %s: repoint current %s -> %s: %v", name, target, info.Commit, err) + continue + } + log.Printf("reconcile %s: current pointed at unregistered commit %s (interrupted seed); rolled back to registered %s", name, target, info.Commit) + } +} + // ReadAuthEnabled reports whether read requests require authentication. func (s *Server) ReadAuthEnabled() bool { return s.readVerifier != nil } diff --git a/src/kg/internal/knowledge/health.go b/src/kg/internal/knowledge/health.go index 7b32e36..0f530c9 100644 --- a/src/kg/internal/knowledge/health.go +++ b/src/kg/internal/knowledge/health.go @@ -21,6 +21,14 @@ type HealthMetrics struct { // a non-zero count here means some other or older writer produced the rows. ZeroTimestampObservations int `json:"zero_timestamp_observations"` + // ObservationAge summarizes stored created_at across observations that + // carry a real timestamp (ADR-009: newest/oldest/median observation age). + // Zero-timestamp rows are excluded — they are what + // ZeroTimestampObservations counts, and including them would report every + // legacy-bearing graph as centuries old. Nil when no timestamped + // observations exist. + ObservationAge *ObservationAge `json:"observation_age,omitempty"` + // OrphanedEntities counts entities with no observations and no relations // in either direction — nodes nothing points at and that say nothing. OrphanedEntities int `json:"orphaned_entities"` @@ -98,6 +106,10 @@ func CollectHealthMetrics(store *Store, projectID string) (*HealthMetrics, error return nil, fmt.Errorf("count zero-timestamp observations: %w", err) } + if m.ObservationAge, err = collectObservationAge(store, projectID, m.Observations-m.ZeroTimestampObservations); err != nil { + return nil, fmt.Errorf("collect observation age: %w", err) + } + // The unlabeled patterns match every relationship table, HAS_OBSERVATION // included, so one predicate covers "no observations AND no relations". m.OrphanedEntities, err = countQuery(store, ` @@ -121,6 +133,94 @@ func CollectHealthMetrics(store *Store, projectID string) (*HealthMetrics, error return m, nil } +// ObservationAge holds the stored created_at of the newest, oldest, and +// median timestamped observation. Timestamps rather than durations, so the +// values are stable in snapshots; the CLI renders them as ages. +type ObservationAge struct { + Newest time.Time `json:"newest"` + Oldest time.Time `json:"oldest"` + Median time.Time `json:"median"` +} + +// timestampedObsFilter excludes NULL and stored-zero created_at rows — the +// same literal-not-parameter rule as the zero-timestamp count above. +const timestampedObsFilter = `o.created_at IS NOT NULL AND o.created_at <> timestamp("0001-01-01 00:00:00")` + +// collectObservationAge computes newest/oldest/median stored created_at over +// the project's timestamped observations. timestamped is their count, already +// known to the caller (total minus zero-timestamp); 0 yields nil. +func collectObservationAge(store *Store, projectID string, timestamped int) (*ObservationAge, error) { + if timestamped <= 0 { + return nil, nil + } + + age := &ObservationAge{} + result, err := store.QueryParams(` + MATCH (e:Entity {project_id: $project_id})-[:HAS_OBSERVATION]->(o:Observation) + WHERE `+timestampedObsFilter+` + RETURN max(o.created_at), min(o.created_at) + `, map[string]any{"project_id": projectID}) + if err != nil { + return nil, err + } + if result.HasNext() { + row, err := result.Next() + if err != nil { + result.Close() + return nil, err + } + if age.Newest, err = timeCell(row, 0); err != nil { + result.Close() + return nil, err + } + if age.Oldest, err = timeCell(row, 1); err != nil { + result.Close() + return nil, err + } + } + result.Close() + + // Median by position: the middle row (lower of the two for even counts) + // of the timestamped observations in created_at order. + result, err = store.QueryParams(fmt.Sprintf(` + MATCH (e:Entity {project_id: $project_id})-[:HAS_OBSERVATION]->(o:Observation) + WHERE `+timestampedObsFilter+` + RETURN o.created_at + ORDER BY o.created_at + SKIP %d LIMIT 1 + `, (timestamped-1)/2), map[string]any{"project_id": projectID}) + if err != nil { + return nil, err + } + defer result.Close() + if !result.HasNext() { + return nil, fmt.Errorf("median query returned no row for %d timestamped observations", timestamped) + } + row, err := result.Next() + if err != nil { + return nil, err + } + if age.Median, err = timeCell(row, 0); err != nil { + return nil, err + } + return age, nil +} + +// timeCell reads a timestamp cell. An unexpected type is an error, not a zero +// time — the silent-fallback rule from intFromCount applies doubly to the +// metric that exists because timestamps were once silently zeroed. +func timeCell(row interface{ GetValue(uint64) (any, error) }, col uint64) (time.Time, error) { + v, err := row.GetValue(col) + if err != nil { + return time.Time{}, err + } + t, ok := v.(time.Time) + if !ok { + return time.Time{}, fmt.Errorf("timestamp cell has unexpected type %T", v) + } + return t.UTC(), nil +} + // countQuery runs a single-row count(*) query and returns the count. func countQuery(store *Store, query string, params map[string]any) (int, error) { result, err := store.QueryParams(query, params) diff --git a/src/kg/stats.go b/src/kg/stats.go index 8aba558..233248e 100644 --- a/src/kg/stats.go +++ b/src/kg/stats.go @@ -37,33 +37,13 @@ By default, shows stats for the default scope. Use --scope to specify a differen projectID := projectIDFromCwd(cwd) // Determine which scope to use - scopeName := statsScopeName - if scopeName == "" { - // Use default scope - defaultScope, err := knowledge.GetDefaultScope(aiDir) - if err != nil { - return err - } - scopeName = defaultScope - } - - // Open appropriate store - var dbPath string - configs, err := knowledge.ListScopeConfigs(aiDir) + // Shared with kg health: an explicitly named scope that cannot be + // loaded is an error, never a silent fallback to the legacy database. + dbPath, scopeName, err := resolveScopeDB(aiDir, statsScopeName) if err != nil { return err } - - if len(configs) == 0 || scopeName == "" { - // Legacy mode - dbPath = filepath.Join(aiDir, "knowledge.db") - } else { - // Load scope config - cfg, err := knowledge.LoadScopeConfig(aiDir, scopeName) - if err != nil { - return err - } - dbPath = filepath.Join(aiDir, cfg.Database) + if scopeName != "" { fmt.Printf("Stats for scope: %s\n", scopeName) } diff --git a/src/kglib/hnsw_index.go b/src/kglib/hnsw_index.go index fc37402..f47d6fb 100644 --- a/src/kglib/hnsw_index.go +++ b/src/kglib/hnsw_index.go @@ -110,11 +110,12 @@ func (s *Store) buildIndex(projectID string) (*projectIndex, error) { if !ok || len(rawEmb) == 0 { continue } - emb := make([]float32, len(rawEmb)) - for i, v := range rawEmb { - if f, ok := v.(float32); ok { - emb[i] = f - } + emb, ok := embeddingFromRaw(rawEmb) + if !ok { + // A component of unhandled type must exclude the whole node: the + // old behaviour left it at 0, silently distorting every distance + // this vector participated in. + continue } entities[entity.ID] = entity @@ -131,3 +132,24 @@ func (s *Store) buildIndex(projectID string) (*projectIndex, error) { builtAt: time.Now().UTC(), }, nil } + +// embeddingFromRaw converts a Kuzu list cell to an embedding vector. go-kuzu +// returns float32 for FLOAT columns today, but a DOUBLE column — or a driver +// change — arrives as float64; both are accepted. Any other component type +// returns ok=false so the caller can drop the vector: zeroing the component, +// which is what a bare type assertion used to do, silently corrupts every +// distance the vector participates in and is invisible in search results. +func embeddingFromRaw(raw []any) ([]float32, bool) { + emb := make([]float32, len(raw)) + for i, v := range raw { + switch f := v.(type) { + case float32: + emb[i] = f + case float64: + emb[i] = float32(f) + default: + return nil, false + } + } + return emb, true +} diff --git a/src/kglib/hnsw_index_test.go b/src/kglib/hnsw_index_test.go new file mode 100644 index 0000000..4d60927 --- /dev/null +++ b/src/kglib/hnsw_index_test.go @@ -0,0 +1,24 @@ +package kglib + +import "testing" + +func TestEmbeddingFromRaw(t *testing.T) { + if emb, ok := embeddingFromRaw([]any{float32(1.5), float32(-2)}); !ok || emb[0] != 1.5 || emb[1] != -2 { + t.Errorf("float32 components: emb=%v ok=%v, want [1.5 -2] true", emb, ok) + } + // float64 components must convert, not zero — a DOUBLE column or driver + // change silently corrupting distances is the bug this guards. + if emb, ok := embeddingFromRaw([]any{float64(1.5), float64(-2)}); !ok || emb[0] != 1.5 || emb[1] != -2 { + t.Errorf("float64 components: emb=%v ok=%v, want [1.5 -2] true", emb, ok) + } + if emb, ok := embeddingFromRaw([]any{float32(1), float64(2)}); !ok || emb[1] != 2 { + t.Errorf("mixed components: emb=%v ok=%v, want [1 2] true", emb, ok) + } + // An unhandled type must reject the vector, never zero the component. + if _, ok := embeddingFromRaw([]any{float32(1), "not a number"}); ok { + t.Error("string component accepted; must return ok=false") + } + if _, ok := embeddingFromRaw([]any{int64(3)}); ok { + t.Error("int64 component accepted; must return ok=false") + } +} From e91fa75399847789b21b49d7412403bb7f7f612b Mon Sep 17 00:00:00 2001 From: Bryan Woodruff Date: Thu, 27 Aug 2026 11:05:20 -0700 Subject: [PATCH 2/5] =?UTF-8?q?fix(kg):=20address=20PR=20review=20?= =?UTF-8?q?=E2=80=94=20scope-fallback=20regression,=20current-symlink=20re?= =?UTF-8?q?pair,=20drop=20visibility?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review of #4 found the scope-parity change regressed a case it did not intend. resolveScopeDB could not tell a scope named with --scope from one inherited out of config.json, so a stale defaultScope naming nothing — reachable, since SetDefaultScope does not verify the scope exists and config.json travels with a repo — turned `kg stats` from a legacy-database report into an error. The resolver now takes the requested scope separately: a NAMED scope that cannot load is still an error, while an inherited default falls back to the legacy database when the project has no scope configs at all. Two more from the same review: - The hub's reconcile pass collapsed "readlink failed" into "nothing to do", so a `current` symlink missing entirely — a partial copy, a restored backup, an operator — was left broken with every search 500ing, strictly worse than the split reconcile exists to fix. Missing links are now recreated from the registry; other readlink errors are logged, not silently skipped, as are registry entries failing path validation. - Dropping an embedding vector with unhandled component types is correct but as invisible as the zeroing it replaced, so the build logs the count once. New tests, each mutation-verified: kg stats routed through its own RunE (a test of the shared helper alone cannot show the caller still uses it) covering both the named-scope error and the inherited-default fallback; a missing-current reconcile case; humanAge's thresholds; and the report's age line asserted for order and real values rather than its label. Also from review: the median query builds its SKIP with strconv rather than embedding a shared constant in a format string, and the reconcile comment records the same-commit re-push residual it cannot detect. Co-Authored-By: Claude Fable 5 --- src/kg/health.go | 41 +++++++++++--- src/kg/health_test.go | 82 +++++++++++++++++++++++++++ src/kg/internal/hub/install_test.go | 32 +++++++++++ src/kg/internal/hub/server.go | 38 +++++++++++-- src/kg/internal/knowledge/health.go | 10 +++- src/kg/stats.go | 6 +- src/kg/stats_test.go | 87 +++++++++++++++++++++++++++++ src/kglib/hnsw_index.go | 15 ++++- 8 files changed, 290 insertions(+), 21 deletions(-) create mode 100644 src/kg/stats_test.go diff --git a/src/kg/health.go b/src/kg/health.go index 0431587..1c137d4 100644 --- a/src/kg/health.go +++ b/src/kg/health.go @@ -122,14 +122,29 @@ func runHealth(root, scopeName string, jsonOut bool, out io.Writer) error { return nil } -// resolveScopeDB resolves the database path for read-only report commands -// (kg health, kg stats): explicit scope, else the default scope, else the -// legacy knowledge.db. Returns the path and the scope name actually used -// ("" for legacy). -// A named scope that cannot be loaded is an error, never a silent fallback — -// reporting the legacy database's health under a scope the user asked for -// would be a wrong answer, not a degraded one. -func resolveScopeDB(aiDir, scopeName string) (string, string, error) { +// resolveScopeDB resolves the database path for the read-only report commands +// (kg health, kg stats): the scope the user named, else the configured default +// scope, else the legacy knowledge.db. Returns the path and the scope name +// actually used ("" for legacy). +// +// The two sources of a scope name are held to different standards, which is +// why requested is separate from the default rather than pre-resolved by the +// caller: +// +// - A scope the user NAMED that cannot be loaded is always an error. +// Reporting the legacy database under a scope someone asked for by name is +// a wrong answer, not a degraded one. +// - A default scope inherited from config.json falls back to the legacy +// database when the project has no scope configs at all. config.json +// travels with a repo and SetDefaultScope does not verify the scope +// exists, so a stale defaultScope naming nothing is a configuration +// leftover, not a request — failing there would break `kg stats` in +// repositories where it used to work. +// +// A default scope in a project that DOES have scope configs still errors: the +// scopes are real, so a default naming a missing one is a genuine mistake. +func resolveScopeDB(aiDir, requested string) (string, string, error) { + scopeName := requested if scopeName == "" { defaultScope, err := knowledge.GetDefaultScope(aiDir) if err != nil { @@ -141,6 +156,16 @@ func resolveScopeDB(aiDir, scopeName string) (string, string, error) { return filepath.Join(aiDir, "knowledge.db"), "", nil } + if requested == "" { + configs, err := knowledge.ListScopeConfigs(aiDir) + if err != nil { + return "", "", err + } + if len(configs) == 0 { + return filepath.Join(aiDir, "knowledge.db"), "", nil + } + } + cfg, err := knowledge.LoadScopeConfig(aiDir, scopeName) if err != nil { return "", "", fmt.Errorf("load scope %q: %w", scopeName, err) diff --git a/src/kg/health_test.go b/src/kg/health_test.go index b3b8c20..a57c41b 100644 --- a/src/kg/health_test.go +++ b/src/kg/health_test.go @@ -3,6 +3,7 @@ package main import ( "bytes" "encoding/json" + "fmt" "os" "path/filepath" "strings" @@ -247,3 +248,84 @@ func TestResolveScopeDBErrorsOnUnloadableScope(t *testing.T) { t.Errorf("legacy resolution = (%q, %q, %v), want knowledge.db path, empty scope, nil", dbPath, scopeName, err) } } + +// humanAge's thresholds, which the report's only age line depends on. The +// human-output test can pass on an empty or misordered rendering, so the +// boundaries are pinned here directly. +func TestHumanAge(t *testing.T) { + cases := []struct { + d time.Duration + want string + }{ + {30 * time.Second, "just now"}, + {time.Minute, "1m"}, + {59 * time.Minute, "59m"}, + {time.Hour, "1h"}, + {23 * time.Hour, "23h"}, + {24 * time.Hour, "1d"}, + {29 * 24 * time.Hour, "29d"}, + {30 * 24 * time.Hour, "1mo"}, + {364 * 24 * time.Hour, "12mo"}, + {365 * 24 * time.Hour, "1y"}, + {800 * 24 * time.Hour, "2y"}, + } + for _, tc := range cases { + if got := humanAge(tc.d); got != tc.want { + t.Errorf("humanAge(%v) = %q, want %q", tc.d, got, tc.want) + } + } +} + +// The age line names newest, median, and oldest in that order with real +// values — an assertion on the label alone passes even if humanAge returns +// empty strings or the three are rendered in the wrong order. +func TestHealthHumanOutputAgeLine(t *testing.T) { + if testing.Short() { + t.Skip("Skipping integration test in short mode") + } + + root := t.TempDir() + seedHealthFixture(t, root) + + var buf bytes.Buffer + if err := runHealth(root, "", false, &buf); err != nil { + t.Fatalf("runHealth: %v", err) + } + // The fixture's timestamped rows are [2020, 2022, now, now], so the line + // must read newest "just now", median ~4y, oldest ~6y — ordered oldest + // last. Years drift with the wall clock, so match the shape and the + // ordering rather than exact ages. + line := "" + for _, l := range strings.Split(buf.String(), "\n") { + if strings.HasPrefix(l, "Observation age:") { + line = l + break + } + } + if line == "" { + t.Fatalf("no observation-age line in report:\n%s", buf.String()) + } + newestIdx := strings.Index(line, "newest just now") + medianIdx := strings.Index(line, "median ") + oldestIdx := strings.Index(line, "oldest ") + if newestIdx < 0 || medianIdx < 0 || oldestIdx < 0 { + t.Fatalf("age line missing a labelled value: %q", line) + } + if !(newestIdx < medianIdx && medianIdx < oldestIdx) { + t.Errorf("age line out of order (newest, median, oldest): %q", line) + } + if strings.Contains(line, "median y") || strings.Contains(line, "median ,") { + t.Errorf("age line has an empty median value: %q", line) + } + // Median (2022) must read as a smaller age than oldest (2020). + var medianYears, oldestYears int + if _, err := fmt.Sscanf(line[medianIdx:], "median %dy", &medianYears); err != nil { + t.Fatalf("median value is not a year age in %q: %v", line, err) + } + if _, err := fmt.Sscanf(line[oldestIdx:], "oldest %dy", &oldestYears); err != nil { + t.Fatalf("oldest value is not a year age in %q: %v", line, err) + } + if medianYears >= oldestYears { + t.Errorf("median age %dy is not younger than oldest %dy: %q", medianYears, oldestYears, line) + } +} diff --git a/src/kg/internal/hub/install_test.go b/src/kg/internal/hub/install_test.go index fca6fb3..2f18022 100644 --- a/src/kg/internal/hub/install_test.go +++ b/src/kg/internal/hub/install_test.go @@ -193,3 +193,35 @@ func TestReconcileRollsBackInterruptedSeed(t *testing.T) { t.Errorf("served commit = %q, want %q", gotCommit, commit) } } + +// A `current` symlink that is missing entirely — a partial copy, a restored +// backup, an operator — is state the hub did not create, and leaving it +// unrepaired is strictly worse than the split reconcile exists to fix: every +// search 500s, permanently. The registry knows the answer, so construction +// recreates the link. +func TestReconcileRestoresMissingCurrent(t *testing.T) { + dataDir := t.TempDir() + ts := httptest.NewServer(NewServer(dataDir, "", "s3cret", "dev").Handler()) + + commit := strings.Repeat("c", 40) + dbPath := buildFixtureDBFor(t, "MissingCurrentEntity", "proj-mc", commit) + if err := pushFixtureFor(t, ts.URL, "mc", dbPath, commit, "proj-mc"); err != nil { + t.Fatalf("push fixture: %v", err) + } + ts.Close() + + gdir := filepath.Join(dataDir, "graphs", "mc") + if err := os.Remove(filepath.Join(gdir, "current")); err != nil { + t.Fatalf("remove current: %v", err) + } + + ts2 := httptest.NewServer(NewServer(dataDir, "", "s3cret", "dev").Handler()) + defer ts2.Close() + + if target, err := os.Readlink(filepath.Join(gdir, "current")); err != nil || target != commit { + t.Fatalf("current -> %q (err %v), want restored to %q", target, err, commit) + } + if names, _ := searchNames(t, ts2.URL, "mc", "MissingCurrentEntity"); len(names) == 0 { + t.Error("graph does not answer after restoring a missing current symlink") + } +} diff --git a/src/kg/internal/hub/server.go b/src/kg/internal/hub/server.go index 21b212f..89fd00d 100644 --- a/src/kg/internal/hub/server.go +++ b/src/kg/internal/hub/server.go @@ -115,6 +115,16 @@ func (s *Server) sweepStaging() { // orphaned commit directory is deliberately left in place — the next // successful install's prune removes it, and deleting data is not a job for // a constructor-time repair pass. +// +// One residual this cannot see: a kill while re-pushing the commit that is +// ALREADY installed. install stashes the existing directory as .old-*, moves +// the new database in, and dies before the registry write; sweepStaging then +// deletes the stash, and `current` still names the registered commit, so +// nothing here looks wrong. The graph serves the new database under the old +// GraphInfo. It needs a same-commit re-push (a dirty push) whose ProjectID +// also changed, and closing it means stamping the installed database's +// identity somewhere this pass can compare — not worth the machinery until +// the failure is observed. func (s *Server) reconcileInstalls() { reg, err := loadRegistry(s.dataDir) if err != nil { @@ -125,23 +135,39 @@ func (s *Server) reconcileInstalls() { // registry.json is hand-editable: validate everything joined into a // path, as everywhere else. if !validPathComponent(name) || !validPathComponent(info.Commit) { + log.Printf("reconcile: registry entry %q records invalid name or commit %q — skipping (the graph will not be served)", name, info.Commit) continue } gdir := s.graphDir(name) currentLink := filepath.Join(gdir, "current") + target, err := os.Readlink(currentLink) - if err != nil || target == info.Commit { + switch { + case err == nil && target == info.Commit: + continue + case err != nil && !os.IsNotExist(err): + // Something is there but unreadable as a symlink; repairing it + // blind could destroy data this pass does not understand. + log.Printf("reconcile %s: read current: %v — leaving as is", name, err) continue } - if _, err := os.Stat(filepath.Join(gdir, info.Commit)); err != nil { - log.Printf("reconcile %s: current -> %s but registry records %s, whose directory is missing — leaving as is", name, target, info.Commit) + // From here: current is missing (a partial copy, a restored backup, an + // operator) or points somewhere the registry does not record. Both are + // repaired from the registry, which is install's commit point. + + if _, serr := os.Stat(filepath.Join(gdir, info.Commit)); serr != nil { + log.Printf("reconcile %s: registry records %s but its directory is missing — leaving as is", name, info.Commit) continue } - if err := replaceSymlink(gdir, currentLink, info.Commit); err != nil { - log.Printf("reconcile %s: repoint current %s -> %s: %v", name, target, info.Commit, err) + if rerr := replaceSymlink(gdir, currentLink, info.Commit); rerr != nil { + log.Printf("reconcile %s: point current at %s: %v", name, info.Commit, rerr) continue } - log.Printf("reconcile %s: current pointed at unregistered commit %s (interrupted seed); rolled back to registered %s", name, target, info.Commit) + if os.IsNotExist(err) { + log.Printf("reconcile %s: current was missing; pointed it at the registered commit %s", name, info.Commit) + } else { + log.Printf("reconcile %s: current pointed at unregistered commit %s (interrupted seed); rolled back to registered %s", name, target, info.Commit) + } } } diff --git a/src/kg/internal/knowledge/health.go b/src/kg/internal/knowledge/health.go index 0f530c9..588365e 100644 --- a/src/kg/internal/knowledge/health.go +++ b/src/kg/internal/knowledge/health.go @@ -2,6 +2,7 @@ package knowledge import ( "fmt" + "strconv" "time" ) @@ -182,13 +183,16 @@ func collectObservationAge(store *Store, projectID string, timestamped int) (*Ob // Median by position: the middle row (lower of the two for even counts) // of the timestamped observations in created_at order. - result, err = store.QueryParams(fmt.Sprintf(` + // Plain concatenation, not Sprintf: the format string would embed + // timestampedObsFilter, so a '%' ever appearing in that shared constant + // would silently corrupt this query. + result, err = store.QueryParams(` MATCH (e:Entity {project_id: $project_id})-[:HAS_OBSERVATION]->(o:Observation) WHERE `+timestampedObsFilter+` RETURN o.created_at ORDER BY o.created_at - SKIP %d LIMIT 1 - `, (timestamped-1)/2), map[string]any{"project_id": projectID}) + SKIP `+strconv.Itoa((timestamped-1)/2)+` LIMIT 1 + `, map[string]any{"project_id": projectID}) if err != nil { return nil, err } diff --git a/src/kg/stats.go b/src/kg/stats.go index 233248e..1fc875f 100644 --- a/src/kg/stats.go +++ b/src/kg/stats.go @@ -36,9 +36,9 @@ By default, shows stats for the default scope. Use --scope to specify a differen aiDir := filepath.Join(root, ".ai") projectID := projectIDFromCwd(cwd) - // Determine which scope to use - // Shared with kg health: an explicitly named scope that cannot be - // loaded is an error, never a silent fallback to the legacy database. + // Shared with kg health: a scope named with --scope that cannot be + // loaded is an error rather than a silent fallback to the legacy + // database; an inherited default scope still falls back (resolveScopeDB). dbPath, scopeName, err := resolveScopeDB(aiDir, statsScopeName) if err != nil { return err diff --git a/src/kg/stats_test.go b/src/kg/stats_test.go new file mode 100644 index 0000000..14bf7b7 --- /dev/null +++ b/src/kg/stats_test.go @@ -0,0 +1,87 @@ +package main + +import ( + "encoding/json" + "os" + "path/filepath" + "strings" + "testing" + + "github.com/cortexa-llc/mcp/kg/internal/knowledge" +) + +// statsFixture makes a project rooted at a temp dir with a legacy +// knowledge.db, chdirs into it, and returns the root. The chdir is what lets +// statsCmd's RunE — which resolves everything from the working directory — +// be exercised directly. +func statsFixture(t *testing.T) string { + t.Helper() + root := t.TempDir() + aiDir := filepath.Join(root, ".ai") + if err := os.MkdirAll(aiDir, 0o755); err != nil { + t.Fatalf("MkdirAll: %v", err) + } + store, err := knowledge.OpenStore(filepath.Join(aiDir, "knowledge.db")) + if err != nil { + t.Fatalf("OpenStore: %v", err) + } + if _, err := store.CreateEntity("stats entity", "topic", projectIDFromCwd(root)); err != nil { + t.Fatalf("CreateEntity: %v", err) + } + store.Close() + t.Chdir(root) + return root +} + +// runStats invokes the command's own RunE, so the test breaks if stats stops +// routing through resolveScopeDB — a test against the helper alone cannot +// tell that the caller still uses it. +func runStats(t *testing.T, scope string) error { + t.Helper() + prev := statsScopeName + statsScopeName = scope + t.Cleanup(func() { statsScopeName = prev }) + return statsCmd.RunE(statsCmd, nil) +} + +// A scope named on the command line that cannot be loaded must fail, in stats +// exactly as in health — the parity this change is about. +func TestStatsErrorsOnUnloadableNamedScope(t *testing.T) { + if testing.Short() { + t.Skip("Skipping integration test in short mode") + } + statsFixture(t) + + err := runStats(t, "no-such-scope") + if err == nil { + t.Fatal("kg stats --scope no-such-scope succeeded, want an error rather than a silent legacy fallback") + } + if !strings.Contains(err.Error(), "no-such-scope") { + t.Errorf("error %q does not name the scope that failed", err) + } +} + +// The regression guard for the fallback path: config.json can carry a +// defaultScope naming a scope that does not exist (SetDefaultScope does not +// verify, and config.json travels with a repo). That is a configuration +// leftover, not a request, so stats must still report the legacy database +// instead of failing — which is how it behaved before the shared resolver. +func TestStatsFallsBackWhenInheritedDefaultScopeHasNoConfigs(t *testing.T) { + if testing.Short() { + t.Skip("Skipping integration test in short mode") + } + root := statsFixture(t) + + cfg := map[string]string{"defaultScope": "team"} + data, err := json.Marshal(cfg) + if err != nil { + t.Fatalf("marshal config: %v", err) + } + if err := os.WriteFile(filepath.Join(root, ".ai", "config.json"), data, 0o644); err != nil { + t.Fatalf("write config.json: %v", err) + } + + if err := runStats(t, ""); err != nil { + t.Errorf("kg stats with a stale defaultScope and no scope configs: %v; want the legacy database", err) + } +} diff --git a/src/kglib/hnsw_index.go b/src/kglib/hnsw_index.go index f47d6fb..d27108c 100644 --- a/src/kglib/hnsw_index.go +++ b/src/kglib/hnsw_index.go @@ -2,6 +2,7 @@ package kglib import ( "fmt" + "log" "sync" "time" @@ -84,6 +85,7 @@ func (s *Store) buildIndex(projectID string) (*projectIndex, error) { entities := make(map[string]*Entity) nodes := make([]hnsw.Node[string], 0, 256) + dropped := 0 for result.HasNext() { tuple, err := result.Next() @@ -114,7 +116,10 @@ func (s *Store) buildIndex(projectID string) (*projectIndex, error) { if !ok { // A component of unhandled type must exclude the whole node: the // old behaviour left it at 0, silently distorting every distance - // this vector participated in. + // this vector participated in. Excluding it is correct but just as + // invisible from the outside — the entity simply stops appearing + // in vector results — so the count is reported below. + dropped++ continue } @@ -122,6 +127,14 @@ func (s *Store) buildIndex(projectID string) (*projectIndex, error) { nodes = append(nodes, hnsw.MakeNode(entity.ID, emb)) } + if dropped > 0 { + // Not an error — the index is still usable — but silently reduced + // recall is exactly the kind of thing nobody discovers from search + // results, so say it once per build. + log.Printf("index build (project %s): dropped %d of %d embedded entities whose vectors held unhandled component types", + projectID, dropped, dropped+len(nodes)) + } + if len(nodes) > 0 { g.Add(nodes...) } From 03cb73dc26283ee73e5f9e31c3000a5dbe0bbf85 Mon Sep 17 00:00:00 2001 From: Bryan Woodruff Date: Thu, 27 Aug 2026 12:55:51 -0700 Subject: [PATCH 3/5] fix(kg): a null registry entry no longer kills the hub at startup MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Re-review of #4 caught a severity escalation the last commit introduced. reconcileInstalls validates registry entries before joining them into paths, but dereferenced info.Commit to do it — and registry.json is hand-editable, where `{"graphs":{"g":null}}` unmarshals to a nil entry. Because the pass runs in the constructor, that panic aborted `kg hub serve` outright; before, the same entry only panicked inside a request, which net/http recovers per connection, so the hub kept serving every other graph. Nil entries are now skipped and logged. Also from the same review: - kg health no longer hard-fails when the observation count shifts mid-report. The median's SKIP offset came from counts taken by two earlier queries, so a concurrent kg index deleting rows made the offset overshoot and failed the entire run; collectObservationAge now takes its own count over the same predicate as the median query, and a vanished row set reports no age rather than an error. - reconcile logs the one state it cannot repair: current naming the registered commit whose directory is gone (searches fail, and the registry has no other answer). - printStats writes to an io.Writer like runHealth does, so the stats tests assert what was read — the previous fallback test passed even with stats bypassing the shared resolver entirely, which the reviewer demonstrated. - humanAge names negative durations instead of calling them "just now", and its year boundary is 12 30-day months so the ladder no longer prints "12mo" before "1y". - Fixture comments corrected to the six observations they now create. New tests for the null entry and the negative duration; the nil guard is mutation-verified (removing it panics the suite). Left as filed follow-ups, both from the reviewer: buildIndex's drop-on-reject branch stays uncovered until the loop body is extracted into a pure function, and the health/stats projectIDFromCwd divergence is provably cosmetic (findProjectRoot is idempotent) rather than a bug. Co-Authored-By: Claude Fable 5 --- src/kg/health.go | 21 +++++--- src/kg/health_test.go | 31 +++++++---- src/kg/internal/hub/install_test.go | 45 ++++++++++++++++ src/kg/internal/hub/server.go | 24 +++++++-- src/kg/internal/knowledge/health.go | 28 ++++++++-- src/kg/stats.go | 79 ++++++++++++++++------------- src/kg/stats_test.go | 35 +++++++++---- src/kglib/hnsw_index.go | 2 +- 8 files changed, 195 insertions(+), 70 deletions(-) diff --git a/src/kg/health.go b/src/kg/health.go index 1c137d4..85feb6e 100644 --- a/src/kg/health.go +++ b/src/kg/health.go @@ -266,19 +266,28 @@ func printHealth(out io.Writer, o healthOutput, snapPath string) { // humanAge renders a duration as a coarse age ("3h", "2d", "5mo") for the // observation-age line; precision beyond this is noise in a health report. func humanAge(d time.Duration) string { + const ( + day = 24 * time.Hour + month = 30 * day + year = 12 * month // 360d, so the ladder never prints "12mo" before "1y" + ) switch { + case d < 0: + // A stored timestamp ahead of the report's own clock: skew between + // machines pushing to a shared hub graph, or a hand-set created_at. + return "in the future" case d < time.Minute: return "just now" case d < time.Hour: return fmt.Sprintf("%dm", int(d.Minutes())) - case d < 24*time.Hour: + case d < day: return fmt.Sprintf("%dh", int(d.Hours())) - case d < 30*24*time.Hour: - return fmt.Sprintf("%dd", int(d.Hours()/24)) - case d < 365*24*time.Hour: - return fmt.Sprintf("%dmo", int(d.Hours()/(24*30))) + case d < month: + return fmt.Sprintf("%dd", int(d/day)) + case d < year: + return fmt.Sprintf("%dmo", int(d/month)) default: - return fmt.Sprintf("%dy", int(d.Hours()/(24*365))) + return fmt.Sprintf("%dy", int(d/year)) } } diff --git a/src/kg/health_test.go b/src/kg/health_test.go index a57c41b..0dce036 100644 --- a/src/kg/health_test.go +++ b/src/kg/health_test.go @@ -14,9 +14,11 @@ import ( ) // seedHealthFixture creates /.ai/knowledge.db with a small graph: -// three entities (one orphaned), two observations (one [OBSOLETE-marked), -// and one relation. The store is closed before returning so runHealth can -// open the database read-only. +// three entities (one orphaned), one relation, and six observations — two +// written normally (one [OBSOLETE-marked), two aged into legacy state (NULL +// and stored-zero created_at), and two back-dated to 2020 and 2022 to pin the +// age stats. The store is closed before returning so runHealth can open the +// database read-only. func seedHealthFixture(t *testing.T, root string) { t.Helper() @@ -54,11 +56,11 @@ func seedHealthFixture(t *testing.T, root string) { t.Fatalf("CreateEntity: %v", err) } - // Two observations with genuinely legacy STORED timestamps — one NULL, one - // stored zero — written the only way this writer can produce them: by - // mutating created_at after the fact. This is what the zero-timestamp - // metric exists to count; a fixture whose every row carries a real - // timestamp cannot fail a broken counter. The stored-zero row in + // Four observations with STORED timestamps this writer cannot produce + // directly — two legacy (NULL and stored zero), two back-dated — written + // the only way it can: by mutating created_at after the fact. The legacy + // pair is what the zero-timestamp metric exists to count; a fixture whose + // every row carries a real timestamp cannot fail a broken counter. The stored-zero row in // particular guards against binding the zero time as a Go parameter, // which go-kuzu mangles through UnixNano into a 1754 date that matches // nothing. @@ -122,8 +124,8 @@ func TestHealthCommandReportsMetricsAndGrowth(t *testing.T) { if out.Current.Observations != 6 { t.Errorf("observations = %d, want 6", out.Current.Observations) } - // Exactly the two deliberately aged rows — not 0 (metric dead) and not 4 - // (metric counting scan artifacts instead of stored values). + // Exactly the two legacy rows — not 0 (metric dead) and not 6 (metric + // counting scan artifacts instead of stored values). if out.Current.ZeroTimestampObservations != 2 { t.Errorf("zero-timestamp observations = %d, want 2 (one NULL + one stored zero)", out.Current.ZeroTimestampObservations) @@ -265,9 +267,16 @@ func TestHumanAge(t *testing.T) { {24 * time.Hour, "1d"}, {29 * 24 * time.Hour, "29d"}, {30 * 24 * time.Hour, "1mo"}, - {364 * 24 * time.Hour, "12mo"}, + // The month/year boundary is 12 months of 30 days, so the ladder goes + // 11mo -> 1y with no "12mo" step. + {359 * 24 * time.Hour, "11mo"}, + {360 * 24 * time.Hour, "1y"}, + {364 * 24 * time.Hour, "1y"}, {365 * 24 * time.Hour, "1y"}, {800 * 24 * time.Hour, "2y"}, + // A stored timestamp ahead of the report's clock (machine skew, or a + // hand-set created_at) must be named, not rendered as "just now". + {-time.Hour, "in the future"}, } for _, tc := range cases { if got := humanAge(tc.d); got != tc.want { diff --git a/src/kg/internal/hub/install_test.go b/src/kg/internal/hub/install_test.go index 2f18022..ba6b8a6 100644 --- a/src/kg/internal/hub/install_test.go +++ b/src/kg/internal/hub/install_test.go @@ -225,3 +225,48 @@ func TestReconcileRestoresMissingCurrent(t *testing.T) { t.Error("graph does not answer after restoring a missing current symlink") } } + +// registry.json is hand-editable, and a null entry unmarshals to a nil +// *GraphInfo. reconcileInstalls runs in the constructor, so dereferencing it +// there does not fail one request — it aborts `kg hub serve` before any graph +// is served. The bad entry must be skipped, and every other graph must still +// answer. +func TestReconcileSurvivesNullRegistryEntry(t *testing.T) { + dataDir := t.TempDir() + ts := httptest.NewServer(NewServer(dataDir, "", "s3cret", "dev").Handler()) + + commit := strings.Repeat("d", 40) + dbPath := buildFixtureDBFor(t, "SurvivorEntity", "proj-null", commit) + if err := pushFixtureFor(t, ts.URL, "survivor", dbPath, commit, "proj-null"); err != nil { + t.Fatalf("push fixture: %v", err) + } + ts.Close() + + // Hand-edit the registry to hold a null entry alongside the good one. + regPath := filepath.Join(dataDir, registryFile) + raw, err := os.ReadFile(regPath) + if err != nil { + t.Fatalf("read registry: %v", err) + } + var reg map[string]map[string]any + if err := json.Unmarshal(raw, ®); err != nil { + t.Fatalf("parse registry: %v", err) + } + reg["graphs"]["broken"] = nil + edited, err := json.Marshal(reg) + if err != nil { + t.Fatalf("marshal registry: %v", err) + } + if err := os.WriteFile(regPath, edited, 0o644); err != nil { + t.Fatalf("write registry: %v", err) + } + + // Construction must not panic... + ts2 := httptest.NewServer(NewServer(dataDir, "", "s3cret", "dev").Handler()) + defer ts2.Close() + + // ...and the healthy graph must still be served. + if names, _ := searchNames(t, ts2.URL, "survivor", "SurvivorEntity"); len(names) == 0 { + t.Error("healthy graph stopped answering after a null registry entry was introduced") + } +} diff --git a/src/kg/internal/hub/server.go b/src/kg/internal/hub/server.go index 89fd00d..b79756c 100644 --- a/src/kg/internal/hub/server.go +++ b/src/kg/internal/hub/server.go @@ -132,8 +132,15 @@ func (s *Server) reconcileInstalls() { return } for name, info := range reg.Graphs { - // registry.json is hand-editable: validate everything joined into a - // path, as everywhere else. + // registry.json is hand-editable, so this runs before the pointer is + // dereferenced: `{"graphs":{"g":null}}` unmarshals to a nil entry, and + // this pass runs in the constructor — a panic here aborts `kg hub + // serve` outright rather than being contained to one request. + if info == nil { + log.Printf("reconcile: registry entry %q is null — skipping (the graph will not be served)", name) + continue + } + // Validate everything joined into a path, as everywhere else. if !validPathComponent(name) || !validPathComponent(info.Commit) { log.Printf("reconcile: registry entry %q records invalid name or commit %q — skipping (the graph will not be served)", name, info.Commit) continue @@ -141,9 +148,20 @@ func (s *Server) reconcileInstalls() { gdir := s.graphDir(name) currentLink := filepath.Join(gdir, "current") + // The registered commit's directory is what any repair points at, and + // its absence is worth reporting even when there is nothing to repair: + // a `current` that already names it still fails every search. + registeredExists := true + if _, serr := os.Stat(filepath.Join(gdir, info.Commit)); serr != nil { + registeredExists = false + } + target, err := os.Readlink(currentLink) switch { case err == nil && target == info.Commit: + if !registeredExists { + log.Printf("reconcile %s: current names the registered commit %s but its directory is missing — searches will fail and reconcile cannot repair it (restore the directory or re-push)", name, info.Commit) + } continue case err != nil && !os.IsNotExist(err): // Something is there but unreadable as a symlink; repairing it @@ -155,7 +173,7 @@ func (s *Server) reconcileInstalls() { // operator) or points somewhere the registry does not record. Both are // repaired from the registry, which is install's commit point. - if _, serr := os.Stat(filepath.Join(gdir, info.Commit)); serr != nil { + if !registeredExists { log.Printf("reconcile %s: registry records %s but its directory is missing — leaving as is", name, info.Commit) continue } diff --git a/src/kg/internal/knowledge/health.go b/src/kg/internal/knowledge/health.go index 588365e..e8241f5 100644 --- a/src/kg/internal/knowledge/health.go +++ b/src/kg/internal/knowledge/health.go @@ -107,7 +107,7 @@ func CollectHealthMetrics(store *Store, projectID string) (*HealthMetrics, error return nil, fmt.Errorf("count zero-timestamp observations: %w", err) } - if m.ObservationAge, err = collectObservationAge(store, projectID, m.Observations-m.ZeroTimestampObservations); err != nil { + if m.ObservationAge, err = collectObservationAge(store, projectID); err != nil { return nil, fmt.Errorf("collect observation age: %w", err) } @@ -148,9 +148,24 @@ type ObservationAge struct { const timestampedObsFilter = `o.created_at IS NOT NULL AND o.created_at <> timestamp("0001-01-01 00:00:00")` // collectObservationAge computes newest/oldest/median stored created_at over -// the project's timestamped observations. timestamped is their count, already -// known to the caller (total minus zero-timestamp); 0 yields nil. -func collectObservationAge(store *Store, projectID string, timestamped int) (*ObservationAge, error) { +// the project's timestamped observations. Nil when there are none. +// +// The count driving the median's offset is taken here, over the same +// predicate as the median query itself, rather than derived by the caller +// from the total and zero-timestamp counts. Those agree in a quiet database, +// but they are separate queries on a read-only connection with no snapshot +// across them: a concurrent `kg index` deleting rows in between would make +// the offset overshoot the result set, and an off-by-one there silently +// shifts the median. One predicate, one count, one query. +func collectObservationAge(store *Store, projectID string) (*ObservationAge, error) { + timestamped, err := countQuery(store, ` + MATCH (e:Entity {project_id: $project_id})-[:HAS_OBSERVATION]->(o:Observation) + WHERE `+timestampedObsFilter+` + RETURN count(*) + `, map[string]any{"project_id": projectID}) + if err != nil { + return nil, err + } if timestamped <= 0 { return nil, nil } @@ -198,7 +213,10 @@ func collectObservationAge(store *Store, projectID string, timestamped int) (*Ob } defer result.Close() if !result.HasNext() { - return nil, fmt.Errorf("median query returned no row for %d timestamped observations", timestamped) + // A writer deleted rows between the count and this query. The age + // stats are one line of a report — losing them is not worth failing + // the whole run over, so report no age rather than an error. + return nil, nil } row, err := result.Next() if err != nil { diff --git a/src/kg/stats.go b/src/kg/stats.go index 1fc875f..96b9e62 100644 --- a/src/kg/stats.go +++ b/src/kg/stats.go @@ -2,6 +2,7 @@ package main import ( "fmt" + "io" "os" "path/filepath" @@ -18,48 +19,56 @@ var statsCmd = &cobra.Command{ By default, shows stats for the default scope. Use --scope to specify a different scope.`, RunE: func(cmd *cobra.Command, args []string) error { - if usePersonal { - store, projectID, err := openPersonalStore(true) - if err != nil { - return err - } - defer store.Close() - fmt.Println("Stats for the personal knowledge store") - return printStats(store, projectID) - } + return runStatsTo(os.Stdout) + }, +} - cwd, err := os.Getwd() +// runStatsTo writes the stats report to out. Split from RunE and given a +// writer for the same reason runHealth has one: a test that cannot read the +// output can only assert that the command did not error, which passes even if +// the command reads the wrong database entirely. +func runStatsTo(out io.Writer) error { + if usePersonal { + store, projectID, err := openPersonalStore(true) if err != nil { return err } - root := findProjectRoot(cwd) - aiDir := filepath.Join(root, ".ai") - projectID := projectIDFromCwd(cwd) + defer store.Close() + fmt.Fprintln(out, "Stats for the personal knowledge store") + return printStats(out, store, projectID) + } - // Shared with kg health: a scope named with --scope that cannot be - // loaded is an error rather than a silent fallback to the legacy - // database; an inherited default scope still falls back (resolveScopeDB). - dbPath, scopeName, err := resolveScopeDB(aiDir, statsScopeName) - if err != nil { - return err - } - if scopeName != "" { - fmt.Printf("Stats for scope: %s\n", scopeName) - } + cwd, err := os.Getwd() + if err != nil { + return err + } + root := findProjectRoot(cwd) + aiDir := filepath.Join(root, ".ai") + projectID := projectIDFromCwd(cwd) - store, err := knowledge.OpenStoreReadOnly(dbPath) - if err != nil { - return err - } - defer store.Close() + // Shared with kg health: a scope named with --scope that cannot be + // loaded is an error rather than a silent fallback to the legacy + // database; an inherited default scope still falls back (resolveScopeDB). + dbPath, scopeName, err := resolveScopeDB(aiDir, statsScopeName) + if err != nil { + return err + } + if scopeName != "" { + fmt.Fprintf(out, "Stats for scope: %s\n", scopeName) + } - return printStats(store, projectID) - }, + store, err := knowledge.OpenStoreReadOnly(dbPath) + if err != nil { + return err + } + defer store.Close() + + return printStats(out, store, projectID) } -// printStats prints entity, relation, and observation counts for one store. +// printStats writes entity, relation, and observation counts for one store. // Uses count queries rather than iterating entities. -func printStats(store *knowledge.Store, projectID string) error { +func printStats(out io.Writer, store *knowledge.Store, projectID string) error { entityCount, err := store.CountEntities(projectID) if err != nil { return err @@ -75,9 +84,9 @@ func printStats(store *knowledge.Store, projectID string) error { return err } - fmt.Printf("Entities: %d\n", entityCount) - fmt.Printf("Relations: %d\n", relationCount) - fmt.Printf("Observations: %d\n", observationCount) + fmt.Fprintf(out, "Entities: %d\n", entityCount) + fmt.Fprintf(out, "Relations: %d\n", relationCount) + fmt.Fprintf(out, "Observations: %d\n", observationCount) return nil } diff --git a/src/kg/stats_test.go b/src/kg/stats_test.go index 14bf7b7..b362964 100644 --- a/src/kg/stats_test.go +++ b/src/kg/stats_test.go @@ -1,6 +1,7 @@ package main import ( + "bytes" "encoding/json" "os" "path/filepath" @@ -33,15 +34,18 @@ func statsFixture(t *testing.T) string { return root } -// runStats invokes the command's own RunE, so the test breaks if stats stops -// routing through resolveScopeDB — a test against the helper alone cannot -// tell that the caller still uses it. -func runStats(t *testing.T, scope string) error { +// runStats invokes the command's own code path, so the test breaks if stats +// stops routing through resolveScopeDB — a test against the helper alone +// cannot tell that the caller still uses it. The captured output is what +// makes "no error" distinguishable from "read some other database". +func runStats(t *testing.T, scope string) (string, error) { t.Helper() prev := statsScopeName statsScopeName = scope t.Cleanup(func() { statsScopeName = prev }) - return statsCmd.RunE(statsCmd, nil) + var buf bytes.Buffer + err := runStatsTo(&buf) + return buf.String(), err } // A scope named on the command line that cannot be loaded must fail, in stats @@ -52,13 +56,16 @@ func TestStatsErrorsOnUnloadableNamedScope(t *testing.T) { } statsFixture(t) - err := runStats(t, "no-such-scope") + out, err := runStats(t, "no-such-scope") if err == nil { - t.Fatal("kg stats --scope no-such-scope succeeded, want an error rather than a silent legacy fallback") + t.Fatalf("kg stats --scope no-such-scope succeeded (output %q), want an error rather than a silent legacy fallback", out) } if !strings.Contains(err.Error(), "no-such-scope") { t.Errorf("error %q does not name the scope that failed", err) } + if strings.Contains(out, "Entities:") { + t.Errorf("stats reported counts despite the scope failing to load: %q", out) + } } // The regression guard for the fallback path: config.json can carry a @@ -81,7 +88,17 @@ func TestStatsFallsBackWhenInheritedDefaultScopeHasNoConfigs(t *testing.T) { t.Fatalf("write config.json: %v", err) } - if err := runStats(t, ""); err != nil { - t.Errorf("kg stats with a stale defaultScope and no scope configs: %v; want the legacy database", err) + out, err := runStats(t, "") + if err != nil { + t.Fatalf("kg stats with a stale defaultScope and no scope configs: %v; want the legacy database", err) + } + // Assert what was read, not merely that nothing failed: the fixture's + // legacy database holds exactly one entity, so a stats run that reported + // nothing — or reported some other database — fails here. + if !strings.Contains(out, "Entities: 1") { + t.Errorf("stats output %q does not show the legacy database's single entity", out) + } + if strings.Contains(out, "Stats for scope:") { + t.Errorf("stats claimed a scope while falling back to the legacy database: %q", out) } } diff --git a/src/kglib/hnsw_index.go b/src/kglib/hnsw_index.go index d27108c..f4e5d7c 100644 --- a/src/kglib/hnsw_index.go +++ b/src/kglib/hnsw_index.go @@ -131,7 +131,7 @@ func (s *Store) buildIndex(projectID string) (*projectIndex, error) { // Not an error — the index is still usable — but silently reduced // recall is exactly the kind of thing nobody discovers from search // results, so say it once per build. - log.Printf("index build (project %s): dropped %d of %d embedded entities whose vectors held unhandled component types", + log.Printf("index build (project %s): dropped %d of %d vectors read, whose components had unhandled types", projectID, dropped, dropped+len(nodes)) } From daa97cf77191ab5711ec90aff98fca64affef873 Mon Sep 17 00:00:00 2001 From: Bryan Woodruff Date: Sat, 29 Aug 2026 15:08:50 -0700 Subject: [PATCH 4/5] fix(pr4): guard the max/min age query against the race the median already handles MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Both findings from #4's first review. collectObservationAge runs two queries after its initial count, on the same non-transactional read connection, and both are exposed to a writer deleting rows in between. The median query says so in a comment and returns (nil, nil) when the row is gone. The max/min query above it had `if result.HasNext()` with no else, so in that window it fell through with Newest and Oldest still at the zero time. printHealth renders those as humanAge(gen.Sub(oa.Newest)) — roughly two thousand years. That is the "every legacy-bearing graph reads as centuries old" symptom this metric was added to expose, so the failure would have been indistinguishable from the bug it reports on. Mirror the median's handling. Hoisting the row read out of the if-block left the median's `row, err :=` declaring nothing new, so it becomes `=` with a note saying why. Also fixed the dropped-vector log's denominator. It computed dropped+len(nodes), but rows whose embedding column is missing or empty are skipped before `dropped` is incremented, so they appeared in neither term — the "of %d vectors read" figure silently excluded them. Count reads per row instead, so the number means what it says. Co-Authored-By: Claude Opus 5 (1M context) --- src/kg/internal/knowledge/health.go | 40 ++++++++++++++++++----------- src/kglib/hnsw_index.go | 7 ++++- 2 files changed, 31 insertions(+), 16 deletions(-) diff --git a/src/kg/internal/knowledge/health.go b/src/kg/internal/knowledge/health.go index e8241f5..53740dd 100644 --- a/src/kg/internal/knowledge/health.go +++ b/src/kg/internal/knowledge/health.go @@ -179,20 +179,29 @@ func collectObservationAge(store *Store, projectID string) (*ObservationAge, err if err != nil { return nil, err } - if result.HasNext() { - row, err := result.Next() - if err != nil { - result.Close() - return nil, err - } - if age.Newest, err = timeCell(row, 0); err != nil { - result.Close() - return nil, err - } - if age.Oldest, err = timeCell(row, 1); err != nil { - result.Close() - return nil, err - } + if !result.HasNext() { + // Same race the median query below guards against, and the same + // answer: a writer deleted the rows between the count and this query. + // + // Falling through instead would leave Newest and Oldest at the zero + // time, which printHealth renders via humanAge(gen.Sub(oa.Newest)) as + // an age of roughly two thousand years — the precise "every graph + // reads as centuries old" symptom this metric was added to expose. + result.Close() + return nil, nil + } + row, err := result.Next() + if err != nil { + result.Close() + return nil, err + } + if age.Newest, err = timeCell(row, 0); err != nil { + result.Close() + return nil, err + } + if age.Oldest, err = timeCell(row, 1); err != nil { + result.Close() + return nil, err } result.Close() @@ -218,7 +227,8 @@ func collectObservationAge(store *Store, projectID string) (*ObservationAge, err // the whole run over, so report no age rather than an error. return nil, nil } - row, err := result.Next() + // `=` not `:=`: row is already declared by the max/min read above. + row, err = result.Next() if err != nil { return nil, err } diff --git a/src/kglib/hnsw_index.go b/src/kglib/hnsw_index.go index f4e5d7c..52a84e0 100644 --- a/src/kglib/hnsw_index.go +++ b/src/kglib/hnsw_index.go @@ -86,8 +86,13 @@ func (s *Store) buildIndex(projectID string) (*projectIndex, error) { entities := make(map[string]*Entity) nodes := make([]hnsw.Node[string], 0, 256) dropped := 0 + // Counted per row, not derived at the end: rows whose embedding column is + // missing or empty are skipped before `dropped` is touched, so + // dropped+len(nodes) would omit them and understate what was actually read. + read := 0 for result.HasNext() { + read++ tuple, err := result.Next() if err != nil { return nil, fmt.Errorf("index build next: %w", err) @@ -132,7 +137,7 @@ func (s *Store) buildIndex(projectID string) (*projectIndex, error) { // recall is exactly the kind of thing nobody discovers from search // results, so say it once per build. log.Printf("index build (project %s): dropped %d of %d vectors read, whose components had unhandled types", - projectID, dropped, dropped+len(nodes)) + projectID, dropped, read) } if len(nodes) > 0 { From cd8ac54fcd7d7b789a09ac35075697ff3b698d2d Mon Sep 17 00:00:00 2001 From: Bryan Woodruff Date: Sat, 29 Aug 2026 15:20:32 -0700 Subject: [PATCH 5/5] fix(pr4): NULL aggregates and null registry entries, both properly this time MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two Majors from the second review round. The first is a correction to my own previous fix. The max/min guard I added last round never fires. `RETURN max(...), min(...)` has no GROUP BY, so it is a full aggregation: over zero matched rows Kuzu returns exactly ONE row with both columns NULL, not an empty result set. Verified directly against the engine rather than reasoned about — a probe against an empty store returns 1 row with nil in both columns. So !HasNext() is not what the race looks like, and the NULLs went straight to timeCell, whose type assertion turned them into "timestamp cell has unexpected type ". That error propagates out of CollectHealthMetrics and fails the whole run — for a command whose own doc says it always exits 0, being a report and not a gate. The previous fix therefore traded a silently wrong age for a dead report in exactly the window it meant to protect. Handle the NULL case where it actually occurs, in a new ageBoundsFromRow that reports presence separately from error. A NULL means the rows went away, so it degrades to no age like the median query does; a non-NULL value of the wrong type stays a hard error, since that is a real defect and silent zeroing is the bug this metric exists to catch. timeCell now delegates to a value-level timeValue so an already-read cell can reuse it. Tested through the existing row interface seam, covering all three outcomes, and mutation-checked. Second: this PR added a nil guard to reconcileInstalls with a test proving `{"graphs":{"g":null}}` is a real hand-editable state, but every other lookup in server.go checked only `ok`. Graphs is map[string]*GraphInfo, so a null entry is a present key with a nil value — ok is true and the next dereference panics. Five request-path sites were exposed, one of them (federated search) with no check at all before reading info.Commit. Route every read through a Registry.graph helper that treats nil as absent. The test for it needed a second attempt worth recording: asserting merely "not 200" passed even with the guard removed, because net/http recovers a handler panic into a 500. Asserting 404 specifically is what makes it real — the mutation now fails with 500 and the panic's error body. Co-Authored-By: Claude Opus 5 (1M context) --- src/kg/internal/hub/install_test.go | 58 +++++++++++++++++++++ src/kg/internal/hub/registry.go | 16 ++++++ src/kg/internal/hub/server.go | 28 ++++++++--- src/kg/internal/knowledge/health.go | 62 ++++++++++++++++++----- src/kg/internal/knowledge/health_test.go | 64 ++++++++++++++++++++++++ 5 files changed, 207 insertions(+), 21 deletions(-) create mode 100644 src/kg/internal/knowledge/health_test.go diff --git a/src/kg/internal/hub/install_test.go b/src/kg/internal/hub/install_test.go index ba6b8a6..f93066f 100644 --- a/src/kg/internal/hub/install_test.go +++ b/src/kg/internal/hub/install_test.go @@ -270,3 +270,61 @@ func TestReconcileSurvivesNullRegistryEntry(t *testing.T) { t.Error("healthy graph stopped answering after a null registry entry was introduced") } } + +// The constructor-time guard above is only half the hazard. A null entry for a +// name a client actually asks for reaches the request handlers, where the +// lookups checked `ok` but not the pointer — so `{"graphs":{"broken":null}}` +// would panic on the first dereference rather than answering "unknown graph". +// +// Searching the null graph directly must fail cleanly, and searching a healthy +// graph in the same request must be unaffected. +func TestRequestHandlersSurviveNullRegistryEntry(t *testing.T) { + dataDir := t.TempDir() + ts := httptest.NewServer(NewServer(dataDir, "", "s3cret", "dev").Handler()) + + commit := strings.Repeat("e", 40) + dbPath := buildFixtureDBFor(t, "SurvivorEntity", "proj-null2", commit) + if err := pushFixtureFor(t, ts.URL, "survivor", dbPath, commit, "proj-null2"); err != nil { + t.Fatalf("push fixture: %v", err) + } + ts.Close() + + regPath := filepath.Join(dataDir, registryFile) + raw, err := os.ReadFile(regPath) + if err != nil { + t.Fatalf("read registry: %v", err) + } + var reg map[string]map[string]any + if err := json.Unmarshal(raw, ®); err != nil { + t.Fatalf("parse registry: %v", err) + } + reg["graphs"]["broken"] = nil + edited, err := json.Marshal(reg) + if err != nil { + t.Fatalf("marshal registry: %v", err) + } + if err := os.WriteFile(regPath, edited, 0o644); err != nil { + t.Fatalf("write registry: %v", err) + } + + ts2 := httptest.NewServer(NewServer(dataDir, "", "s3cret", "dev").Handler()) + defer ts2.Close() + + // Asking for the null graph by name must produce 404 "unknown graph". + // + // Asserting 404 specifically, not merely "not 200": net/http recovers a + // handler panic into a 500, so a weaker check passes whether the lookup is + // nil-safe or crashes. 404 is reachable only by the guard actually + // treating a null entry as absent. + resp, body := postJSON(t, ts2.URL+"/v1/graphs/broken/search", map[string]any{"query": "anything"}) + if resp.StatusCode != http.StatusNotFound { + t.Errorf("null graph answered %d, want %d (unknown graph). A 500 means the "+ + "handler panicked and net/http recovered it. Body: %s", + resp.StatusCode, http.StatusNotFound, strings.TrimSpace(string(body))) + } + + // The healthy graph must still answer afterwards. + if names, _ := searchNames(t, ts2.URL, "survivor", "SurvivorEntity"); len(names) == 0 { + t.Error("healthy graph stopped answering after a null entry was requested") + } +} diff --git a/src/kg/internal/hub/registry.go b/src/kg/internal/hub/registry.go index d03181e..c86d879 100644 --- a/src/kg/internal/hub/registry.go +++ b/src/kg/internal/hub/registry.go @@ -30,6 +30,22 @@ type Registry struct { Graphs map[string]*GraphInfo `json:"graphs"` } +// graph returns the entry for name, treating a nil entry as absent. +// +// registry.json is hand-editable, and `{"graphs":{"g":null}}` unmarshals to a +// present key with a nil value — so a bare `info, ok := reg.Graphs[name]` +// reports ok == true and the next dereference panics. reconcileInstalls already +// guards its own loop for exactly this reason; every other lookup goes through +// here so the same state fails as "unknown graph" instead of taking down a +// request handler. +func (r *Registry) graph(name string) (*GraphInfo, bool) { + info, ok := r.Graphs[name] + if !ok || info == nil { + return nil, false + } + return info, true +} + const registryFile = "registry.json" // loadRegistry reads registry.json from dataDir. A missing file yields an diff --git a/src/kg/internal/hub/server.go b/src/kg/internal/hub/server.go index b79756c..e831d8e 100644 --- a/src/kg/internal/hub/server.go +++ b/src/kg/internal/hub/server.go @@ -345,7 +345,7 @@ func (s *Server) handleGetGraph(w http.ResponseWriter, r *http.Request) { internalError(w, "load registry", err) return } - info, ok := reg.Graphs[name] + info, ok := reg.graph(name) if !ok { writeError(w, http.StatusNotFound, fmt.Sprintf("unknown graph %q", name)) return @@ -416,7 +416,7 @@ func (s *Server) handleGraphSearch(w http.ResponseWriter, r *http.Request) { internalError(w, "load registry", err) return } - info, ok := reg.Graphs[name] + info, ok := reg.graph(name) if !ok { writeError(w, http.StatusNotFound, fmt.Sprintf("unknown graph %q", name)) return @@ -435,7 +435,12 @@ func (s *Server) handleGraphSearch(w http.ResponseWriter, r *http.Request) { // whose search failed is logged and excluded, so the field never // over-reports coverage. for _, layer := range expandLayers(reg, name) { - layerResults, err := s.searchGraph(layer, reg.Graphs[layer], req.Query, req.Limit) + layerInfo, ok := reg.graph(layer) + if !ok { + log.Printf("layer search graph %q (layer of %q): registry entry missing or null — skipping", layer, name) + continue + } + layerResults, err := s.searchGraph(layer, layerInfo, req.Query, req.Limit) if err != nil { log.Printf("layer search graph %q (layer of %q): %v", layer, name, err) continue @@ -477,7 +482,7 @@ func expandLayers(reg *Registry, graph string) []string { for depth := 0; len(queue) > 0 && depth < maxLayerExpansion; depth++ { var next []string for _, g := range queue { - info, ok := reg.Graphs[g] + info, ok := reg.graph(g) if !ok { continue } @@ -490,7 +495,7 @@ func expandLayers(reg *Registry, graph string) []string { log.Printf("expand layers of %q: invalid layer name %q — skipping", graph, layer) continue } - if _, ok := reg.Graphs[layer]; !ok { + if _, ok := reg.graph(layer); !ok { log.Printf("expand layers of %q: layer graph %q not on this hub — skipping", graph, layer) continue } @@ -565,7 +570,7 @@ func (s *Server) handleFederatedSearch(w http.ResponseWriter, r *http.Request) { writeError(w, http.StatusBadRequest, fmt.Sprintf("invalid graph name %q", name)) return } - if _, ok := reg.Graphs[name]; !ok { + if _, ok := reg.graph(name); !ok { writeError(w, http.StatusNotFound, fmt.Sprintf("unknown graph %q", name)) return } @@ -575,7 +580,14 @@ func (s *Server) handleFederatedSearch(w http.ResponseWriter, r *http.Request) { out := make(map[string]any, len(names)) for _, name := range names { - info := reg.Graphs[name] + info, ok := reg.graph(name) + if !ok { + // Checked above when the name list was validated, so this only + // fires if the entry is null rather than absent — which would + // otherwise panic on info.Commit below. + out[name] = map[string]any{"error": "unknown graph"} + continue + } results, err := s.searchGraph(name, info, req.Query, req.Limit) if err != nil { log.Printf("federated search graph %q: %v", name, err) @@ -727,7 +739,7 @@ func (s *Server) checkGraphOwnership(name, repo, force string) error { // let the seed proceed and fail later if it is going to. return nil } - existing, ok := reg.Graphs[name] + existing, ok := reg.graph(name) if !ok || existing.Repo == "" || repo == "" || existing.Repo == repo { return nil } diff --git a/src/kg/internal/knowledge/health.go b/src/kg/internal/knowledge/health.go index 53740dd..4909689 100644 --- a/src/kg/internal/knowledge/health.go +++ b/src/kg/internal/knowledge/health.go @@ -180,13 +180,6 @@ func collectObservationAge(store *Store, projectID string) (*ObservationAge, err return nil, err } if !result.HasNext() { - // Same race the median query below guards against, and the same - // answer: a writer deleted the rows between the count and this query. - // - // Falling through instead would leave Newest and Oldest at the zero - // time, which printHealth renders via humanAge(gen.Sub(oa.Newest)) as - // an age of roughly two thousand years — the precise "every graph - // reads as centuries old" symptom this metric was added to expose. result.Close() return nil, nil } @@ -195,15 +188,17 @@ func collectObservationAge(store *Store, projectID string) (*ObservationAge, err result.Close() return nil, err } - if age.Newest, err = timeCell(row, 0); err != nil { - result.Close() + present, err := ageBoundsFromRow(row, age) + result.Close() + if err != nil { return nil, err } - if age.Oldest, err = timeCell(row, 1); err != nil { - result.Close() - return nil, err + if !present { + // The race the median query below also guards against: a writer + // deleted the rows between the count and this query. Reported as no + // age rather than an error — see that comment for why. + return nil, nil } - result.Close() // Median by position: the middle row (lower of the two for even counts) // of the timestamped observations in created_at order. @@ -246,6 +241,11 @@ func timeCell(row interface{ GetValue(uint64) (any, error) }, col uint64) (time. if err != nil { return time.Time{}, err } + return timeValue(v) +} + +// timeValue converts an already-read cell to a UTC time. +func timeValue(v any) (time.Time, error) { t, ok := v.(time.Time) if !ok { return time.Time{}, fmt.Errorf("timestamp cell has unexpected type %T", v) @@ -253,6 +253,42 @@ func timeCell(row interface{ GetValue(uint64) (any, error) }, col uint64) (time. return t.UTC(), nil } +// ageBoundsFromRow reads the max/min row into age, reporting whether the bounds +// were actually present. +// +// The absent case is NOT an empty result set. `RETURN max(...), min(...)` has +// no GROUP BY, so it is a full aggregation: over zero matched rows Kuzu returns +// exactly ONE row with both columns NULL (verified directly against the engine, +// not assumed). Guarding with !HasNext() therefore never fires, and feeding +// those NULLs to timeCell trips its type assertion and fails the whole run — +// while `kg health` is documented to always exit 0, being a report and not a +// gate. So a NULL here means "the rows went away", handled like the median +// query's missing row. +// +// A non-NULL value of the wrong type stays a hard error: that is a real defect, +// and timeCell exists precisely because these timestamps were once silently +// zeroed. +func ageBoundsFromRow(row interface{ GetValue(uint64) (any, error) }, age *ObservationAge) (bool, error) { + newestRaw, err := row.GetValue(0) + if err != nil { + return false, err + } + oldestRaw, err := row.GetValue(1) + if err != nil { + return false, err + } + if newestRaw == nil || oldestRaw == nil { + return false, nil + } + if age.Newest, err = timeValue(newestRaw); err != nil { + return false, err + } + if age.Oldest, err = timeValue(oldestRaw); err != nil { + return false, err + } + return true, nil +} + // countQuery runs a single-row count(*) query and returns the count. func countQuery(store *Store, query string, params map[string]any) (int, error) { result, err := store.QueryParams(query, params) diff --git a/src/kg/internal/knowledge/health_test.go b/src/kg/internal/knowledge/health_test.go new file mode 100644 index 0000000..a4f3ce5 --- /dev/null +++ b/src/kg/internal/knowledge/health_test.go @@ -0,0 +1,64 @@ +package knowledge + +import ( + "testing" + "time" +) + +// fakeAgeRow stands in for a query row so the NULL-aggregate path can be +// exercised without racing a real writer. +type fakeAgeRow struct { + vals []any + err error +} + +func (f fakeAgeRow) GetValue(col uint64) (any, error) { + if f.err != nil { + return nil, f.err + } + return f.vals[col], nil +} + +// A bare aggregate over zero matched rows returns one row of NULLs, not an +// empty result set — so NULL is what the max/min race actually looks like, and +// it must degrade to "no age" rather than failing the run. kg health is +// documented to always exit 0. +func TestAgeBoundsFromRow(t *testing.T) { + newest := time.Date(2026, 8, 1, 12, 0, 0, 0, time.UTC) + oldest := time.Date(2026, 1, 1, 12, 0, 0, 0, time.UTC) + + t.Run("both present", func(t *testing.T) { + age := &ObservationAge{} + present, err := ageBoundsFromRow(fakeAgeRow{vals: []any{newest, oldest}}, age) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if !present { + t.Fatal("present = false, want true") + } + if !age.Newest.Equal(newest) || !age.Oldest.Equal(oldest) { + t.Errorf("got newest=%v oldest=%v, want %v / %v", age.Newest, age.Oldest, newest, oldest) + } + }) + + t.Run("NULL aggregate degrades to absent, not an error", func(t *testing.T) { + age := &ObservationAge{} + present, err := ageBoundsFromRow(fakeAgeRow{vals: []any{nil, nil}}, age) + if err != nil { + t.Fatalf("NULL bounds returned an error, which would fail the whole health run: %v", err) + } + if present { + t.Error("present = true for NULL bounds") + } + if !age.Newest.IsZero() || !age.Oldest.IsZero() { + t.Errorf("age was populated from NULLs: newest=%v oldest=%v", age.Newest, age.Oldest) + } + }) + + t.Run("wrong type stays a hard error", func(t *testing.T) { + age := &ObservationAge{} + if _, err := ageBoundsFromRow(fakeAgeRow{vals: []any{"not-a-time", oldest}}, age); err == nil { + t.Error("expected an error for a non-time value — silent zeroing is the bug this metric exists to catch") + } + }) +}