359 lines
14 KiB
Go
359 lines
14 KiB
Go
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, fixturePraxisAttentionScoped("ent_muzick",
|
||
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"), routeCandidate("entity_attention"))
|
||
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"), routeCandidate("entity_attention"))
|
||
if !strings.Contains(reply, "descaled in june") {
|
||
t.Fatalf("expected entity-scoped local facts in the reply, got %q", reply)
|
||
}
|
||
}
|
||
|
||
// TestEntityAttention_UnscopedPraxisResponseIsRefused: a Praxis old enough to
|
||
// ignore the entity_id parameter answers the scoped question with the whole
|
||
// unscoped list. Relabelling those items "по «X»" is the same fabrication the
|
||
// canonical ref exists to prevent, arriving through a different door.
|
||
func TestEntityAttention_UnscopedPraxisResponseIsRefused(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": "disk almost full", "importance": 3.0},
|
||
))
|
||
h := ecoHandler(t, nexus, praxis, nil)
|
||
|
||
reply := h.handlePraxisAct(ctx, entityAttentionDec("muzick indexer"), routeCandidate("entity_attention"))
|
||
if strings.Contains(reply, "disk almost full") {
|
||
t.Fatalf("an unscoped response must not be read back as entity-scoped, got %q", reply)
|
||
}
|
||
if reply == "" {
|
||
t.Fatal("refusing the answer must still say something")
|
||
}
|
||
if praxis.Count("POST", "/api/v1/tools/surface") != 0 {
|
||
t.Error("items that were never spoken must not be surfaced")
|
||
}
|
||
}
|
||
|
||
// TestEntityAttention_ForeignItemsAreDropped: items tagged with another entity
|
||
// are dropped rather than spoken under this entity's name.
|
||
func TestEntityAttention_ForeignItemsAreDropped(t *testing.T) {
|
||
ctx := context.Background()
|
||
nexus := newFakeNexus(t, fixtureNexusResolved("ent_muzick", "Muzick indexer", "service"))
|
||
mixed := []map[string]any{
|
||
{"id": "item_1", "title": "indexer queue is backing up", "importance": 3.0, "entity_id": "ent_muzick"},
|
||
{"id": "item_2", "title": "the kettle is descaling", "importance": 1.0, "entity_id": "ent_kettle"},
|
||
}
|
||
praxis := newFakePraxis(t, mustJSON(mixed))
|
||
h := ecoHandler(t, nexus, praxis, nil)
|
||
|
||
reply := h.handlePraxisAct(ctx, entityAttentionDec("muzick indexer"), routeCandidate("entity_attention"))
|
||
if !strings.Contains(reply, "indexer queue is backing up") {
|
||
t.Fatalf("the matching item must be spoken, got %q", reply)
|
||
}
|
||
if strings.Contains(reply, "kettle") {
|
||
t.Fatalf("another entity's item must not be spoken here, got %q", reply)
|
||
}
|
||
}
|
||
|
||
// TestEntityAttention_TruncationIsNamed: reading three of many remembered
|
||
// facts must not be presented as everything she knows.
|
||
func TestEntityAttention_TruncationIsNamed(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)
|
||
|
||
for i := 0; i < 5; i++ {
|
||
id, err := h.dataStore.WriteFactAboutSubject(ctx, time.Now(), store.KindEnv,
|
||
"note", "the espresso machine", "факт "+string(rune('а'+i)), "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"), routeCandidate("entity_attention"))
|
||
if !strings.Contains(reply, "и это не всё") {
|
||
t.Fatalf("a truncated recall must say it is truncated, 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"), routeCandidate("entity_attention"))
|
||
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("нечто"), routeCandidate("entity_attention"))
|
||
if missing == "" {
|
||
t.Fatal("an unknown entity must still get an answer")
|
||
}
|
||
|
||
nexus.SetFault(503)
|
||
degraded := h.handlePraxisAct(ctx, entityAttentionDec("нечто"), routeCandidate("entity_attention"))
|
||
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"), routeCandidate("entity_attention"))
|
||
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"), routeCandidate("entity_attention"))
|
||
if strings.Contains(reply, "disk almost full") {
|
||
t.Fatalf("without nexus, 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))
|
||
}
|
||
}
|
||
|
||
// TestEnrichment_BackedOffFactsDoNotStallTheQueue: the pending queue is ordered
|
||
// by id, so the oldest facts are pulled first whether or not they are eligible.
|
||
// A batch of facts in backoff at the head must not hold every slot and stop
|
||
// enrichment for everything younger.
|
||
func TestEnrichment_BackedOffFactsDoNotStallTheQueue(t *testing.T) {
|
||
ctx := context.Background()
|
||
st := newTestStore(t)
|
||
total := 5
|
||
for i := 0; i < total; i++ {
|
||
if _, err := st.WriteFactAboutSubject(ctx, time.Now(), store.KindEnv, "likes",
|
||
"subject-"+string(rune('a'+i)), `"true"`, "infer:pref", 0.8, sql.NullInt64{}); err != nil {
|
||
t.Fatalf("WriteFactAboutSubject: %v", err)
|
||
}
|
||
}
|
||
|
||
nexus := newFakeNexus(t, fixtureNexusResolved("ent_x", "X", "service"))
|
||
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
|
||
// A batch smaller than the queue, so with no scan the last fact never
|
||
// reaches the head while the first ones are backed off.
|
||
w.batch = total - 1
|
||
|
||
nexus.SetFault(503)
|
||
w.tick(ctx)
|
||
if got := nexus.Count("POST", "/api/v1/resolve"); got != total-1 {
|
||
t.Fatalf("expected the first batch attempted, got %d calls", got)
|
||
}
|
||
|
||
// Second tick with Nexus healthy: the backed-off head must be skipped and
|
||
// the fact behind it resolved, not the same batch pulled and dropped.
|
||
nexus.SetFault(0)
|
||
w.tick(ctx)
|
||
facts, err := st.FactsByEntity(ctx, "ent_x", 10)
|
||
if err != nil {
|
||
t.Fatalf("FactsByEntity: %v", err)
|
||
}
|
||
if len(facts) == 0 {
|
||
t.Fatal("a due fact behind a backed-off batch must still be resolved")
|
||
}
|
||
}
|
||
|
||
// TestEnrichment_StoreWriteFailureBacksOffToo: the one failure mode where the
|
||
// resolve worked and the write did not must be paced like any other, not
|
||
// retried at full rate forever.
|
||
func TestEnrichment_StoreWriteFailureBacksOffToo(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)
|
||
}
|
||
pending, err := st.PendingFactResolutions(ctx, 10)
|
||
if err != nil || len(pending) != 1 {
|
||
t.Fatalf("setup: pending = %+v, %v", pending, 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
|
||
|
||
// Closing the store makes the resolution write fail while the Nexus call
|
||
// still succeeds — the split this path gets wrong.
|
||
if err := st.Close(); err != nil {
|
||
t.Fatalf("close store: %v", err)
|
||
}
|
||
if w.resolveOne(ctx, pending[0]) {
|
||
t.Fatal("a failed store write must not report success")
|
||
}
|
||
if w.due(pending[0].ID) {
|
||
t.Fatal("a failed store write must back the fact off like a failed resolve")
|
||
}
|
||
}
|