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.

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.

cmd/mavend/ecosystem_degraded_test.go

@@ -0,0 +1,312 @@
+package main
+
+import (
+	"context"
+	"net/http"
+	"strings"
+	"testing"
+	"time"
+
+	hexisclient "github.com/kami/hexis/pkg/client"
+
+	"github.com/kami/maven/internal/ipc"
+	"github.com/kami/maven/internal/store"
+)
+
+// Phase-5 hardening suite (Vikunja #276). Everything here drives the shared
+// fake ecosystem (fakeecosystem_test.go) rather than one-off inline handlers,
+// so the same fault levers — SetFault, SetBody, SetDelay — cover every
+// service. What is asserted is the degraded-mode contract:
+//
+//   - services degrade independently: one outage never mutes the others,
+//   - a degraded reply is never silent, never fabricated, never "success",
+//   - contract drift (old shape, unknown fields, garbage) is survivable,
+//   - Maven never acts on an ambiguous target and never chains
+//     Praxis observation into Hexis execution on its own.
+
+// ecoHandler wires a handler against whichever of the three fakes is given
+// (pass nil to leave a service unconfigured, which is a different state from
+// "configured but down").
+func ecoHandler(t *testing.T, nexus, praxis, hexis *fakeServer) *reactiveHandler {
+	t.Helper()
+	st := newTestStore(t)
+	clock := newFakeClock(time.Date(2026, 8, 1, 9, 0, 0, 0, time.UTC))
+	w := &ecosystemWiring{}
+	if nexus != nil {
+		w.nexus = newNexusClient(nexus.URL)
+	}
+	if praxis != nil {
+		w.praxis = newPraxisClient(praxis.URL)
+	}
+	if hexis != nil {
+		w.hexis = hexisclient.New(hexis.URL)
+	}
+	return &reactiveHandler{
+		api:       ipc.NewStoreAPI(st),
+		dataStore: st,
+		now:       clock.Now,
+		ecosystem: w,
+	}
+}
+
+func traceFacts(t *testing.T, h *reactiveHandler) []store.Fact {
+	t.Helper()
+	facts, err := h.dataStore.RecentFacts(context.Background(), 50)
+	if err != nil {
+		t.Fatalf("read facts: %v", err)
+	}
+	var out []store.Fact
+	for _, f := range facts {
+		if f.Source == "praxis:trace" {
+			out = append(out, f)
+		}
+	}
+	return out
+}
+
+func restartCaps() string {
+	return fixtureHexisCapabilities(map[string]any{
+		"id": "cap_restart", "name": "restart", "read_only": true,
+	})
+}
+
+// TestEcosystem_OutagesAreIndependent: Praxis being down must not disable the
+// Nexus+Hexis action path, and vice versa. A shared "ecosystem is broken"
+// mode would take away working capability for no reason.
+func TestEcosystem_OutagesAreIndependent(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,
+	}))
+	hexis := newFakeHexis(t, restartCaps(), fixtureHexisExecuted("exec_1", "succeeded"))
+	h := ecoHandler(t, nexus, praxis, hexis)
+
+	praxis.SetFault(503)
+	if reply := h.handleHexisAct(ctx, actDec("muzick indexer")); !strings.Contains(reply, "выполнена") {
+		t.Fatalf("praxis outage must not block the hexis path, got %q", reply)
+	}
+
+	praxis.SetFault(0)
+	hexis.SetFault(503)
+	nexus.SetFault(503)
+	reply := h.handlePraxisAct(ctx, praxisActDec("list_attention"))
+	if !strings.Contains(reply, "disk almost full") {
+		t.Fatalf("nexus/hexis outage must not block the praxis digest, got %q", reply)
+	}
+}
+
+// TestEcosystem_MalformedNexusResponseFailsClosed: a 200 carrying garbage is a
+// dependency failure, not "no such entity". It must stop before Hexis.
+func TestEcosystem_MalformedNexusResponseFailsClosed(t *testing.T) {
+	ctx := context.Background()
+	nexus := newFakeNexus(t, fixtureNexusResolved("ent_muzick", "Muzick indexer", "service"))
+	hexis := newFakeHexis(t, restartCaps(), fixtureHexisExecuted("exec_1", "succeeded"))
+	h := ecoHandler(t, nexus, nil, hexis)
+
+	nexus.SetBody(`{"status":"resolved","entity":`)
+	reply := h.handleHexisAct(ctx, actDec("muzick indexer"))
+	if reply == "" || strings.Contains(reply, "выполнена") {
+		t.Fatalf("malformed nexus body must degrade, got %q", reply)
+	}
+	if hexis.Count("", "/api/v1") != 0 {
+		t.Fatal("hexis must not be contacted after a malformed nexus response")
+	}
+}
+
+// TestEcosystem_UnknownContractFieldsTolerated: a newer Nexus adding fields
+// must not break an older Maven. Same for the older flat resolve shape.
+func TestEcosystem_UnknownContractFieldsTolerated(t *testing.T) {
+	ctx := context.Background()
+	for name, body := range map[string]string{
+		"future": fixtureNexusResolvedFuture("ent_muzick", "Muzick indexer", "service"),
+		"flat":   fixtureNexusResolvedFlat("ent_muzick", "Muzick indexer", "service"),
+	} {
+		t.Run(name, func(t *testing.T) {
+			nexus := newFakeNexus(t, body)
+			hexis := newFakeHexis(t, restartCaps(), fixtureHexisExecuted("exec_1", "succeeded"))
+			h := ecoHandler(t, nexus, nil, hexis)
+			if reply := h.handleHexisAct(ctx, actDec("muzick indexer")); !strings.Contains(reply, "выполнена") {
+				t.Fatalf("%s contract shape must still resolve and execute, got %q", name, reply)
+			}
+		})
+	}
+}
+
+// TestEcosystem_CancelledContextDegrades: a caller hanging up (turn abandoned,
+// deadline hit) must surface as degradation, never as a fabricated result.
+func TestEcosystem_CancelledContextDegrades(t *testing.T) {
+	nexus := newFakeNexus(t, fixtureNexusResolved("ent_muzick", "Muzick indexer", "service"))
+	hexis := newFakeHexis(t, restartCaps(), fixtureHexisExecuted("exec_1", "succeeded"))
+	h := ecoHandler(t, nexus, nil, hexis)
+	nexus.SetDelay(2 * time.Second)
+
+	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Millisecond)
+	defer cancel()
+	reply := h.handleHexisAct(ctx, actDec("muzick indexer"))
+	if reply == "" || strings.Contains(reply, "выполнена") {
+		t.Fatalf("cancelled resolve must degrade, got %q", reply)
+	}
+	if hexis.Count("", "/api/v1") != 0 {
+		t.Fatal("hexis must not be contacted after a cancelled resolve")
+	}
+}
+
+// TestEcosystem_ExecutionFailureIsNotSuccess: Hexis answering 200 with
+// status=failed is a partial failure — the call worked, the command did not.
+// Maven must report it as a failure and must not write a success trace.
+func TestEcosystem_ExecutionFailureIsNotSuccess(t *testing.T) {
+	ctx := context.Background()
+	nexus := newFakeNexus(t, fixtureNexusResolved("ent_muzick", "Muzick indexer", "service"))
+	hexis := newFakeHexis(t, restartCaps(), fixtureHexisExecutionFailed("exec_1", "unit not found"))
+	h := ecoHandler(t, nexus, nil, hexis)
+
+	reply := h.handleHexisAct(ctx, actDec("muzick indexer"))
+	if strings.Contains(reply, "выполнена") {
+		t.Fatalf("failed execution must not read as success, got %q", reply)
+	}
+	if reply == "" {
+		t.Fatal("failed execution must say something")
+	}
+	for _, f := range traceFacts(t, h) {
+		if strings.HasPrefix(f.Key, "praxis:hexis:") {
+			t.Fatalf("failed execution must not write a success trace: %+v", f)
+		}
+	}
+}
+
+// TestEcosystem_AmbiguousTargetBlocksExecution: ambiguity blocks mutation, and
+// the clarification must name the candidates rather than pick one.
+func TestEcosystem_AmbiguousTargetBlocksExecution(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"},
+	))
+	hexis := newFakeHexis(t, restartCaps(), fixtureHexisExecuted("exec_1", "succeeded"))
+	h := ecoHandler(t, nexus, nil, hexis)
+
+	reply := h.handleHexisAct(ctx, actDec("muzick"))
+	if !strings.Contains(reply, "Muzick indexer") || !strings.Contains(reply, "Muzick web") {
+		t.Fatalf("ambiguous resolve must list candidates, got %q", reply)
+	}
+	if hexis.Count("POST", "/api/v1/execute") != 0 {
+		t.Fatal("ambiguous target must never execute")
+	}
+}
+
+// TestEcosystem_NoAutonomousPraxisToHexis: reading the attention digest is an
+// observation. Maven must never turn an observed problem into a Hexis command
+// by herself — she is not autonomous.
+func TestEcosystem_NoAutonomousPraxisToHexis(t *testing.T) {
+	ctx := context.Background()
+	praxis := newFakePraxis(t, fixturePraxisAttentionItems(
+		map[string]any{"id": "item_1", "title": "muzick indexer is down", "importance": 4.0, "rule": "service_down"},
+	))
+	nexus := newFakeNexus(t, fixtureNexusResolved("ent_muzick", "Muzick indexer", "service"))
+	hexis := newFakeHexis(t, restartCaps(), fixtureHexisExecuted("exec_1", "succeeded"))
+	h := ecoHandler(t, nexus, praxis, hexis)
+
+	_ = h.handlePraxisAct(ctx, praxisActDec("list_attention"))
+	if hexis.Count("", "/api/v1") != 0 {
+		t.Fatal("attention digest must not contact hexis on its own")
+	}
+	if nexus.Count("", "/api/v1/resolve") != 0 {
+		t.Fatal("attention digest must not resolve targets for autonomous action")
+	}
+}
+
+// TestEcosystem_MutatingCapabilityWaitsForConfirmation: a non-read-only
+// capability parks for an explicit spoken confirm bound to capability+target.
+func TestEcosystem_MutatingCapabilityWaitsForConfirmation(t *testing.T) {
+	ctx := context.Background()
+	nexus := newFakeNexus(t, fixtureNexusResolved("ent_muzick", "Muzick indexer", "service"))
+	caps := fixtureHexisCapabilities(map[string]any{"id": "cap_restart", "name": "restart", "read_only": false})
+	hexis := newFakeHexis(t, caps, fixtureHexisExecuted("exec_1", "succeeded"))
+	h := ecoHandler(t, nexus, nil, hexis)
+
+	reply := h.handleHexisAct(ctx, actDec("restart"))
+	if !strings.Contains(reply, "restart") || !strings.Contains(reply, "да") {
+		t.Fatalf("mutating capability must ask for confirmation, got %q", reply)
+	}
+	if hexis.Count("POST", "/api/v1/execute") != 0 {
+		t.Fatal("mutating capability must not execute before confirmation")
+	}
+	h.mu.Lock()
+	pending := h.pendingHexis
+	h.mu.Unlock()
+	if pending == nil || pending.capabilityID != "cap_restart" || pending.entityID != "ent_muzick" {
+		t.Fatalf("confirmation must be bound to capability+target, got %+v", pending)
+	}
+}
+
+// TestEcosystem_SurfaceFailureStillDelivers: surfacing is bookkeeping. If the
+// surface call fails the digest must still be spoken — a partial failure
+// downgrades bookkeeping, not the answer.
+func TestEcosystem_SurfaceFailureStillDelivers(t *testing.T) {
+	ctx := context.Background()
+	praxis := newFakeServer(t, map[string]http.HandlerFunc{
+		"GET /api/v1/tools/attention": jsonHandler(200, fixturePraxisAttentionItems(
+			map[string]any{"id": "item_1", "title": "disk almost full", "importance": 3.0},
+		)),
+		"POST /api/v1/tools/surface": jsonHandler(500, `{"error":"boom"}`),
+	})
+	h := ecoHandler(t, nil, praxis, nil)
+
+	reply := h.handlePraxisAct(ctx, praxisActDec("list_attention"))
+	if !strings.Contains(reply, "disk almost full") {
+		t.Fatalf("failed surface must not swallow the digest, got %q", reply)
+	}
+	if praxis.Count("POST", "/api/v1/tools/surface") == 0 {
+		t.Fatal("expected the surface attempt")
+	}
+}
+
+// TestEcosystem_TotalOutageSaysSoForEveryPath: with all three down, every
+// entry point degrades explicitly instead of returning empty or inventing.
+func TestEcosystem_TotalOutageSaysSoForEveryPath(t *testing.T) {
+	ctx := context.Background()
+	nexus := newFakeNexus(t, fixtureNexusResolved("ent_muzick", "Muzick indexer", "service"))
+	praxis := newFakePraxis(t, fixturePraxisAttentionItems())
+	hexis := newFakeHexis(t, restartCaps(), fixtureHexisExecuted("exec_1", "succeeded"))
+	for _, fs := range []*fakeServer{nexus, praxis, hexis} {
+		fs.SetFault(503)
+	}
+	h := ecoHandler(t, nexus, praxis, hexis)
+
+	for name, reply := range map[string]string{
+		"hexis act":   h.handleHexisAct(ctx, actDec("muzick indexer")),
+		"attention":   h.handlePraxisAct(ctx, praxisActDec("list_attention")),
+		"changes":     h.handlePraxisAct(ctx, praxisActDec("list_changes")),
+		"acknowledge": h.handlePraxisAct(ctx, praxisActDec("acknowledge_item")),
+	} {
+		if reply == "" {
+			t.Errorf("%s: total outage must not answer with silence", name)
+		}
+		if strings.Contains(reply, "выполнена") {
+			t.Errorf("%s: total outage must not claim success: %q", name, reply)
+		}
+	}
+	if len(traceFacts(t, h)) != 0 {
+		t.Fatal("a total outage must not leave success traces behind")
+	}
+}
+
+// TestEcosystem_RecoveryAfterOutageNeedsNoRestart: once the dependency comes
+// back the very next turn works — no cached failure state, no restart.
+func TestEcosystem_RecoveryAfterOutageNeedsNoRestart(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)
+
+	praxis.SetFault(503)
+	if reply := h.handlePraxisAct(ctx, praxisActDec("list_attention")); strings.Contains(reply, "disk") {
+		t.Fatalf("outage must not serve content, got %q", reply)
+	}
+	praxis.SetFault(0)
+	if reply := h.handlePraxisAct(ctx, praxisActDec("list_attention")); !strings.Contains(reply, "disk almost full") {
+		t.Fatalf("recovery must work on the next turn, got %q", reply)
+	}
+}

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))
+	}
+}

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
 }

