Maven/cmd/mavend/ecosystem_trace_test.go

package main

import (
	"context"
	"encoding/json"
	"strings"
	"testing"

	"github.com/kami/maven/internal/store"
)

// Versioning, authentication and tracing of ecosystem calls (Vikunja #273).

func ecoTraces(t *testing.T, h *reactiveHandler) []map[string]any {
	t.Helper()
	facts, err := h.dataStore.RecentFacts(context.Background(), 100)
	if err != nil {
		t.Fatalf("read facts: %v", err)
	}
	var out []map[string]any
	for _, f := range facts {
		if f.Source != "ecosystem:trace" {
			continue
		}
		i := strings.Index(f.Value, "{")
		if i < 0 {
			t.Fatalf("trace fact carries no detail object: %q", f.Value)
		}
		var d map[string]any
		if err := json.Unmarshal([]byte(f.Value[i:]), &d); err != nil {
			t.Fatalf("decode trace %q: %v", f.Value, err)
		}
		out = append(out, d)
	}
	return out
}

func findTrace(traces []map[string]any, service, op string) map[string]any {
	for _, d := range traces {
		if d["service"] == service && d["operation"] == op {
			return d
		}
	}
	return nil
}

// TestEcosystemHeaders_VersionRequesterAndAuth: every outgoing request carries
// the contract version, the requester, and the bearer token when configured.
func TestEcosystemHeaders_VersionRequesterAndAuth(t *testing.T) {
	ctx := context.Background()
	nexus := newFakeNexus(t, fixtureNexusResolved("ent_muzick", "Muzick indexer", "service"))
	praxis := newFakePraxis(t, fixturePraxisAttentionItems())
	h := ecoHandler(t, nexus, praxis, nil)
	h.ecosystem.nexus = newNexusClient(nexus.URL).withToken("nexus-secret")
	h.ecosystem.praxis = newPraxisClient(praxis.URL).withToken("praxis-secret")

	_, _, _, err := h.ecosystem.resolveEntityReference(ctx, "muzick indexer", nil)
	if err != nil {
		t.Fatalf("resolve: %v", err)
	}
	if _, err := h.ecosystem.praxis.ListAttention(ctx, 5); err != nil {
		t.Fatalf("attention: %v", err)
	}

	for _, tc := range []struct {
		fs            *fakeServer
		versionHeader string
		token         string
	}{
		{nexus, "X-Nexus-Version", "nexus-secret"},
		{praxis, "X-Praxis-Version", "praxis-secret"},
	} {
		reqs := tc.fs.Requests()
		if len(reqs) == 0 {
			t.Fatalf("%s: no request captured", tc.versionHeader)
		}
		r := reqs[0]
		if got := r.Header.Get(tc.versionHeader); got != ecosystemAPIVersion {
			t.Errorf("%s = %q, want %q", tc.versionHeader, got, ecosystemAPIVersion)
		}
		if got := r.Header.Get("X-Requested-By"); got != mavenRequester {
			t.Errorf("X-Requested-By = %q, want %q", got, mavenRequester)
		}
		if got := r.Header.Get("Authorization"); got != "Bearer "+tc.token {
			t.Errorf("Authorization = %q, want bearer %q", got, tc.token)
		}
		if r.Header.Get("X-Correlation-ID") == "" {
			t.Errorf("%s: missing correlation ID", tc.versionHeader)
		}
	}
}

// TestEcosystemHeaders_NoTokenSendsNoAuth: an unconfigured token means the
// transport is trusted, not that a bogus header is sent.
func TestEcosystemHeaders_NoTokenSendsNoAuth(t *testing.T) {
	ctx := context.Background()
	nexus := newFakeNexus(t, fixtureNexusResolved("ent_muzick", "Muzick indexer", "service"))
	h := ecoHandler(t, nexus, nil, nil)

	if _, _, _, err := h.ecosystem.resolveEntityReference(ctx, "muzick indexer", nil); err != nil {
		t.Fatalf("resolve: %v", err)
	}
	if got := nexus.Requests()[0].Header.Get("Authorization"); got != "" {
		t.Fatalf("unauthenticated client must send no Authorization header, got %q", got)
	}
}

// TestEcosystemError_ClassifiesRefusals: callers must be able to tell a
// rejected credential from a version refusal from an unreachable service
// without matching on message text.
func TestEcosystemError_ClassifiesRefusals(t *testing.T) {
	ctx := context.Background()
	for _, tc := range []struct {
		name    string
		status  int
		check   func(*ecosystemError) bool
		wantCls string
	}{
		{"unauthorized", 401, (*ecosystemError).Unauthorized, "unauthorized"},
		{"forbidden", 403, (*ecosystemError).Unauthorized, "unauthorized"},
		{"contract", 426, (*ecosystemError).ContractMismatch, "contract_mismatch"},
	} {
		t.Run(tc.name, func(t *testing.T) {
			nexus := newFakeNexus(t, fixtureNexusResolved("ent_x", "X", "service"))
			nexus.SetFault(tc.status)
			c := newNexusClient(nexus.URL)
			_, err := c.Resolve(ctx, "x", nil)
			ee, ok := err.(*ecosystemError)
			if !ok {
				t.Fatalf("expected *ecosystemError, got %T (%v)", err, err)
			}
			if ee.Service != "nexus" || ee.Status != tc.status {
				t.Fatalf("unexpected typed error %+v", ee)
			}
			if !tc.check(ee) {
				t.Fatalf("%s not classified: %+v", tc.name, ee)
			}
			if got := traceErrorFields(err)["class"]; got != tc.wantCls {
				t.Fatalf("trace class = %v, want %s", got, tc.wantCls)
			}
		})
	}
}

