test: make the ecosystem fault suite fail when the feature is deleted
Several assertions passed against code with the behaviour removed. The independent-outage test shared no state to begin with, the capability fixture used to prove read-only filtering was already mutating, and route-level faults were simulated with a separate fake instead of the shared one. The harness now takes per-route faults and a ticking clock, so durations are measurable and one dead endpoint can be shown not to mute a whole service. New cases cover a resolved reference with no entity, a rejected credential, a malformed Praxis body, foreign items in a scoped response, named truncation, traces staying out of facts, and enrichment making progress while its oldest batch is backed off. Found in review of #82.
This commit is contained in:
@@ -2,7 +2,6 @@ package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"net/http"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
@@ -30,7 +29,7 @@ import (
|
||||
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))
|
||||
clock := newTickingClock(time.Date(2026, 8, 1, 9, 0, 0, 0, time.UTC), time.Millisecond)
|
||||
w := &ecosystemWiring{}
|
||||
if nexus != nil {
|
||||
w.nexus = newNexusClient(nexus.URL)
|
||||
@@ -49,31 +48,43 @@ func ecoHandler(t *testing.T, nexus, praxis, hexis *fakeServer) *reactiveHandler
|
||||
}
|
||||
}
|
||||
|
||||
func traceFacts(t *testing.T, h *reactiveHandler) []store.Fact {
|
||||
// traces reads the ecosystem trace table. Traces live there and not in facts,
|
||||
// so a bounded reader of facts never fills up with machine-rate rows.
|
||||
func traces(t *testing.T, h *reactiveHandler) []store.EcosystemTrace {
|
||||
t.Helper()
|
||||
facts, err := h.dataStore.RecentFacts(context.Background(), 50)
|
||||
out, err := h.dataStore.RecentEcosystemTraces(context.Background(), 100)
|
||||
if err != nil {
|
||||
t.Fatalf("read facts: %v", err)
|
||||
t.Fatalf("read traces: %v", err)
|
||||
}
|
||||
var out []store.Fact
|
||||
for _, f := range facts {
|
||||
if f.Source == "praxis:trace" {
|
||||
out = append(out, f)
|
||||
return out
|
||||
}
|
||||
|
||||
// tracesFor returns the traces recorded for one service+operation.
|
||||
func tracesFor(t *testing.T, h *reactiveHandler, service, op string) []store.EcosystemTrace {
|
||||
t.Helper()
|
||||
var out []store.EcosystemTrace
|
||||
for _, tr := range traces(t, h) {
|
||||
if tr.Service == service && tr.Operation == op {
|
||||
out = append(out, tr)
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// restartCaps is a read-only capability. Restarting a service is a mutation,
|
||||
// so the read-only one this suite runs through the happy paths is named for
|
||||
// what it is; the mutating restart lives in the confirmation tests.
|
||||
func restartCaps() string {
|
||||
return fixtureHexisCapabilities(map[string]any{
|
||||
"id": "cap_restart", "name": "restart", "read_only": true,
|
||||
"id": "cap_status", "name": "restart status", "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) {
|
||||
// TestEcosystem_OutagesLeaveNoSharedFailureState: the two act paths share a
|
||||
// handler, a store and a clock, so what is worth asserting is that a failure
|
||||
// on one leaves nothing behind that degrades the other. Faulting one disjoint
|
||||
// call graph and exercising the other only tests the call graph.
|
||||
func TestEcosystem_OutagesLeaveNoSharedFailureState(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
nexus := newFakeNexus(t, fixtureNexusResolved("ent_muzick", "Muzick indexer", "service"))
|
||||
praxis := newFakePraxis(t, fixturePraxisAttentionItems(map[string]any{
|
||||
@@ -82,17 +93,112 @@ func TestEcosystem_OutagesAreIndependent(t *testing.T) {
|
||||
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)
|
||||
// A Nexus outage during a Hexis act writes a failure trace, and a shared
|
||||
// store is the one thing the Praxis path could inherit it through.
|
||||
nexus.SetFault(503)
|
||||
if reply := h.handleHexisAct(ctx, actDec("muzick indexer")); strings.Contains(reply, "выполнена") {
|
||||
t.Fatalf("nexus outage must not report success, got %q", reply)
|
||||
}
|
||||
if len(tracesFor(t, h, "nexus", "resolve")) == 0 {
|
||||
t.Fatal("the failed resolve must be recorded")
|
||||
}
|
||||
|
||||
praxis.SetFault(0)
|
||||
hexis.SetFault(503)
|
||||
nexus.SetFault(503)
|
||||
nexus.SetFault(0)
|
||||
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)
|
||||
t.Fatalf("a recorded nexus failure must not degrade the praxis digest, got %q", reply)
|
||||
}
|
||||
if got := tracesFor(t, h, "praxis", "list_attention"); len(got) != 1 || got[0].Status != traceOK {
|
||||
t.Fatalf("the praxis digest must trace its own success, got %+v", got)
|
||||
}
|
||||
|
||||
// And the reverse: a Praxis outage mid-session leaves the Hexis path whole.
|
||||
praxis.SetFault(503)
|
||||
if reply := h.handlePraxisAct(ctx, praxisActDec("list_attention")); strings.Contains(reply, "disk") {
|
||||
t.Fatalf("praxis outage must not serve content, got %q", reply)
|
||||
}
|
||||
if reply := h.handleHexisAct(ctx, actDec("muzick indexer")); !strings.Contains(reply, "выполнена") {
|
||||
t.Fatalf("a praxis outage must not block the hexis path, got %q", reply)
|
||||
}
|
||||
}
|
||||
|
||||
// TestEcosystem_OneEndpointDownDoesNotMuteTheService: real outages are usually
|
||||
// partial. Attention answering while surface is down must still deliver.
|
||||
func TestEcosystem_OneEndpointDownDoesNotMuteTheService(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.SetRouteFault("/api/v1/tools/surface", 503)
|
||||
reply := h.handlePraxisAct(ctx, praxisActDec("list_attention"))
|
||||
if !strings.Contains(reply, "disk almost full") {
|
||||
t.Fatalf("a downed surface endpoint must not mute the digest, got %q", reply)
|
||||
}
|
||||
if praxis.Count("POST", "/api/v1/tools/surface") == 0 {
|
||||
t.Fatal("expected the surface attempt")
|
||||
}
|
||||
}
|
||||
|
||||
// TestEcosystem_ResolvedWithoutEntityFailsClosed: the contract violation that
|
||||
// decodes cleanly. Nexus says "resolved" and delivers no entity; treating that
|
||||
// as "no such entity" put the user's verb through to the local executor.
|
||||
func TestEcosystem_ResolvedWithoutEntityFailsClosed(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
nexus := newFakeNexus(t, fixtureNexusResolvedEmpty())
|
||||
hexis := newFakeHexis(t, restartCaps(), fixtureHexisExecuted("exec_1", "succeeded"))
|
||||
h := ecoHandler(t, nexus, nil, hexis)
|
||||
|
||||
reply := h.handleHexisAct(ctx, actDec("muzick indexer"))
|
||||
if reply == "" {
|
||||
t.Fatal("a resolve with no entity must degrade, not fall through to local execution")
|
||||
}
|
||||
if strings.Contains(reply, "выполнена") {
|
||||
t.Fatalf("a resolve with no entity must not report success, got %q", reply)
|
||||
}
|
||||
if hexis.Count("", "/api/v1") != 0 {
|
||||
t.Fatal("hexis must not be contacted after a contract-violating resolve")
|
||||
}
|
||||
}
|
||||
|
||||
// TestEcosystem_RejectedCredentialSaysSo: 401 and 403 must not read as an
|
||||
// outage. "Try again" is advice that never works for a misconfigured token.
|
||||
func TestEcosystem_RejectedCredentialSaysSo(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
for _, status := range []int{401, 403} {
|
||||
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(status)
|
||||
|
||||
reply := h.handleHexisAct(ctx, actDec("muzick indexer"))
|
||||
if !strings.Contains(reply, "токен") {
|
||||
t.Fatalf("http %d must read as a credential problem, got %q", status, reply)
|
||||
}
|
||||
tr := tracesFor(t, h, "nexus", "resolve")
|
||||
if len(tr) != 1 || tr[0].Status != traceRefused || tr[0].HTTPStatus != status {
|
||||
t.Fatalf("http %d must trace as refused with its status, got %+v", status, tr)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// TestEcosystem_MalformedPraxisBodyDegrades: Praxis has the same decode path
|
||||
// Nexus does, and a 200 carrying garbage there is a dependency failure too.
|
||||
func TestEcosystem_MalformedPraxisBodyDegrades(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.SetBody(`[{"title":`)
|
||||
reply := h.handlePraxisAct(ctx, praxisActDec("list_attention"))
|
||||
if reply == "" {
|
||||
t.Fatal("a malformed praxis body must not answer with silence")
|
||||
}
|
||||
if strings.Contains(reply, "disk almost full") {
|
||||
t.Fatalf("a malformed body must not produce content, got %q", reply)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -168,13 +274,60 @@ func TestEcosystem_ExecutionFailureIsNotSuccess(t *testing.T) {
|
||||
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)
|
||||
for _, tr := range tracesFor(t, h, "hexis", "execute") {
|
||||
if tr.Status == traceOK {
|
||||
t.Fatalf("failed execution must not write a success trace: %+v", tr)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// TestEcosystem_SuccessfulActionWritesATrace is the positive half the failure
|
||||
// assertions above depend on: without it, "no success trace" passes with the
|
||||
// trace writer deleted. It was, for a while — both writers used a fact kind the
|
||||
// store's CHECK constraint rejects and the error was discarded.
|
||||
func TestEcosystem_SuccessfulActionWritesATrace(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)
|
||||
}
|
||||
exec := tracesFor(t, h, "hexis", "execute")
|
||||
if len(exec) != 1 || exec[0].Status != traceOK {
|
||||
t.Fatalf("a successful execution must leave exactly one ok trace, got %+v", exec)
|
||||
}
|
||||
if exec[0].CorrelationID == "" {
|
||||
t.Error("a trace with no correlation id cannot be stitched to anything")
|
||||
}
|
||||
}
|
||||
|
||||
// TestEcosystem_TracesStayOutOfFacts: traces are written at machine rate and
|
||||
// facts at human rate. One act turn used to write four fact rows, which pushed
|
||||
// his facts out of every bounded reader (the habit profile's window, memeval's
|
||||
// prompt, /dash, /history).
|
||||
func TestEcosystem_TracesStayOutOfFacts(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)
|
||||
}
|
||||
if len(traces(t, h)) == 0 {
|
||||
t.Fatal("setup: expected traces")
|
||||
}
|
||||
facts, err := h.dataStore.RecentFacts(ctx, 100)
|
||||
if err != nil {
|
||||
t.Fatalf("read facts: %v", err)
|
||||
}
|
||||
if len(facts) != 0 {
|
||||
t.Fatalf("an ecosystem act must write no facts at all, got %+v", facts)
|
||||
}
|
||||
}
|
||||
|
||||
// TestEcosystem_AmbiguousTargetBlocksExecution: ambiguity blocks mutation, and
|
||||
// the clarification must name the candidates rather than pick one.
|
||||
func TestEcosystem_AmbiguousTargetBlocksExecution(t *testing.T) {
|
||||
@@ -245,12 +398,10 @@ func TestEcosystem_MutatingCapabilityWaitsForConfirmation(t *testing.T) {
|
||||
// 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"}`),
|
||||
})
|
||||
praxis := newFakePraxis(t, fixturePraxisAttentionItems(
|
||||
map[string]any{"id": "item_1", "title": "disk almost full", "importance": 3.0},
|
||||
))
|
||||
praxis.SetRouteFault("/api/v1/tools/surface", 500)
|
||||
h := ecoHandler(t, nil, praxis, nil)
|
||||
|
||||
reply := h.handlePraxisAct(ctx, praxisActDec("list_attention"))
|
||||
@@ -278,7 +429,7 @@ func TestEcosystem_TotalOutageSaysSoForEveryPath(t *testing.T) {
|
||||
"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")),
|
||||
"acknowledge": h.handlePraxisAct(ctx, praxisItemDec("acknowledge_item", "item_1")),
|
||||
} {
|
||||
if reply == "" {
|
||||
t.Errorf("%s: total outage must not answer with silence", name)
|
||||
@@ -287,8 +438,13 @@ func TestEcosystem_TotalOutageSaysSoForEveryPath(t *testing.T) {
|
||||
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")
|
||||
for _, tr := range traces(t, h) {
|
||||
if tr.Status == traceOK {
|
||||
t.Fatalf("a total outage must not leave success traces behind: %+v", tr)
|
||||
}
|
||||
}
|
||||
if len(tracesFor(t, h, "praxis", "acknowledge")) == 0 {
|
||||
t.Fatal("the acknowledge arm must reach praxis and record the refusal")
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -19,6 +19,13 @@ func praxisActDec(fn string) router.Decision {
|
||||
return router.Decision{Intent: router.IntentAct, Slots: router.Slots{Fn: fn, HasFn: true}}
|
||||
}
|
||||
|
||||
// praxisItemDec is praxisActDec for the lifecycle verbs, which need an item id
|
||||
// in the value slot. Without one they answer "which item?" and never reach
|
||||
// Praxis at all, which makes them useless for testing a Praxis outage.
|
||||
func praxisItemDec(fn, itemID string) router.Decision {
|
||||
return router.Decision{Intent: router.IntentAct, Slots: router.Slots{Fn: fn, HasFn: true, Value: itemID}}
|
||||
}
|
||||
|
||||
func newPraxisTestHandler(t *testing.T, praxis *fakeServer) *reactiveHandler {
|
||||
t.Helper()
|
||||
st := newTestStore(t)
|
||||
|
||||
@@ -59,8 +59,11 @@ func newHexisTestHandler(t *testing.T, resolveBody string, caps string) (*reacti
|
||||
}, executed
|
||||
}
|
||||
|
||||
func actDec(text string) router.Decision {
|
||||
return router.Decision{Intent: router.IntentAct, Slots: router.Slots{Text: text, Fn: "restart", HasFn: true}}
|
||||
// actDec builds an act decision about subject. The verb is always "restart":
|
||||
// the argument is the utterance the entity is resolved from, never the verb,
|
||||
// so actDec("restart") reads as a verb and is not one.
|
||||
func actDec(subject string) router.Decision {
|
||||
return router.Decision{Intent: router.IntentAct, Slots: router.Slots{Text: subject, Fn: "restart", HasFn: true}}
|
||||
}
|
||||
|
||||
func TestHexisMutatingRequiresConfirm(t *testing.T) {
|
||||
|
||||
@@ -28,7 +28,7 @@ func entityAttentionDec(subject string) router.Decision {
|
||||
func TestEntityAttention_ScopesPraxisByCanonicalID(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
nexus := newFakeNexus(t, fixtureNexusResolved("ent_muzick", "Muzick indexer", "service"))
|
||||
praxis := newFakePraxis(t, fixturePraxisAttentionItems(
|
||||
praxis := newFakePraxis(t, fixturePraxisAttentionScoped("ent_muzick",
|
||||
map[string]any{"id": "item_1", "title": "indexer queue is backing up", "importance": 3.0},
|
||||
))
|
||||
h := ecoHandler(t, nexus, praxis, nil)
|
||||
@@ -76,6 +76,76 @@ func TestEntityAttention_FoldsInLocalFactsForSameEntity(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// TestEntityAttention_UnscopedPraxisResponseIsRefused: a Praxis old enough to
|
||||
// ignore the entity_id parameter answers the scoped question with the whole
|
||||
// unscoped list. Relabelling those items "по «X»" is the same fabrication the
|
||||
// canonical ref exists to prevent, arriving through a different door.
|
||||
func TestEntityAttention_UnscopedPraxisResponseIsRefused(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},
|
||||
))
|
||||
h := ecoHandler(t, nexus, praxis, nil)
|
||||
|
||||
reply := h.handlePraxisAct(ctx, entityAttentionDec("muzick indexer"))
|
||||
if strings.Contains(reply, "disk almost full") {
|
||||
t.Fatalf("an unscoped response must not be read back as entity-scoped, got %q", reply)
|
||||
}
|
||||
if reply == "" {
|
||||
t.Fatal("refusing the answer must still say something")
|
||||
}
|
||||
if praxis.Count("POST", "/api/v1/tools/surface") != 0 {
|
||||
t.Error("items that were never spoken must not be surfaced")
|
||||
}
|
||||
}
|
||||
|
||||
// TestEntityAttention_ForeignItemsAreDropped: items tagged with another entity
|
||||
// are dropped rather than spoken under this entity's name.
|
||||
func TestEntityAttention_ForeignItemsAreDropped(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
nexus := newFakeNexus(t, fixtureNexusResolved("ent_muzick", "Muzick indexer", "service"))
|
||||
mixed := []map[string]any{
|
||||
{"id": "item_1", "title": "indexer queue is backing up", "importance": 3.0, "entity_id": "ent_muzick"},
|
||||
{"id": "item_2", "title": "the kettle is descaling", "importance": 1.0, "entity_id": "ent_kettle"},
|
||||
}
|
||||
praxis := newFakePraxis(t, mustJSON(mixed))
|
||||
h := ecoHandler(t, nexus, praxis, nil)
|
||||
|
||||
reply := h.handlePraxisAct(ctx, entityAttentionDec("muzick indexer"))
|
||||
if !strings.Contains(reply, "indexer queue is backing up") {
|
||||
t.Fatalf("the matching item must be spoken, got %q", reply)
|
||||
}
|
||||
if strings.Contains(reply, "kettle") {
|
||||
t.Fatalf("another entity's item must not be spoken here, got %q", reply)
|
||||
}
|
||||
}
|
||||
|
||||
// TestEntityAttention_TruncationIsNamed: reading three of many remembered
|
||||
// facts must not be presented as everything she knows.
|
||||
func TestEntityAttention_TruncationIsNamed(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)
|
||||
|
||||
for i := 0; i < 5; i++ {
|
||||
id, err := h.dataStore.WriteFactAboutSubject(ctx, time.Now(), store.KindEnv,
|
||||
"note", "the espresso machine", "факт "+string(rune('а'+i)), "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, "и это не всё") {
|
||||
t.Fatalf("a truncated recall must say it is truncated, got %q", reply)
|
||||
}
|
||||
}
|
||||
|
||||
// TestEntityAttention_AmbiguousAsksInsteadOfGuessing.
|
||||
func TestEntityAttention_AmbiguousAsksInsteadOfGuessing(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
@@ -145,7 +215,7 @@ func TestEntityAttention_WithoutNexusSaysSo(t *testing.T) {
|
||||
|
||||
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)
|
||||
t.Fatalf("without nexus, 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")
|
||||
@@ -211,3 +281,78 @@ func TestEnrichmentBackoff_GrowsAndIsCapped(t *testing.T) {
|
||||
t.Fatalf("backoff must cap at an hour, got %v", enrichmentBackoff(50))
|
||||
}
|
||||
}
|
||||
|
||||
// TestEnrichment_BackedOffFactsDoNotStallTheQueue: the pending queue is ordered
|
||||
// by id, so the oldest facts are pulled first whether or not they are eligible.
|
||||
// A batch of facts in backoff at the head must not hold every slot and stop
|
||||
// enrichment for everything younger.
|
||||
func TestEnrichment_BackedOffFactsDoNotStallTheQueue(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
st := newTestStore(t)
|
||||
total := 5
|
||||
for i := 0; i < total; i++ {
|
||||
if _, err := st.WriteFactAboutSubject(ctx, time.Now(), store.KindEnv, "likes",
|
||||
"subject-"+string(rune('a'+i)), `"true"`, "infer:pref", 0.8, sql.NullInt64{}); err != nil {
|
||||
t.Fatalf("WriteFactAboutSubject: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
nexus := newFakeNexus(t, fixtureNexusResolved("ent_x", "X", "service"))
|
||||
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
|
||||
// A batch smaller than the queue, so with no scan the last fact never
|
||||
// reaches the head while the first ones are backed off.
|
||||
w.batch = total - 1
|
||||
|
||||
nexus.SetFault(503)
|
||||
w.tick(ctx)
|
||||
if got := nexus.Count("POST", "/api/v1/resolve"); got != total-1 {
|
||||
t.Fatalf("expected the first batch attempted, got %d calls", got)
|
||||
}
|
||||
|
||||
// Second tick with Nexus healthy: the backed-off head must be skipped and
|
||||
// the fact behind it resolved, not the same batch pulled and dropped.
|
||||
nexus.SetFault(0)
|
||||
w.tick(ctx)
|
||||
facts, err := st.FactsByEntity(ctx, "ent_x", 10)
|
||||
if err != nil {
|
||||
t.Fatalf("FactsByEntity: %v", err)
|
||||
}
|
||||
if len(facts) == 0 {
|
||||
t.Fatal("a due fact behind a backed-off batch must still be resolved")
|
||||
}
|
||||
}
|
||||
|
||||
// TestEnrichment_StoreWriteFailureBacksOffToo: the one failure mode where the
|
||||
// resolve worked and the write did not must be paced like any other, not
|
||||
// retried at full rate forever.
|
||||
func TestEnrichment_StoreWriteFailureBacksOffToo(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)
|
||||
}
|
||||
pending, err := st.PendingFactResolutions(ctx, 10)
|
||||
if err != nil || len(pending) != 1 {
|
||||
t.Fatalf("setup: pending = %+v, %v", pending, 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
|
||||
|
||||
// Closing the store makes the resolution write fail while the Nexus call
|
||||
// still succeeds — the split this path gets wrong.
|
||||
if err := st.Close(); err != nil {
|
||||
t.Fatalf("close store: %v", err)
|
||||
}
|
||||
if w.resolveOne(ctx, pending[0]) {
|
||||
t.Fatal("a failed store write must not report success")
|
||||
}
|
||||
if w.due(pending[0].ID) {
|
||||
t.Fatal("a failed store write must back the fact off like a failed resolve")
|
||||
}
|
||||
}
|
||||
|
||||
@@ -27,11 +27,12 @@ type capturedRequest struct {
|
||||
type fakeServer struct {
|
||||
*httptest.Server
|
||||
|
||||
mu sync.Mutex
|
||||
requests []capturedRequest
|
||||
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
|
||||
mu sync.Mutex
|
||||
requests []capturedRequest
|
||||
fault int // non-zero: every request gets this HTTP status instead of routing
|
||||
routeFaults map[string]int // path prefix → status, for one endpoint failing alone
|
||||
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
|
||||
@@ -59,6 +60,14 @@ func newFakeServer(t *testing.T, routes map[string]http.HandlerFunc) *fakeServer
|
||||
Header: r.Header.Clone(),
|
||||
})
|
||||
fault := fs.fault
|
||||
if fault == 0 {
|
||||
for prefix, status := range fs.routeFaults {
|
||||
if hasPrefix(r.URL.Path, prefix) {
|
||||
fault = status
|
||||
break
|
||||
}
|
||||
}
|
||||
}
|
||||
garbage := fs.garbage
|
||||
delay := fs.delay
|
||||
fs.mu.Unlock()
|
||||
@@ -114,6 +123,22 @@ func (fs *fakeServer) SetFault(status int) {
|
||||
fs.fault = status
|
||||
}
|
||||
|
||||
// SetRouteFault fails one endpoint while the rest of the server stays healthy,
|
||||
// which is the shape most real outages take: attention answers and pin is
|
||||
// down. Pass 0 to clear that route. A server-wide SetFault still wins.
|
||||
func (fs *fakeServer) SetRouteFault(pathPrefix string, status int) {
|
||||
fs.mu.Lock()
|
||||
defer fs.mu.Unlock()
|
||||
if fs.routeFaults == nil {
|
||||
fs.routeFaults = map[string]int{}
|
||||
}
|
||||
if status == 0 {
|
||||
delete(fs.routeFaults, pathPrefix)
|
||||
return
|
||||
}
|
||||
fs.routeFaults[pathPrefix] = 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.
|
||||
@@ -198,6 +223,14 @@ func fixtureNexusResolvedFuture(entityID, displayName, entityType string) string
|
||||
})
|
||||
}
|
||||
|
||||
// fixtureNexusResolvedEmpty is the contract violation that decodes cleanly:
|
||||
// Nexus claims a resolve and delivers no entity. It must not read as "no such
|
||||
// entity", which would let the caller fall through to local execution with the
|
||||
// user's verb intact.
|
||||
func fixtureNexusResolvedEmpty() string {
|
||||
return `{"status":"resolved"}`
|
||||
}
|
||||
|
||||
func fixtureNexusNotFound() string {
|
||||
return `{"status":"not_found"}`
|
||||
}
|
||||
@@ -225,6 +258,16 @@ func fixtureHexisExecutionFailed(id, message string) string {
|
||||
return mustJSON(map[string]any{"id": id, "status": "failed", "error": message})
|
||||
}
|
||||
|
||||
// fixturePraxisAttentionScoped tags each item with an entity_id, which is what
|
||||
// a Praxis that understands the entity_id query parameter returns. A Praxis
|
||||
// that ignores it answers with untagged items from every entity.
|
||||
func fixturePraxisAttentionScoped(entityID string, items ...map[string]any) string {
|
||||
for _, item := range items {
|
||||
item["entity_id"] = entityID
|
||||
}
|
||||
return mustJSON(items)
|
||||
}
|
||||
|
||||
func fixturePraxisAttentionItems(items ...map[string]any) string {
|
||||
return mustJSON(items)
|
||||
}
|
||||
@@ -243,18 +286,28 @@ func mustJSON(v any) string {
|
||||
// (e.g. asserting age-based digest ordering without sleeping).
|
||||
|
||||
type fakeClock struct {
|
||||
mu sync.Mutex
|
||||
t time.Time
|
||||
mu sync.Mutex
|
||||
t time.Time
|
||||
step time.Duration // advanced on every read, so elapsed time is measurable
|
||||
}
|
||||
|
||||
func newFakeClock(start time.Time) *fakeClock {
|
||||
return &fakeClock{t: start}
|
||||
}
|
||||
|
||||
// newTickingClock advances by step on every read. Durations measured across
|
||||
// hops are then non-zero without sleeping, which is what lets a test tell a
|
||||
// trace that measured something from one that measured nothing.
|
||||
func newTickingClock(start time.Time, step time.Duration) *fakeClock {
|
||||
return &fakeClock{t: start, step: step}
|
||||
}
|
||||
|
||||
func (c *fakeClock) Now() time.Time {
|
||||
c.mu.Lock()
|
||||
defer c.mu.Unlock()
|
||||
return c.t
|
||||
now := c.t
|
||||
c.t = c.t.Add(c.step)
|
||||
return now
|
||||
}
|
||||
|
||||
func (c *fakeClock) Advance(d time.Duration) {
|
||||
|
||||
Reference in New Issue
Block a user