cmd/mavend/fakeecosystem_test.go

@@ -14,7 +14,9 @@ import (
 type capturedRequest struct {
 	Method string
 	Path   string
+	Query  string
 	Body   []byte
+	Header http.Header
 }
 
 // fakeServer is the common shell behind fakeNexus/fakePraxis/fakeHexis: an
@@ -27,7 +29,9 @@ type fakeServer struct {
 
 	mu       sync.Mutex
 	requests []capturedRequest
-	fault    int // non-zero: every request gets this HTTP status instead of routing
+	fault    int    // non-zero: every request gets this HTTP status instead of routing
+	garbage  string // non-empty: returned 200 verbatim instead of routing (malformed-contract lever)
+	delay    time.Duration
 }
 
 // newFakeServer starts a server dispatching to routes keyed by "METHOD
@@ -47,14 +51,34 @@ func newFakeServer(t *testing.T, routes map[string]http.HandlerFunc) *fakeServer
 			}
 		}
 		fs.mu.Lock()
-		fs.requests = append(fs.requests, capturedRequest{Method: r.Method, Path: r.URL.Path, Body: body})
+		fs.requests = append(fs.requests, capturedRequest{
+			Method: r.Method,
+			Path:   r.URL.Path,
+			Query:  r.URL.RawQuery,
+			Body:   body,
+			Header: r.Header.Clone(),
+		})
 		fault := fs.fault