func TestEcosystemError_UnreachableHasNoStatus(t *testing.T) {
	c := newNexusClient("http://127.0.0.1:1")
	_, err := c.Resolve(context.Background(), "x", nil)
	ee, ok := err.(*ecosystemError)
	if !ok {
		t.Fatalf("expected *ecosystemError, got %T", err)
	}
	if !ee.Unreachable() || ee.Unauthorized() || ee.ContractMismatch() {
		t.Fatalf("a refused connection must classify as unreachable only: %+v", ee)
	}
}

// TestEcosystemTrace_SuccessfulActionTracesEveryHop: resolution, discovery and
// execution each leave a record sharing one correlation chain, with timing and
// status, and execution carries the causation link back to the resolve.
func TestEcosystemTrace_SuccessfulActionTracesEveryHop(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)

	if reply := h.handleHexisAct(ctx, actDec("muzick indexer")); !strings.Contains(reply, "выполнена") {
		t.Fatalf("setup: expected success, got %q", reply)
	}

	traces := ecoTraces(t, h)
	for _, want := range [][2]string{{"nexus", "resolve"}, {"hexis", "capabilities"}, {"hexis", "execute"}} {
		d := findTrace(traces, want[0], want[1])
		if d == nil {
			t.Fatalf("missing trace for %s %s, got %+v", want[0], want[1], traces)
		}
		if d["status"] != traceOK {
			t.Errorf("%s %s status = %v, want ok", want[0], want[1], d["status"])
		}
		if _, ok := d["duration_ms"]; !ok {
			t.Errorf("%s %s trace has no timing", want[0], want[1])
		}
		if d["correlation_id"] == nil || d["correlation_id"] == "" {
			t.Errorf("%s %s trace has no correlation id", want[0], want[1])
		}
	}
	exec := findTrace(traces, "hexis", "execute")
	if exec["causation_id"] == nil || exec["causation_id"] == "" {
		t.Error("execute trace must carry the causation id of the turn that caused it")
	}
	if exec["correlation_id"] == exec["causation_id"] {
		t.Error("execute correlation and causation must be distinguishable")
	}
}

// TestEcosystemTrace_FailuresAreTracedToo: the whole point of the change —
// a failed hop is exactly the one worth having recorded.
func TestEcosystemTrace_FailuresAreTracedToo(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.SetFault(401)

	_ = h.handleHexisAct(ctx, actDec("muzick indexer"))

	d := findTrace(ecoTraces(t, h), "nexus", "resolve")
	if d == nil {
		t.Fatal("a failed resolve must still be traced")
	}
	if d["status"] != traceFailed {
		t.Errorf("status = %v, want failed", d["status"])
	}
	if d["class"] != "unauthorized" {
		t.Errorf("class = %v, want unauthorized", d["class"])
	}
	if d["http_status"] != float64(401) {
		t.Errorf("http_status = %v, want 401", d["http_status"])
	}
}

// TestEcosystemTrace_RedactsTheUtterance: traces are diagnostics, his words
// are not. The subject must never be persisted verbatim.
func TestEcosystemTrace_RedactsTheUtterance(t *testing.T) {
	ctx := context.Background()
	nexus := newFakeNexus(t, fixtureNexusNotFound())
	h := ecoHandler(t, nexus, nil, nil)

	_ = h.handleHexisAct(ctx, actDec("перезапусти кофемашину"))

	facts, err := h.dataStore.RecentFacts(ctx, 100)
	if err != nil {
		t.Fatalf("read facts: %v", err)
	}
	var traced []store.Fact
	for _, f := range facts {
		if f.Source == "ecosystem:trace" {
			traced = append(traced, f)
		}
		if strings.Contains(f.Value, "кофемашину") && f.Source == "ecosystem:trace" {
			t.Fatalf("trace leaked the utterance: %q", f.Value)
		}
	}
	if len(traced) == 0 {
		t.Fatal("expected a not_found resolve trace")
	}
	d := findTrace(ecoTraces(t, h), "nexus", "resolve")
	if d["status"] != traceNotFound {
		t.Errorf("status = %v, want not_found", d["status"])
	}
	if d["subject"] != redactSubject("перезапусти кофемашину") {
		t.Errorf("subject = %v, want a redacted length", d["subject"])
	}
}

// TestEcosystemTrace_AmbiguityAndConfirmationAreRecorded: the two moments
// where Maven deliberately does not act still leave a trail.
func TestEcosystemTrace_AmbiguityAndConfirmationAreRecorded(t *testing.T) {
	ctx := context.Background()
	ambig := 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, ambig, nil, hexis)
	_ = h.handleHexisAct(ctx, actDec("muzick"))
	if d := findTrace(ecoTraces(t, h), "nexus", "resolve"); d == nil || d["status"] != traceAmbig {
		t.Fatalf("ambiguous resolve must be traced as such, got %+v", d)
	}

	nexus := newFakeNexus(t, fixtureNexusResolved("ent_muzick", "Muzick indexer", "service"))
	mutating := fixtureHexisCapabilities(map[string]any{"id": "cap_restart", "name": "restart", "read_only": false})
	h2 := ecoHandler(t, nexus, nil, newFakeHexis(t, mutating, fixtureHexisExecuted("exec_1", "succeeded")))
	_ = h2.handleHexisAct(ctx, actDec("restart"))
	if d := findTrace(ecoTraces(t, h2), "hexis", "confirmation"); d == nil || d["status"] != "pending" {
		t.Fatalf("a parked confirmation must be traced, got %+v", d)
	}
}