diff --git a/cmd/mavend/ecosystem_acts.go b/cmd/mavend/ecosystem_acts.go index d395540..e0c1f72 100644 --- a/cmd/mavend/ecosystem_acts.go +++ b/cmd/mavend/ecosystem_acts.go @@ -72,6 +72,7 @@ var praxisCapabilities = []praxisCapability{ }, }, listChangesCapability{}, + entityAttentionCapability{}, } // handlePraxisAct — dispatches ecosystem tool acts through the Praxis tools API. @@ -190,6 +191,106 @@ func (listChangesCapability) handle(ctx context.Context, h *reactiveHandler, px return "изменения: " + strings.Join(parts, "; ") } +// entityAttentionCapability answers "what's going on with X" by resolving X to +// a canonical Nexus entity and asking Praxis for that entity's attention items +// (Vikunja #272). Unlike listAttentionCapability it is scoped: the entity_id +// travels to Praxis as a query parameter instead of Maven filtering an unscoped +// list client-side, which is what makes the ref canonical end to end. +// +// It also folds in what Maven herself knows about the same entity — facts the +// enrichment worker has already resolved to that entity_id — so one question +// gets one answer across both stores. +type entityAttentionCapability struct{} + +func (entityAttentionCapability) aliases() []string { + return []string{"entity_attention", "что с", "как дела у", "статус"} +} + +func (entityAttentionCapability) handle(ctx context.Context, h *reactiveHandler, px *praxisClient, dec router.Decision) string { + subject := dec.Slots.Value + if subject == "" { + subject = dec.Slots.Text + } + if subject == "" { + return "про что именно спросить?" + } + if h.ecosystem == nil || h.ecosystem.nexus == nil { + // Without Nexus there is no canonical ref to scope by. Say so rather + // than quietly answering about something else. + return "не могу связать это с сущностью — Nexus не настроен." + } + + entityID, displayName, ambiguous, err := h.ecosystem.resolveEntityReference(ctx, subject, nil) + if err != nil { + log.Printf("ecosystem: entity attention resolve %q: %v", subject, err) + return "экосистема недоступна, попробуй ещё раз." + } + if len(ambiguous) > 0 { + return "уточни, что именно: " + strings.Join(ambiguous, ", ") + "?" + } + if entityID == "" { + return "не знаю такой сущности." + } + if displayName == "" { + displayName = subject + } + + items, err := px.ListAttentionForEntity(ctx, entityID, 20) + if err != nil { + log.Printf("ecosystem: praxis attention for %s: %v", entityID, err) + return "не могу сейчас узнать, что требует внимания по «" + displayName + "»." + } + h.recordPraxisTrace(ctx, "entity_attention", map[string]any{ + "entity_id": entityID, "count": len(items), + }) + + var parts []string + for _, item := range items { + title, _ := item["title"].(string) + if title == "" { + continue + } + parts = append(parts, title) + // Same surfaced != acknowledged rule as the unscoped digest. + if id, ok := item["id"].(string); ok && id != "" { + if _, err := px.Surface(ctx, id); err != nil { + log.Printf("ecosystem: praxis surface %s: %v", id, err) + } + } + } + if known := h.localFactsForEntity(ctx, entityID); known != "" { + parts = append(parts, known) + } + if len(parts) == 0 { + return "по «" + displayName + "» ничего нет." + } + return "по «" + displayName + "»: " + strings.Join(parts, "; ") +} + +// localFactsForEntity summarises Maven's own facts already resolved to this +// canonical entity. Empty when the store is unavailable or nothing matched — +// entity-scoped memory is an enrichment of the answer, never a precondition. +func (h *reactiveHandler) localFactsForEntity(ctx context.Context, entityID string) string { + if h.dataStore == nil || entityID == "" { + return "" + } + facts, err := h.dataStore.FactsByEntity(ctx, entityID, 3) + if err != nil { + log.Printf("ecosystem: facts by entity %s: %v", entityID, err) + return "" + } + var parts []string + for _, f := range facts { + if f.Value != "" { + parts = append(parts, f.Value) + } + } + if len(parts) == 0 { + return "" + } + return "я помню: " + strings.Join(parts, ", ") +} + // recordPraxisTrace — writes a fact recording a cross-service ecosystem call. // The fact is stored with source "praxis:trace" so the proactive loop can // reference it and the dashboard can display recent ecosystem activity. diff --git a/cmd/mavend/entityrefs_test.go b/cmd/mavend/entityrefs_test.go new file mode 100644 index 0000000..2f7d236 --- /dev/null +++ b/cmd/mavend/entityrefs_test.go @@ -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)) + } +} diff --git a/cmd/mavend/factenrichment.go b/cmd/mavend/factenrichment.go index e228215..a19ea55 100644 --- a/cmd/mavend/factenrichment.go +++ b/cmd/mavend/factenrichment.go @@ -9,6 +9,7 @@ package main import ( "context" "log" + "sync" "time" "github.com/kami/maven/internal/store" @@ -24,10 +25,69 @@ type factEnrichmentWorker struct { eco *ecosystemWiring interval time.Duration batch int // facts resolved per tick; keeps a single slow tick bounded + now func() time.Time + + // Retry state for facts whose resolution failed transiently. Kept in + // memory rather than in the DB: a restart legitimately retries + // everything, and the backoff exists to spare a struggling Nexus, not + // to be durable. A fact is never given up on — degraded means slower, + // not dropped. + mu sync.Mutex + attempt map[int64]int // fact id → consecutive failures + nextTry map[int64]time.Time // fact id → earliest retry + skipped int // facts held back by backoff on the last tick +} + +// enrichmentBackoff is the wait before retrying a fact after n consecutive +// failures, capped so a long Nexus outage still retries about hourly. +func enrichmentBackoff(n int) time.Duration { + d := time.Minute + for i := 1; i < n && d < time.Hour; i++ { + d *= 2 + } + if d > time.Hour { + d = time.Hour + } + return d } func newFactEnrichmentWorker(st *store.Store, eco *ecosystemWiring, interval time.Duration) *factEnrichmentWorker { - return &factEnrichmentWorker{store: st, eco: eco, interval: interval, batch: 20} + return &factEnrichmentWorker{ + store: st, + eco: eco, + interval: interval, + batch: 20, + now: time.Now, + attempt: map[int64]int{}, + nextTry: map[int64]time.Time{}, + } +} + +// enrichmentStatus is what the worker reports about its own health: how many +// facts are waiting, how many are currently in backoff, and the worst retry +// count seen. Degradation is reported, never hidden — a Nexus that has been +// down all day must be visible as a backlog, not as facts that silently +// never got tagged. +type enrichmentStatus struct { + Pending int + InBackoff int + MaxAttempts int +} + +func (w *factEnrichmentWorker) status(ctx context.Context) enrichmentStatus { + var st enrichmentStatus + if pending, err := w.store.PendingFactResolutions(ctx, 1000); err == nil { + st.Pending = len(pending) + } + w.mu.Lock() + defer w.mu.Unlock() + st.InBackoff = w.skipped + for _, n := range w.attempt { + if n > st.MaxAttempts { + st.MaxAttempts = n + } + } + return st } func (w *factEnrichmentWorker) run(ctx context.Context) { @@ -57,18 +117,50 @@ func (w *factEnrichmentWorker) tick(ctx context.Context) { log.Printf("factenrichment: list pending: %v", err) return } + skipped, failed := 0, 0 for _, f := range pending { - w.resolveOne(ctx, f) + if !w.due(f.ID) { + skipped++ + continue + } + if !w.resolveOne(ctx, f) { + failed++ + } + } + w.mu.Lock() + w.skipped = skipped + w.mu.Unlock() + if failed > 0 { + log.Printf("factenrichment: %d/%d resolutions failed this tick, %d held in backoff", + failed, len(pending), skipped) } } -func (w *factEnrichmentWorker) resolveOne(ctx context.Context, f store.Fact) { +// due reports whether a fact's backoff window has elapsed. +func (w *factEnrichmentWorker) due(id int64) bool { + w.mu.Lock() + defer w.mu.Unlock() + next, ok := w.nextTry[id] + return !ok || !w.now().Before(next) +} + +// resolveOne resolves one pending fact. It returns false when the attempt +// failed transiently: the fact stays pending and is retried on a backoff. +func (w *factEnrichmentWorker) resolveOne(ctx context.Context, f store.Fact) bool { entityID, _, ambiguous, err := w.eco.resolveEntityReference(ctx, f.Subject, nil) if err != nil { - // Transient (Nexus unreachable) — leave pending, retry next tick. + // Transient (Nexus unreachable) — leave pending, back off, retry later. log.Printf("factenrichment: resolve fact %d subject %q: %v", f.ID, f.Subject, err) - return + w.mu.Lock() + w.attempt[f.ID]++ + w.nextTry[f.ID] = w.now().Add(enrichmentBackoff(w.attempt[f.ID])) + w.mu.Unlock() + return false } + w.mu.Lock() + delete(w.attempt, f.ID) + delete(w.nextTry, f.ID) + w.mu.Unlock() state := store.ResolutionNotFound switch { case entityID != "": @@ -78,5 +170,7 @@ func (w *factEnrichmentWorker) resolveOne(ctx context.Context, f store.Fact) { } if err := w.store.ResolveFactEntity(ctx, f.ID, entityID, state); err != nil { log.Printf("factenrichment: record resolution for fact %d: %v", f.ID, err) + return false } + return true }