+		garbage := fs.garbage
+		delay := fs.delay
 		fs.mu.Unlock()
 
+		if delay > 0 {
+			select {
+			case <-time.After(delay):
+			case <-r.Context().Done():
+				return
+			}
+		}
 		if fault != 0 {
 			http.Error(w, "injected fault", fault)
 			return
 		}
+		if garbage != "" {
+			w.Header().Set("Content-Type", "application/json")
+			w.Write([]byte(garbage))
+			return
+		}
 
 		for key, handler := range routes {
 			method, prefix := splitRouteKey(key)
@@ -90,6 +114,36 @@ func (fs *fakeServer) SetFault(status int) {
 	fs.fault = status
 }
 
+// SetBody makes every subsequent request answer 200 with the given body,
+// bypassing the route table. Used to serve a malformed or contract-violating
+// payload where the transport itself is healthy. Pass "" to clear it.
+func (fs *fakeServer) SetBody(body string) {
+	fs.mu.Lock()
+	defer fs.mu.Unlock()
+	fs.garbage = body
+}
+
+// SetDelay stalls every subsequent request for d before answering, so callers
+// can drive client timeouts and context cancellation deterministically. The
+// delay is abandoned as soon as the client hangs up.
+func (fs *fakeServer) SetDelay(d time.Duration) {
+	fs.mu.Lock()
+	defer fs.mu.Unlock()
+	fs.delay = d
+}
+
+// Count returns how many captured requests used the given method and path
+// prefix. "" matches any method.
+func (fs *fakeServer) Count(method, prefix string) int {
+	n := 0
+	for _, r := range fs.Requests() {
+		if (method == "" || r.Method == method) && hasPrefix(r.Path, prefix) {
+			n++
+		}
+	}
+	return n
+}
+
 // Requests returns a snapshot of captured requests, in arrival order.
 func (fs *fakeServer) Requests() []capturedRequest {
 	fs.mu.Lock()
@@ -118,6 +172,32 @@ func fixtureNexusResolved(entityID, displayName, entityType string) string {
 	})
 }
 
