ecosystem: assign one correlation ID per action and fail closed on scope

The act path minted IDs per hop and trusted whatever Praxis returned for a scoped attention query. A service that ignored the entity filter would have had its unrelated items read back to the owner as his. The handler now assigns one correlation ID at the top of the action and passes it down, and drops any item the response did not tag with the requested entity. Traces are written to the trace table with the causation ID and HTTP status hoisted into columns, the duplicate legacy Hexis trace is gone, truncated lists say so, and a rejected credential gets its own reply instead of looking like an outage. Found in review of #83 and #84.
2026-08-01 14:15:33 +04:00
parent f432eb0b25
commit 252f773223
1 changed files with 167 additions and 69 deletions
@@ -2,7 +2,6 @@ package main

 import (
 	"context"
-	"encoding/json"
 	"errors"
 	"fmt"
 	"log"
@@ -10,8 +9,8 @@ import (
 	"time"

 	hexisclient "github.com/kami/hexis/pkg/client"
-	"github.com/kami/maven/internal/ipc"
 	"github.com/kami/maven/internal/router"
+	"github.com/kami/maven/internal/store"
 )

 // praxisCapability is one arm of the Praxis act dispatch. This is an interface
@@ -84,6 +83,12 @@ func (h *reactiveHandler) handlePraxisAct(ctx context.Context, dec router.Decisi
 	if h.ecosystem == nil || h.ecosystem.praxis == nil {
 		return ""
 	}
+	// Every hop of this action shares one correlation ID, assigned here, so a
+	// digest that calls attention once and surface N times reads as one turn
+	// on the Praxis side instead of N+1 unrelated request ids.
+	if correlationIDFromCtx(ctx) == "" {
+		ctx = withCorrelationID(ctx, newCorrelationID())
+	}
 	px := h.ecosystem.praxis
 	for _, capability := range praxisCapabilities {
 		for _, alias := range capability.aliases() {
@@ -114,11 +119,14 @@ func (a praxisItemAction) handle(ctx context.Context, h *reactiveHandler, px *pr
 	if id == "" {
 		return a.ask
 	}
+	started := h.now()
 	if err := a.call(ctx, px, id); err != nil {
 		log.Printf("ecosystem: praxis %s %s: %v", a.op, id, err)
+		h.recordEcosystemTrace(ctx, "praxis", a.op, traceStatusForError(err), started,
+			mergeFields(traceErrorFields(err), map[string]any{"item_id": id}))
 		return a.failure
 	}
-	h.recordPraxisTrace(ctx, a.op, map[string]any{"item_id": id})
+	h.recordPraxisTrace(ctx, a.op, started, map[string]any{"item_id": id})
 	return a.success
 }

@@ -130,15 +138,18 @@ func (listAttentionCapability) aliases() []string {
 }

 func (listAttentionCapability) handle(ctx context.Context, h *reactiveHandler, px *praxisClient, _ router.Decision) string {
+	started := h.now()
 	items, err := px.ListAttention(ctx, 20)
 	if err != nil {
 		log.Printf("ecosystem: praxis attention: %v", err)
+		h.recordEcosystemTrace(ctx, "praxis", "list_attention", traceStatusForError(err),
+			started, traceErrorFields(err))
 		return "не могу сейчас узнать, что требует внимания."
 	}
 	if len(items) == 0 {
 		return "ничего не требует внимания."
 	}
-	h.recordPraxisTrace(ctx, "list_attention", map[string]any{"count": len(items)})
+	h.recordPraxisTrace(ctx, "list_attention", started, map[string]any{"count": len(items)})
 	var parts []string
 	for _, item := range items {
 		title, _ := item["title"].(string)
@@ -175,15 +186,18 @@ func (listChangesCapability) aliases() []string {
 }

 func (listChangesCapability) handle(ctx context.Context, h *reactiveHandler, px *praxisClient, _ router.Decision) string {
+	started := h.now()
 	changes, err := px.ListChanges(ctx, 20)
 	if err != nil {
 		log.Printf("ecosystem: praxis changes: %v", err)
+		h.recordEcosystemTrace(ctx, "praxis", "list_changes", traceStatusForError(err),
+			started, traceErrorFields(err))
 		return "не могу сейчас узнать об изменениях."
 	}
 	if len(changes) == 0 {
 		return "нет изменений."
 	}
-	h.recordPraxisTrace(ctx, "list_changes", map[string]any{"count": len(changes)})
+	h.recordPraxisTrace(ctx, "list_changes", started, map[string]any{"count": len(changes)})
 	var parts []string
 	for _, c := range changes {
 		title, _ := c["title"].(string)
@@ -204,8 +218,10 @@ func (listChangesCapability) handle(ctx context.Context, h *reactiveHandler, px
 // gets one answer across both stores.
 type entityAttentionCapability struct{}

+// aliases are matched against Slots.Fn, which carries a function slot from the
+// act grammar and never free Russian, so only grammar names belong here.
 func (entityAttentionCapability) aliases() []string {
-	return []string{"entity_attention", "что с", "как дела у", "статус"}
+	return []string{"entity_attention", "entity_status"}
 }

 func (entityAttentionCapability) handle(ctx context.Context, h *reactiveHandler, px *praxisClient, dec router.Decision) string {
@@ -222,9 +238,18 @@ func (entityAttentionCapability) handle(ctx context.Context, h *reactiveHandler,
 		return "не могу связать это с сущностью — Nexus не настроен."
 	}

+	started := h.now()
 	entityID, displayName, ambiguous, err := h.ecosystem.resolveEntityReference(ctx, subject, nil)
 	if err != nil {
-		log.Printf("ecosystem: entity attention resolve %q: %v", subject, err)
+		// The subject is his words, so the log gets the same redaction the
+		// trace gets. A trace that stores a rune count next to a log line
+		// storing the runes is not redacted at all.
+		log.Printf("ecosystem: entity attention resolve %s: %v", redactSubject(subject), err)
+		h.recordEcosystemTrace(ctx, "nexus", "resolve", traceStatusForError(err), started,
+			mergeFields(traceErrorFields(err), map[string]any{"subject": redactSubject(subject)}))
+		if unauthorizedEcosystemError(err) {
+			return "экосистема отклоняет доступ, проверь токен."
+		}
 		return "экосистема недоступна, попробуй ещё раз."
 	}
 	if len(ambiguous) > 0 {
@@ -237,12 +262,26 @@ func (entityAttentionCapability) handle(ctx context.Context, h *reactiveHandler,
 		displayName = subject
 	}

+	queried := h.now()
 	items, err := px.ListAttentionForEntity(ctx, entityID, 20)
 	if err != nil {
 		log.Printf("ecosystem: praxis attention for %s: %v", entityID, err)
+		h.recordEcosystemTrace(ctx, "praxis", "entity_attention", traceStatusForError(err),
+			queried, mergeFields(traceErrorFields(err), map[string]any{"entity_id": entityID}))
 		return "не могу сейчас узнать, что требует внимания по «" + displayName + "»."
 	}
-	h.recordPraxisTrace(ctx, "entity_attention", map[string]any{
+	items, scoped := scopedToEntity(items, entityID)
+	if !scoped {
+		// A Praxis old enough to ignore an unknown query parameter answers the
+		// scoped question with the unscoped list. Reading that back as "по
+		// «X»: ..." is the exact fabrication the entity ref exists to prevent,
+		// so refuse the answer instead of relabelling someone else's items.
+		log.Printf("ecosystem: praxis returned unscoped items for %s, refusing to answer", entityID)
+		h.recordEcosystemTrace(ctx, "praxis", "entity_attention", traceFailed, queried,
+			map[string]any{"entity_id": entityID, "class": "unscoped_response"})
+		return "не могу сейчас узнать, что требует внимания по «" + displayName + "»."
+	}
+	h.recordPraxisTrace(ctx, "entity_attention", queried, map[string]any{
 		"entity_id": entityID, "count": len(items),
 	})

@@ -269,6 +308,40 @@ func (entityAttentionCapability) handle(ctx context.Context, h *reactiveHandler,
 	return "по «" + displayName + "»: " + strings.Join(parts, "; ")
 }

+// scopedToEntity drops items that carry an entity_id other than the one asked
+// about, and reports whether the response can be trusted as scoped at all. An
+// item without an entity_id is kept only when at least one sibling carries the
+// matching id: a whole page with no entity_id is a Praxis that ignored the
+// scope, not a page of untagged items.
+func scopedToEntity(items []map[string]any, entityID string) ([]map[string]any, bool) {
+	if len(items) == 0 {
+		return items, true
+	}
+	var kept []map[string]any
+	var sawMatch, sawMismatch bool
+	for _, item := range items {
+		id, _ := item["entity_id"].(string)
+		switch {
+		case id == entityID:
+			sawMatch = true
+			kept = append(kept, item)
+		case id != "":
+			sawMismatch = true
+		default:
+			kept = append(kept, item)
+		}
+	}
+	if sawMatch {
+		return kept, true
+	}
+	if sawMismatch {
+		// Some items were tagged and none matched: the far side answered about
+		// other entities, so nothing here belongs to this one.
+		return nil, true
+	}
+	return nil, false
+}
+
 // 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.
@@ -276,11 +349,18 @@ func (h *reactiveHandler) localFactsForEntity(ctx context.Context, entityID stri
 	if h.dataStore == nil || entityID == "" {
 		return ""
 	}
-	facts, err := h.dataStore.FactsByEntity(ctx, entityID, 3)
+	const spoken = 3
+	// One over the spoken limit, so a truncation can be named rather than
+	// passed off as everything she knows.
+	facts, err := h.dataStore.FactsByEntity(ctx, entityID, spoken+1)
 	if err != nil {
 		log.Printf("ecosystem: facts by entity %s: %v", entityID, err)
 		return ""
 	}
+	more := false
+	if len(facts) > spoken {
+		facts, more = facts[:spoken], true
+	}
 	var parts []string
 	for _, f := range facts {
 		if f.Value != "" {
@@ -290,7 +370,11 @@ func (h *reactiveHandler) localFactsForEntity(ctx context.Context, entityID stri
 	if len(parts) == 0 {
 		return ""
 	}
-	return "я помню: " + strings.Join(parts, ", ")
+	out := "я помню: " + strings.Join(parts, ", ")
+	if more {
+		out += ", и это не всё"
+	}
+	return out
 }

 // mergeFields overlays b onto a and returns a.
@@ -301,30 +385,11 @@ func mergeFields(a, b map[string]any) map[string]any {
 	return a
 }

-// 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.
-func (h *reactiveHandler) recordPraxisTrace(ctx context.Context, operation string, details map[string]any) {
-	now := h.now()
-	value := operation
-	if len(details) > 0 {
-		if b, err := json.Marshal(details); err == nil {
-			value = operation + " " + string(b)
-		}
-	}
-	// Kind must be one of the store's allowed kinds ('self','env','config');
-	// "system" was silently rejected by the CHECK constraint, so no praxis
-	// trace was ever persisted.
-	if _, err := h.api.WriteFact(ctx, ipc.WriteFactReq{
-		Ts:         now,
-		Kind:       "env",
-		Key:        "praxis:" + operation,
-		Value:      value,
-		Source:     "praxis:trace",
-		Confidence: 1.0,
-	}); err != nil {
-		log.Printf("ecosystem: record praxis trace %s: %v", operation, err)
-	}
+// recordPraxisTrace — records a completed Praxis call. Thin wrapper over
+// recordEcosystemTrace so every ecosystem hop lands in one table with one
+// shape.
+func (h *reactiveHandler) recordPraxisTrace(ctx context.Context, operation string, started time.Time, details map[string]any) {
+	h.recordEcosystemTrace(ctx, "praxis", operation, traceOK, started, details)
 }

 // traceStatus classifies an ecosystem call for the trace record. Kept coarse
@@ -332,12 +397,25 @@ func (h *reactiveHandler) recordPraxisTrace(ctx context.Context, operation strin
 // it take", not to re-derive the error.
 const (
 	traceOK       = "ok"
-	traceFailed   = "failed"
+	traceFailed   = "failed"  // the call never got an answer
 	traceRefused  = "refused" // the far side answered, and said no
 	traceAmbig    = "ambiguous"
 	traceNotFound = "not_found"
+	tracePending  = "pending" // deliberately not done yet, awaiting a confirm
 )

+// traceStatusForError distinguishes "I could not reach it" from "it answered
+// and refused". Both degrade the same way for him and not at all the same way
+// for whoever reads the trace: one is a network or a dead service, the other
+// is a token, a version or a rejected argument.
+func traceStatusForError(err error) string {
+	var ee *ecosystemError
+	if errors.As(err, &ee) && !ee.Unreachable() {
+		return traceRefused
+	}
+	return traceFailed
+}
+
 // redactSubject reduces a user utterance to something safe to persist in a
 // trace: its length only. Traces are diagnostics, and his words are not
 // diagnostics — the correlation ID is what ties a trace to the turn.
@@ -347,38 +425,55 @@ func redactSubject(s string) string {

 // recordEcosystemTrace writes one hop of a cross-service call: which service,
 // which operation, the outcome, how long it took, and the correlation ID that
-// stitches the hops together. Unlike recordPraxisTrace it is written for every
-// outcome, not only success — an unrecorded failure is exactly the hop you
-// need when something went wrong at 3am.
+// stitches the hops together. It is written for every outcome, not only
+// success — an unrecorded failure is exactly the hop you need when something
+// went wrong at 3am.
+//
+// Traces go to their own store table, never to facts. One act turn produces
+// three or four of them, at machine rate, while facts arrive at human rate:
+// sharing the table meant the habit profile's 2000-row window, memeval's
+// prompt snapshot and the /dash and /history pages all filled with traces and
+// stopped seeing his actual facts.
 func (h *reactiveHandler) recordEcosystemTrace(ctx context.Context, service, op, status string, started time.Time, fields map[string]any) {
-	details := map[string]any{
-		"service":     service,
-		"operation":   op,
-		"status":      status,
-		"duration_ms": h.now().Sub(started).Milliseconds(),
+	if h.dataStore == nil {
+		return
 	}
-	if id := correlationIDFromCtx(ctx); id != "" {
-		details["correlation_id"] = id
+	tr := store.EcosystemTrace{
+		Ts:            h.now(),
+		Service:       service,
+		Operation:     op,
+		Status:        status,
+		DurationMs:    h.now().Sub(started).Milliseconds(),
+		CorrelationID: correlationIDFromCtx(ctx),
+		Fields:        map[string]any{},
 	}
 	for k, v := range fields {
-		details[k] = v
+		switch k {
+		case "causation_id":
+			tr.CausationID, _ = v.(string)
+		case "http_status":
+			if n, ok := v.(int); ok {
+				tr.HTTPStatus = n
+				continue
+			}
+			tr.Fields[k] = v
+		default:
+			tr.Fields[k] = v
+		}
 	}
-	value := service + ":" + op + " " + status
-	if b, err := json.Marshal(details); err == nil {
-		value = value + " " + string(b)
-	}
-	if _, err := h.api.WriteFact(ctx, ipc.WriteFactReq{
-		Ts:         h.now(),
-		Kind:       "env",
-		Key:        "ecosystem:" + service + ":" + op,
-		Value:      value,
-		Source:     "ecosystem:trace",
-		Confidence: 1.0,
-	}); err != nil {
+	if _, err := h.dataStore.WriteEcosystemTrace(ctx, tr); err != nil {
 		log.Printf("ecosystem: record trace %s:%s: %v", service, op, err)
 	}
 }

+// unauthorizedEcosystemError reports a credential the far side rejected. It
+// gets its own reply: a missing or wrong token looks exactly like an outage to
+// him, and "try again" is advice that will never work.
+func unauthorizedEcosystemError(err error) bool {
+	var ee *ecosystemError
+	return errors.As(err, &ee) && ee.Unauthorized()
+}
+
 // traceErrorFields describes an ecosystemError for a trace without leaking the
 // payload: the HTTP status and the failure class, nothing else.
 func traceErrorFields(err error) map[string]any {
@@ -422,8 +517,11 @@ func (h *reactiveHandler) handleHexisAct(ctx context.Context, dec router.Decisio
 	started := h.now()
 	entityID, displayName, ambiguous, err := h.ecosystem.resolveEntityReference(ctx, dec.Slots.Text, nil)
 	if err != nil {
-		h.recordEcosystemTrace(ctx, "nexus", "resolve", traceFailed, started,
+		h.recordEcosystemTrace(ctx, "nexus", "resolve", traceStatusForError(err), started,
 			mergeFields(traceErrorFields(err), map[string]any{"subject": redactSubject(dec.Slots.Text)}))
+		if unauthorizedEcosystemError(err) {
+			return "экосистема отклоняет доступ, проверь токен."
+		}
 		// A genuine Nexus dependency failure, not "no such entity" — stop here
 		// and report degradation rather than silently falling through to the
 		// local command executor (ECOSYSTEM-SPEC.md: services degrade
@@ -449,8 +547,11 @@ func (h *reactiveHandler) handleHexisAct(ctx context.Context, dec router.Decisio
 	discovered := h.now()
 	caps, err := h.ecosystem.discoverCapabilities(ctx, entityID)
 	if err != nil {
-		h.recordEcosystemTrace(ctx, "hexis", "capabilities", traceFailed, discovered,
+		h.recordEcosystemTrace(ctx, "hexis", "capabilities", traceStatusForError(err), discovered,
 			mergeFields(traceErrorFields(err), map[string]any{"entity_id": entityID}))
+		if unauthorizedEcosystemError(err) {
+			return "экосистема отклоняет доступ, проверь токен."
+		}
 		return "экосистема недоступна, попробуй ещё раз."
 	}
 	h.recordEcosystemTrace(ctx, "hexis", "capabilities", traceOK, discovered,
@@ -499,7 +600,7 @@ func (h *reactiveHandler) handleHexisAct(ctx context.Context, dec router.Decisio
 			expiry:       h.now().Add(confirmTTL),
 		}
 		h.mu.Unlock()
-		h.recordEcosystemTrace(ctx, "hexis", "confirmation", "pending", h.now(),
+		h.recordEcosystemTrace(ctx, "hexis", "confirmation", tracePending, started,
 			map[string]any{"entity_id": entityID, "capability": matched.Name})
 		return "выполнить «" + matched.Name + "» для " + displayName + "? скажи «да» или «нет»."
 	}
@@ -517,20 +618,17 @@ func (h *reactiveHandler) execHexis(ctx context.Context, capID, capName, entityI
 	traced := withCorrelationID(ctx, correlationID)
 	if err != nil {
 		log.Printf("ecosystem: hexis execute error (cor=%s): %v", correlationID, err)
-		h.recordEcosystemTrace(traced, "hexis", "execute", traceFailed, started,
+		h.recordEcosystemTrace(traced, "hexis", "execute", traceStatusForError(err), started,
 			mergeFields(traceErrorFields(err), map[string]any{
 				"entity_id": entityID, "capability": capName, "causation_id": causationID,
 			}))
 		return "не получилось выполнить команду для " + displayName + "."
 	}
+	// One record per hop: the second write this used to make said the same
+	// thing under a different key, in a different shape.
 	h.recordEcosystemTrace(traced, "hexis", "execute", traceOK, started, map[string]any{
-		"entity_id": entityID, "capability": capName, "causation_id": causationID,
-	})
-	h.recordPraxisTrace(ctx, "hexis:"+capName, map[string]any{
-		"entity_id":      entityID,
-		"entity_name":    displayName,
-		"capability":     capName,
-		"correlation_id": correlationID,
+		"entity_id": entityID, "entity_name": displayName,
+		"capability": capName, "causation_id": causationID,
 	})
 	return "команда выполнена для " + displayName + "."
 }