Add entity-aware fact resolution against Nexus (Vikunja #279)

facts gain a Subject/EntityID/ResolutionState triple and an async enrichment worker that resolves free-text subjects to canonical Nexus entity_ids, mirroring Praxis's enrichment-worker pattern. Ambiguous or unreachable Nexus never guesses an entity_id — the fact stays pending or terminal-ambiguous instead. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_018ghELqYhZNLub2TXGMazqA
2026-07-20 12:00:37 +04:00
parent c932cd8677
commit 03fa52dfd4
8 changed files with 485 additions and 0 deletions
@@ -0,0 +1,82 @@
+// factenrichment.go — resolves the Subject of entity-aware facts against
+// Nexus, turning free-text ("the espresso machine", "Kate") into a canonical
+// entity_id (Vikunja #279). Mirrors Praxis's async enrichment worker
+// (internal/enrichment/worker.go there): a plain poll-queue-update loop, no
+// shared state with the tick loop beyond the store and ecosystemWiring both
+// already hold.
+package main
+
+import (
+	"context"
+	"log"
+	"time"
+
+	"github.com/kami/maven/internal/store"
+)
+
+// factEnrichmentWorker polls store.PendingFactResolutions and resolves each
+// one's Subject through ecosystemWiring.resolveEntityReference. A nil nexus
+// client (ecosystem.nexus unconfigured) makes every tick a no-op rather than
+// erroring — entity-aware memory degrades to "facts just aren't tagged",
+// never to a crash loop.
+type factEnrichmentWorker struct {
+	store    *store.Store
+	eco      *ecosystemWiring
+	interval time.Duration
+	batch    int // facts resolved per tick; keeps a single slow tick bounded
+}
+
+func newFactEnrichmentWorker(st *store.Store, eco *ecosystemWiring, interval time.Duration) *factEnrichmentWorker {
+	return &factEnrichmentWorker{store: st, eco: eco, interval: interval, batch: 20}
+}
+
+func (w *factEnrichmentWorker) run(ctx context.Context) {
+	if w.eco == nil || w.eco.nexus == nil {
+		log.Printf("factenrichment: nexus not configured, worker idle")
+		<-ctx.Done()
+		return
+	}
+
+	ticker := time.NewTicker(w.interval)
+	defer ticker.Stop()
+
+	w.tick(ctx)
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-ticker.C:
+			w.tick(ctx)
+		}
+	}
+}
+
+func (w *factEnrichmentWorker) tick(ctx context.Context) {
+	pending, err := w.store.PendingFactResolutions(ctx, w.batch)
+	if err != nil {
+		log.Printf("factenrichment: list pending: %v", err)
+		return
+	}
+	for _, f := range pending {
+		w.resolveOne(ctx, f)
+	}
+}
+
+func (w *factEnrichmentWorker) resolveOne(ctx context.Context, f store.Fact) {
+	entityID, _, ambiguous, err := w.eco.resolveEntityReference(ctx, f.Subject, nil)
+	if err != nil {
+		// Transient (Nexus unreachable) — leave pending, retry next tick.
+		log.Printf("factenrichment: resolve fact %d subject %q: %v", f.ID, f.Subject, err)
+		return
+	}
+	state := store.ResolutionNotFound
+	switch {
+	case entityID != "":
+		state = store.ResolutionResolved
+	case len(ambiguous) > 0:
+		state = store.ResolutionAmbiguous
+	}
+	if err := w.store.ResolveFactEntity(ctx, f.ID, entityID, state); err != nil {
+		log.Printf("factenrichment: record resolution for fact %d: %v", f.ID, err)
+	}
+}
@@ -0,0 +1,100 @@
+package main
+
+import (
+	"context"
+	"database/sql"
+	"testing"
+	"time"
+
+	"github.com/kami/maven/internal/store"
+)
+
+func TestFactEnrichmentWorker_ResolvesPendingSubject(t *testing.T) {
+	ctx := context.Background()
+	nexus := newFakeNexus(t, fixtureNexusResolved("ent_espresso", "the espresso machine", "device"))
+	st := newTestStore(t)
+
+	id, err := st.WriteFactAboutSubject(ctx, time.Now(), store.KindEnv, "likes", "the espresso machine", `"true"`, "infer:pref", 0.8, sql.NullInt64{})
+	if err != nil {
+		t.Fatalf("WriteFactAboutSubject: %v", err)
+	}
+
+	w := newFactEnrichmentWorker(st, stubEcosystem(nexus.URL, ""), time.Hour)
+	w.tick(ctx)
+
+	facts, err := st.FactsByEntity(ctx, "ent_espresso", 10)
+	if err != nil {
+		t.Fatalf("FactsByEntity: %v", err)
+	}
+	if len(facts) != 1 || facts[0].ID != id {
+		t.Fatalf("expected fact %d resolved to ent_espresso, got %+v", id, facts)
+	}
+
+	pending, err := st.PendingFactResolutions(ctx, 10)
+	if err != nil {
+		t.Fatalf("PendingFactResolutions: %v", err)
+	}
+	if len(pending) != 0 {
+		t.Fatalf("expected no facts left pending, got %+v", pending)
+	}
+}
+
+// TestFactEnrichmentWorker_NexusDownLeavesFactPending covers the fail-closed
+// contract: a Nexus outage must not mark a fact resolved/not_found — it
+// should stay pending so a later successful tick can still resolve it.
+func TestFactEnrichmentWorker_NexusDownLeavesFactPending(t *testing.T) {
+	ctx := context.Background()
+	nexus := newFakeNexus(t, fixtureNexusResolved("ent_espresso", "the espresso machine", "device"))
+	nexus.SetFault(503)
+	st := newTestStore(t)
+
+	id, err := st.WriteFactAboutSubject(ctx, time.Now(), store.KindEnv, "likes", "the espresso machine", `"true"`, "infer:pref", 0.8, sql.NullInt64{})
+	if err != nil {
+		t.Fatalf("WriteFactAboutSubject: %v", err)
+	}
+
+	w := newFactEnrichmentWorker(st, stubEcosystem(nexus.URL, ""), time.Hour)
+	w.tick(ctx)
+
+	pending, err := st.PendingFactResolutions(ctx, 10)
+	if err != nil {
+		t.Fatalf("PendingFactResolutions: %v", err)
+	}
+	if len(pending) != 1 || pending[0].ID != id {
+		t.Fatalf("expected fact %d to remain pending during nexus outage, got %+v", id, pending)
+	}
+}
+
+func TestFactEnrichmentWorker_AmbiguousLeavesEntityIDUnset(t *testing.T) {
+	ctx := context.Background()
+	nexus := newFakeNexus(t, fixtureNexusAmbiguous(
+		map[string]string{"entity_id": "ent_kate_1", "display_name": "Kate Smith"},
+		map[string]string{"entity_id": "ent_kate_2", "display_name": "Kate Jones"},
+	))
+	st := newTestStore(t)
+
+	id, err := st.WriteFactAboutSubject(ctx, time.Now(), store.KindEnv, "likes", "kate", `"true"`, "infer:pref", 0.8, sql.NullInt64{})
+	if err != nil {
+		t.Fatalf("WriteFactAboutSubject: %v", err)
+	}
+
+	w := newFactEnrichmentWorker(st, stubEcosystem(nexus.URL, ""), time.Hour)
+	w.tick(ctx)
+
+	pending, err := st.PendingFactResolutions(ctx, 10)
+	if err != nil {
+		t.Fatalf("PendingFactResolutions: %v", err)
+	}
+	if len(pending) != 0 {
+		t.Fatalf("ambiguous resolution should leave the pending queue, got %+v", pending)
+	}
+
+	facts, err := st.FactsByEntity(ctx, "ent_kate_1", 10)
+	if err != nil {
+		t.Fatalf("FactsByEntity: %v", err)
+	}
+	if len(facts) != 0 {
+		t.Fatalf("ambiguous resolution must not guess an entity_id, got %+v", facts)
+	}
+	_ = id
+}
@@ -249,6 +249,7 @@ func run(args []string) error {
 		tl          *tickLoop
 		coreAPI     ipc.CoreAPI
 		eco         *ecosystemWiring
+		factWorker  *factEnrichmentWorker
 	)

 	if !locked {
@@ -341,6 +342,7 @@ func run(args []string) error {
 		repeatInterval := time.Duration(cfg.RepeatInterval)
 		autotuneInterval := time.Duration(cfg.AutotuneInterval)
 		tl = newTickLoop(st, gatherer, dispatcher, phr, rules, tickInterval, repeatInterval, autotuneInterval, cfg.Digest, routinesFromConfig(cfg.Routines))
+		factWorker = newFactEnrichmentWorker(st, eco, time.Duration(cfg.FactEnrichmentInterval))

 		coreAPI = &daemonAPI{
 			CoreAPI:  ipc.NewStoreAPI(st),
@@ -500,6 +502,7 @@ func run(args []string) error {
 			repeatInterval := time.Duration(cfg.RepeatInterval)
 			autotuneInterval := time.Duration(cfg.AutotuneInterval)
 			tl = newTickLoop(st, gatherer, dispatcher, phr, rules, tickInterval, repeatInterval, autotuneInterval, cfg.Digest, routinesFromConfig(cfg.Routines))
+			factWorker = newFactEnrichmentWorker(st, eco, time.Duration(cfg.FactEnrichmentInterval))

 			// Swap the CoreAPI from lockedAPI to the real store adapter.
 			newAPI := &daemonAPI{
@@ -530,6 +533,11 @@ func run(args []string) error {
 				tl.run(ctx)
 			}()

+			// Start fact-entity enrichment worker.
+			go func() {
+				factWorker.run(ctx)
+			}()
+
 			dl.unlock()
 			log.Printf("mavend: unlocked via passkey assertion")
 			return nil
@@ -563,6 +571,11 @@ func run(args []string) error {
 			defer wg.Done()
 			tl.run(ctx)
 		}()
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			factWorker.run(ctx)
+		}()
 	}

 	<-ctx.Done()