Query Praxis by canonical entity ref and back off enrichment retries (#272)

Add an entity-scoped attention capability: the subject is resolved to a canonical Nexus entity_id, the id travels to Praxis as a query scope instead of being dropped after resolution, and Maven's own facts already tagged with the same id join the answer. Ambiguous, unknown, degraded and no-Nexus cases each get a distinct reply and never a scoped query without a scope. Give the fact-enrichment worker per-fact exponential backoff capped at an hour and a status report of pending/in-backoff/worst-attempt counts, so a long Nexus outage shows as a visible backlog rather than facts that silently never got tagged. Nothing is ever given up on.
2026-08-01 06:52:25 +04:00
parent 69e2800ef3
commit 08f3db318f
3 changed files with 413 additions and 5 deletions
@@ -0,0 +1,213 @@
+package main
+
+import (
+	"context"
+	"database/sql"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/kami/maven/internal/router"
+	"github.com/kami/maven/internal/store"
+)
+
+// Entity-ref propagation, Maven side (Vikunja #272): the canonical Nexus
+// entity_id must reach Praxis as a query scope rather than being resolved and
+// then thrown away, and the enrichment that produces those ids must degrade
+// visibly instead of silently.
+
+func entityAttentionDec(subject string) router.Decision {
+	return router.Decision{
+		Intent: router.IntentAct,
+		Slots:  router.Slots{Fn: "entity_attention", HasFn: true, Value: subject},
+	}
+}
+
+// TestEntityAttention_ScopesPraxisByCanonicalID: the resolved id must travel
+// to Praxis in the request, not be used for client-side filtering.
+func TestEntityAttention_ScopesPraxisByCanonicalID(t *testing.T) {
+	ctx := context.Background()
+	nexus := newFakeNexus(t, fixtureNexusResolved("ent_muzick", "Muzick indexer", "service"))
+	praxis := newFakePraxis(t, fixturePraxisAttentionItems(
+		map[string]any{"id": "item_1", "title": "indexer queue is backing up", "importance": 3.0},
+	))
+	h := ecoHandler(t, nexus, praxis, nil)
+
+	reply := h.handlePraxisAct(ctx, entityAttentionDec("muzick indexer"))
+	if !strings.Contains(reply, "indexer queue is backing up") {
+		t.Fatalf("expected the scoped item in the reply, got %q", reply)
+	}
+
+	var scoped bool
+	for _, r := range praxis.Requests() {
+		if r.Method == "GET" && strings.HasPrefix(r.Path, "/api/v1/tools/attention") &&
+			strings.Contains(r.Query, "entity_id=ent_muzick") {
+			scoped = true
+		}
+	}
+	if !scoped {
+		t.Fatalf("expected attention scoped by entity_id, got requests %+v", praxis.Requests())
+	}
+	if praxis.Count("POST", "/api/v1/tools/surface") == 0 {
+		t.Error("a spoken scoped item must be surfaced, like the unscoped digest")
+	}
+}
+
+// TestEntityAttention_FoldsInLocalFactsForSameEntity: facts the enrichment
+// worker already tagged with the same canonical id join the same answer.
+func TestEntityAttention_FoldsInLocalFactsForSameEntity(t *testing.T) {
+	ctx := context.Background()
+	nexus := newFakeNexus(t, fixtureNexusResolved("ent_espresso", "the espresso machine", "device"))
+	praxis := newFakePraxis(t, fixturePraxisAttentionItems())
+	h := ecoHandler(t, nexus, praxis, nil)
+
+	id, err := h.dataStore.WriteFactAboutSubject(ctx, time.Now(), store.KindEnv,
+		"descaled", "the espresso machine", "descaled in june", "infer:pref", 0.8, sql.NullInt64{})
+	if err != nil {
+		t.Fatalf("WriteFactAboutSubject: %v", err)
+	}
+	if err := h.dataStore.ResolveFactEntity(ctx, id, "ent_espresso", store.ResolutionResolved); err != nil {
+		t.Fatalf("ResolveFactEntity: %v", err)
+	}
+
+	reply := h.handlePraxisAct(ctx, entityAttentionDec("the espresso machine"))
+	if !strings.Contains(reply, "descaled in june") {
+		t.Fatalf("expected entity-scoped local facts in the reply, got %q", reply)
+	}
+}
+
+// TestEntityAttention_AmbiguousAsksInsteadOfGuessing.
+func TestEntityAttention_AmbiguousAsksInsteadOfGuessing(t *testing.T) {
+	ctx := context.Background()
+	nexus := newFakeNexus(t, fixtureNexusAmbiguous(
+		map[string]string{"entity_id": "ent_a", "display_name": "Muzick indexer"},
+		map[string]string{"entity_id": "ent_b", "display_name": "Muzick web"},
+	))
+	praxis := newFakePraxis(t, fixturePraxisAttentionItems())
+	h := ecoHandler(t, nexus, praxis, nil)
+
+	reply := h.handlePraxisAct(ctx, entityAttentionDec("muzick"))
+	if !strings.Contains(reply, "Muzick indexer") || !strings.Contains(reply, "Muzick web") {
+		t.Fatalf("ambiguous subject must ask, got %q", reply)
+	}
+	if praxis.Count("GET", "/api/v1/tools/attention") != 0 {
+		t.Fatal("an ambiguous subject must not be queried against praxis")
+	}
+}
+
+// TestEntityAttention_MissingAndDegradedAreDistinct: "no such entity" and
+// "Nexus is down" must not produce the same answer.
+func TestEntityAttention_MissingAndDegradedAreDistinct(t *testing.T) {
+	ctx := context.Background()
+	nexus := newFakeNexus(t, fixtureNexusNotFound())
+	praxis := newFakePraxis(t, fixturePraxisAttentionItems())
+	h := ecoHandler(t, nexus, praxis, nil)
+
+	missing := h.handlePraxisAct(ctx, entityAttentionDec("нечто"))
+	if missing == "" {
+		t.Fatal("an unknown entity must still get an answer")
+	}
+
+	nexus.SetFault(503)
+	degraded := h.handlePraxisAct(ctx, entityAttentionDec("нечто"))
+	if degraded == missing {
+		t.Fatalf("outage and unknown-entity must not read the same: %q", degraded)
+	}
+}
+
+// TestEntityAttention_DelayedNexusDegradesNotHangs: a slow Nexus past the
+// caller's deadline degrades and never queries Praxis with an empty scope.
+func TestEntityAttention_DelayedNexusDegradesNotHangs(t *testing.T) {
+	nexus := newFakeNexus(t, fixtureNexusResolved("ent_muzick", "Muzick indexer", "service"))
+	praxis := newFakePraxis(t, fixturePraxisAttentionItems())
+	h := ecoHandler(t, nexus, praxis, nil)
+	nexus.SetDelay(2 * time.Second)
+
+	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Millisecond)
+	defer cancel()
+	reply := h.handlePraxisAct(ctx, entityAttentionDec("muzick indexer"))
+	if reply == "" {
+		t.Fatal("a delayed resolve must still answer")
+	}
+	if praxis.Count("GET", "/api/v1/tools/attention") != 0 {
+		t.Fatal("praxis must not be queried without a resolved scope")
+	}
+}
+
+// TestEntityAttention_WithoutNexusSaysSo: no Nexus means no canonical ref, so
+// the scoped query is refused rather than answered about something else.
+func TestEntityAttention_WithoutNexusSaysSo(t *testing.T) {
+	ctx := context.Background()
+	praxis := newFakePraxis(t, fixturePraxisAttentionItems(
+		map[string]any{"id": "item_1", "title": "disk almost full", "importance": 3.0},
+	))
+	h := ecoHandler(t, nil, praxis, nil)
+
+	reply := h.handlePraxisAct(ctx, entityAttentionDec("muzick indexer"))
+	if strings.Contains(reply, "disk almost full") {
+		t.Fatalf("unscoped items must not be passed off as entity-scoped, got %q", reply)
+	}
+	if praxis.Count("GET", "/api/v1/tools/attention") != 0 {
+		t.Fatal("no canonical ref means no scoped query at all")
+	}
+}
+
+// TestEnrichmentBackoff_HoldsAndReleases: repeated Nexus failures back the
+// fact off instead of hammering, and the fact is retried once the window
+// elapses. Nothing is ever given up on.
+func TestEnrichmentBackoff_HoldsAndReleases(t *testing.T) {
+	ctx := context.Background()
+	nexus := newFakeNexus(t, fixtureNexusResolved("ent_espresso", "the espresso machine", "device"))
+	st := newTestStore(t)
+	if _, err := st.WriteFactAboutSubject(ctx, time.Now(), store.KindEnv, "likes",
+		"the espresso machine", `"true"`, "infer:pref", 0.8, sql.NullInt64{}); err != nil {
+		t.Fatalf("WriteFactAboutSubject: %v", err)
+	}
+
+	clock := newFakeClock(time.Date(2026, 8, 1, 3, 0, 0, 0, time.UTC))
+	w := newFactEnrichmentWorker(st, stubEcosystem(nexus.URL, ""), time.Hour)
+	w.now = clock.Now
+
+	nexus.SetFault(503)
+	w.tick(ctx)
+	failedCalls := nexus.Count("POST", "/api/v1/resolve")
+	if failedCalls != 1 {
+		t.Fatalf("expected one resolve attempt, got %d", failedCalls)
+	}
+
+	// Immediately after a failure the fact is in backoff: no second call.
+	w.tick(ctx)
+	if nexus.Count("POST", "/api/v1/resolve") != failedCalls {
+		t.Fatal("a fact in backoff must not be retried on the very next tick")
+	}
+	if s := w.status(ctx); s.Pending != 1 || s.InBackoff != 1 || s.MaxAttempts != 1 {
+		t.Fatalf("degradation must be reported, got %+v", s)
+	}
+
+	// Once the window elapses and Nexus recovers, the fact resolves.
+	clock.Advance(2 * time.Minute)
+	nexus.SetFault(0)
+	w.tick(ctx)
+	facts, err := st.FactsByEntity(ctx, "ent_espresso", 10)
+	if err != nil {
+		t.Fatalf("FactsByEntity: %v", err)
+	}
+	if len(facts) != 1 {
+		t.Fatalf("expected the fact resolved after recovery, got %+v", facts)
+	}
+	if s := w.status(ctx); s.Pending != 0 || s.MaxAttempts != 0 {
+		t.Fatalf("recovery must clear the degradation report, got %+v", s)
+	}
+}
+
+func TestEnrichmentBackoff_GrowsAndIsCapped(t *testing.T) {
+	if enrichmentBackoff(1) != time.Minute {
+		t.Fatalf("first retry should be a minute, got %v", enrichmentBackoff(1))
+	}
+	if enrichmentBackoff(3) != 4*time.Minute {
+		t.Fatalf("third retry should be four minutes, got %v", enrichmentBackoff(3))
+	}
+	if enrichmentBackoff(50) != time.Hour {
+		t.Fatalf("backoff must cap at an hour, got %v", enrichmentBackoff(50))
+	}
+}