+// fixtureNexusResolvedFlat is the flat resolve shape documented in
+// ECOSYSTEM-SPEC.md §1.5 (entity_id/entity_type/display_name at the top
+// level) rather than the nested "entity" object — the older of the two
+// wire shapes Maven must keep accepting.
+func fixtureNexusResolvedFlat(entityID, displayName, entityType string) string {
+	return mustJSON(map[string]any{
+		"status":       "resolved",
+		"entity_id":    entityID,
+		"entity_type":  entityType,
+		"display_name": displayName,
+	})
+}
+
+// fixtureNexusResolvedFuture is a resolved response from a hypothetical newer
+// Nexus: same required fields plus unknown ones. Decoding must ignore the
+// extras, not fail — forward compatibility is what lets the ecosystem be
+// upgraded one service at a time.
+func fixtureNexusResolvedFuture(entityID, displayName, entityType string) string {
+	return mustJSON(map[string]any{
+		"status":          "resolved",
+		"entity":          map[string]any{"id": entityID, "display_name": displayName, "type": entityType, "tenant": "home"},
+		"provenance":      map[string]any{"resolver": "v3", "graph_epoch": 42},
+		"score_breakdown": []any{map[string]any{"signal": "alias", "weight": 0.9}},
+	})
+}
+
 func fixtureNexusNotFound() string {
 	return `{"status":"not_found"}`
 }
