Maven/cmd/mavend/ecosystem_trace_test.go

package main

import (
	"context"
	"strings"
	"testing"

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

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

func findTrace(t *testing.T, h *reactiveHandler, service, op string) *store.EcosystemTrace {
	t.Helper()
	for _, tr := range traces(t, h) {
		if tr.Service == service && tr.Operation == op {
			found := tr
			return &found
		}
	}
	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)
	}
	// A bare client call carries whatever the caller assigned. Entry points
	// assign the ID, the header layer only reads it, so mirror an action here.
	if _, err := h.ecosystem.praxis.ListAttention(withCorrelationID(ctx, newCorrelationID()), 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")); !actRan(reply) {
		t.Fatalf("setup: expected success, got %q", reply)
	}

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

// TestEcosystemTrace_OneCorrelationIDPerPraxisAction: a digest calls attention
// once and surface once per item. All of it is one turn, so the far side must
// see one ID and not N+1 unrelated ones.
func TestEcosystemTrace_OneCorrelationIDPerPraxisAction(t *testing.T) {
	ctx := context.Background()
	praxis := newFakePraxis(t, fixturePraxisAttentionItems(
		map[string]any{"id": "item_1", "title": "disk almost full", "importance": 3.0},
		map[string]any{"id": "item_2", "title": "backup is stale", "importance": 2.0},
	))
	h := ecoHandler(t, nil, praxis, nil)

	if reply := h.handlePraxisAct(ctx, praxisActDec("list_attention")); !strings.Contains(reply, "disk almost full") {
		t.Fatalf("setup: expected the digest, got %q", reply)
	}

	reqs := praxis.Requests()
	if len(reqs) < 3 {
		t.Fatalf("expected attention plus one surface per item, got %d requests", len(reqs))
	}
	first := reqs[0].Header.Get("X-Correlation-ID")
	if first == "" {
		t.Fatal("every ecosystem request must carry a correlation id")
	}
	for _, r := range reqs {
		if got := r.Header.Get("X-Correlation-ID"); got != first {
			t.Fatalf("%s %s carried %q, want the action's id %q", r.Method, r.Path, got, first)
		}
	}
	tr := findTrace(t, h, "praxis", "list_attention")
	if tr == nil || tr.CorrelationID != first {
		t.Fatalf("the trace must carry the id that was actually sent, got %+v", tr)
	}
}

// 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(t, h, "nexus", "resolve")
	if d == nil {
		t.Fatal("a failed resolve must still be traced")
	}
	if d.Status != traceRefused {
		t.Errorf("status = %v, want refused: the far side answered", d.Status)
	}
	if d.Fields["class"] != "unauthorized" {
		t.Errorf("class = %v, want unauthorized", d.Fields["class"])
	}
	if d.HTTPStatus != 401 {
		t.Errorf("http_status = %v, want 401", d.HTTPStatus)
	}
}

// TestEcosystemTrace_UnreachableIsNotRefused: never got an answer and answered
// with a refusal are different failures, and the trace must say which.
func TestEcosystemTrace_UnreachableIsNotRefused(t *testing.T) {
	ctx := context.Background()
	h := ecoHandler(t, nil, nil, nil)
	h.ecosystem.nexus = newNexusClient("http://127.0.0.1:1")

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

	d := findTrace(t, h, "nexus", "resolve")
	if d == nil {
		t.Fatal("an unreachable resolve must still be traced")
	}
	if d.Status != traceFailed {
		t.Errorf("status = %v, want failed", d.Status)
	}
	if d.Fields["class"] != "unreachable" {
		t.Errorf("class = %v, want unreachable", d.Fields["class"])
	}
}

// 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("перезапусти кофемашину"))

	recorded := traces(t, h)
	if len(recorded) == 0 {
		t.Fatal("expected a not_found resolve trace")
	}
	for _, tr := range recorded {
		for k, v := range tr.Fields {
			if s, ok := v.(string); ok && strings.Contains(s, "кофемашину") {
				t.Fatalf("trace leaked the utterance in %s: %q", k, s)
			}
		}
	}
	d := findTrace(t, h, "nexus", "resolve")
	if d.Status != traceNotFound {
		t.Errorf("status = %v, want not_found", d.Status)
	}
	if d.Fields["subject"] != redactSubject("перезапусти кофемашину") {
		t.Errorf("subject = %v, want a redacted length", d.Fields["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(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"))
	d := findTrace(t, h2, "hexis", "confirmation")
	if d == nil || d.Status != tracePending {
		t.Fatalf("a parked confirmation must be traced, got %+v", d)
	}
	// The confirmation hop is measured from the top of the action, not from
	// the instant it is recorded, which was always zero.
	if d.DurationMs == 0 {
		t.Error("the confirmation trace must report the time the action took to get there")
	}
}