@@ -138,6 +218,13 @@ func fixtureHexisExecuted(id, status string) string {
 	return mustJSON(map[string]any{"id": id, "status": status})
 }
 
+// fixtureHexisExecutionFailed is a well-formed Hexis response reporting that
+// the command itself failed: the call succeeded, the execution did not. Maven
+// must distinguish this from a transport failure and from success.
+func fixtureHexisExecutionFailed(id, message string) string {
+	return mustJSON(map[string]any{"id": id, "status": "failed", "error": message})
+}
+
 func fixturePraxisAttentionItems(items ...map[string]any) string {
 	return mustJSON(items)
 }
@@ -191,8 +278,13 @@ func newFakeNexus(t *testing.T, resolveBody string) *fakeServer {
 // fault is injected via SetFault.
 func newFakePraxis(t *testing.T, attentionBody string) *fakeServer {
 	return newFakeServer(t, map[string]http.HandlerFunc{
-		"GET /api/v1/tools/attention": jsonHandler(http.StatusOK, attentionBody),
-		"POST /api/v1/tools/surface":  jsonHandler(http.StatusOK, `{}`),
+		"GET /api/v1/tools/attention":    jsonHandler(http.StatusOK, attentionBody),
+		"GET /api/v1/tools/changes":      jsonHandler(http.StatusOK, `[]`),
+		"POST /api/v1/tools/surface":     jsonHandler(http.StatusOK, `{}`),
+		"POST /api/v1/tools/acknowledge": jsonHandler(http.StatusOK, `{}`),
+		"POST /api/v1/tools/resolve":     jsonHandler(http.StatusOK, `{}`),
+		"POST /api/v1/tools/ignore":      jsonHandler(http.StatusOK, `{}`),
+		"POST /api/v1/tools/pin":         jsonHandler(http.StatusOK, `{}`),
 	})
 }