From 927e46bca376f5056185218e1df5b0949083e182 Mon Sep 17 00:00:00 2001 From: kami Date: Sat, 1 Aug 2026 06:57:52 +0400 Subject: [PATCH] Version, authenticate and fully trace ecosystem calls (#273) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Every Nexus and Praxis request now carries the contract version, an X-Requested-By identifying Maven, a correlation ID (generated per request when the call is not part of a traced action), and a bearer token when one is configured. Nexus/Praxis/Hexis config blocks grew an optional token field, env-expandable so the secret stays out of the committed config; the vendored hexis client predates bearer auth, so a configured Hexis token logs a loud warning instead of pretending to authenticate. Client failures are now a typed *ecosystemError carrying service, operation and HTTP status, classifying unauthorized, contract-mismatch and unreachable without matching on message text. Trace records are written for resolution, discovery, confirmation and execution — on failure as well as success — with status, duration, correlation and causation ids, HTTP status and failure class, and the utterance redacted to its length. Traces were never actually persisted before: both trace writers used fact kind "system", which the store's CHECK constraint rejects, and the error was discarded. --- cmd/mavend/ecosystem.go | 129 +++++++++++--- cmd/mavend/ecosystem_acts.go | 129 +++++++++++++- cmd/mavend/ecosystem_trace_test.go | 277 +++++++++++++++++++++++++++++ internal/config/config.go | 12 ++ 4 files changed, 523 insertions(+), 24 deletions(-) create mode 100644 cmd/mavend/ecosystem_trace_test.go diff --git a/cmd/mavend/ecosystem.go b/cmd/mavend/ecosystem.go index 6a7e3a1..c249fcc 100644 --- a/cmd/mavend/ecosystem.go +++ b/cmd/mavend/ecosystem.go @@ -6,6 +6,7 @@ import ( "crypto/rand" "encoding/hex" "encoding/json" + "errors" "fmt" "io" "log" @@ -31,18 +32,82 @@ func correlationIDFromCtx(ctx context.Context) string { return id } -// setEcosystemHeaders stamps the version and correlation headers common to -// every outgoing ecosystem request. -func setEcosystemHeaders(req *http.Request, ctx context.Context, versionHeader string) { +// ecosystemAPIVersion is the contract version Maven speaks to Nexus and +// Praxis. It is sent on every request so a service that has moved on can +// refuse or adapt explicitly instead of misreading an older payload. +const ecosystemAPIVersion = "v1" + +// mavenRequester identifies the calling system on every ecosystem request, so +// a trace on the far side can attribute a call to Maven rather than to an +// anonymous HTTP client. +const mavenRequester = "maven" + +// setEcosystemHeaders stamps the version, requester, auth and correlation +// headers common to every outgoing ecosystem request. token may be empty, +// which means the transport itself is trusted (loopback or unix socket). +func setEcosystemHeaders(req *http.Request, ctx context.Context, versionHeader, token string) { req.Header.Set("Content-Type", "application/json") - req.Header.Set(versionHeader, "v1") - if id := correlationIDFromCtx(ctx); id != "" { - req.Header.Set("X-Correlation-ID", id) + req.Header.Set(versionHeader, ecosystemAPIVersion) + req.Header.Set("Accept", "application/json") + req.Header.Set("X-Requested-By", mavenRequester) + if token != "" { + req.Header.Set("Authorization", "Bearer "+token) + } + id := correlationIDFromCtx(ctx) + if id == "" { + // A call made outside a traced action still gets an ID, so the far + // side's log line can be matched to this one request. + id = newCorrelationID() + } + req.Header.Set("X-Correlation-ID", id) +} + +// ecosystemError is the typed failure every ecosystem client returns, so +// callers can tell a transport failure from a refusal from a contract +// mismatch without matching on message text. The distinction matters: +// "the service is down" and "the service rejected my version" degrade the +// same way to the user but not to whoever reads the trace. +type ecosystemError struct { + Service string // "nexus", "praxis", "hexis" + Op string // logical operation, e.g. "resolve" + Status int // HTTP status, 0 when the call never got an answer + Err error +} + +func (e *ecosystemError) Error() string { + if e.Status != 0 { + return fmt.Sprintf("%s %s: http %d: %v", e.Service, e.Op, e.Status, e.Err) + } + return fmt.Sprintf("%s %s: %v", e.Service, e.Op, e.Err) +} + +func (e *ecosystemError) Unwrap() error { return e.Err } + +// Unauthorized reports a rejected or missing credential. +func (e *ecosystemError) Unauthorized() bool { + return e.Status == http.StatusUnauthorized || e.Status == http.StatusForbidden +} + +// ContractMismatch reports that the far side refused the version Maven speaks. +func (e *ecosystemError) ContractMismatch() bool { + return e.Status == http.StatusNotAcceptable || e.Status == http.StatusUpgradeRequired +} + +// Unreachable reports a call that never produced an HTTP answer at all +// (connection refused, timeout, cancelled). +func (e *ecosystemError) Unreachable() bool { return e.Status == 0 } + +// httpError builds an ecosystemError from a response status. +func httpError(service, op string, status int) *ecosystemError { + return &ecosystemError{ + Service: service, Op: op, Status: status, + Err: errors.New(http.StatusText(status)), } } type nexusClient struct { baseURL string + token string httpClient *http.Client } @@ -53,6 +118,13 @@ func newNexusClient(url string) *nexusClient { } } +// withToken sets the bearer token sent on every request. Returns the client so +// wiring reads as one expression. +func (c *nexusClient) withToken(token string) *nexusClient { + c.token = token + return c +} + type nexusEntity struct { ID string `json:"id"` Type string `json:"type"` @@ -107,22 +179,22 @@ func (c *nexusClient) Resolve(ctx context.Context, query string, types []string) if err != nil { return nil, fmt.Errorf("create request: %w", err) } - setEcosystemHeaders(req, ctx, "X-Nexus-Version") + setEcosystemHeaders(req, ctx, "X-Nexus-Version", c.token) resp, err := c.httpClient.Do(req) if err != nil { - return nil, fmt.Errorf("do request: %w", err) + return nil, &ecosystemError{Service: "nexus", Op: "resolve", Err: err} } defer resp.Body.Close() bodyBytes, _ := io.ReadAll(resp.Body) if resp.StatusCode != 200 { - return nil, fmt.Errorf("nexus: %s", http.StatusText(resp.StatusCode)) + return nil, httpError("nexus", "resolve", resp.StatusCode) } var result nexusResolveResult if err := json.Unmarshal(bodyBytes, &result); err != nil { - return nil, fmt.Errorf("decode: %w", err) + return nil, &ecosystemError{Service: "nexus", Op: "resolve", Status: resp.StatusCode, Err: err} } return &result, nil } @@ -132,7 +204,7 @@ func (c *nexusClient) Health(ctx context.Context) error { if err != nil { return err } - setEcosystemHeaders(req, ctx, "X-Nexus-Version") + setEcosystemHeaders(req, ctx, "X-Nexus-Version", c.token) resp, err := c.httpClient.Do(req) if err != nil { return err @@ -149,6 +221,7 @@ func (c *nexusClient) Health(ctx context.Context) error { // so attention/changes/lifecycle all go over this HTTP contract against praxisd. type praxisClient struct { baseURL string + token string httpClient *http.Client } @@ -159,22 +232,30 @@ func newPraxisClient(url string) *praxisClient { } } +func (c *praxisClient) withToken(token string) *praxisClient { + c.token = token + return c +} + // getJSON performs a GET and decodes the JSON body into out. func (c *praxisClient) getJSON(ctx context.Context, path string, out any) error { req, err := http.NewRequestWithContext(ctx, http.MethodGet, c.baseURL+path, nil) if err != nil { return err } - setEcosystemHeaders(req, ctx, "X-Praxis-Version") + setEcosystemHeaders(req, ctx, "X-Praxis-Version", c.token) resp, err := c.httpClient.Do(req) if err != nil { - return err + return &ecosystemError{Service: "praxis", Op: path, Err: err} } defer resp.Body.Close() if resp.StatusCode != 200 { - return fmt.Errorf("praxis: %s", http.StatusText(resp.StatusCode)) + return httpError("praxis", path, resp.StatusCode) } - return json.NewDecoder(resp.Body).Decode(out) + if err := json.NewDecoder(resp.Body).Decode(out); err != nil { + return &ecosystemError{Service: "praxis", Op: path, Status: resp.StatusCode, Err: err} + } + return nil } func (c *praxisClient) ListAttention(ctx context.Context, limit int) ([]map[string]any, error) { @@ -227,14 +308,14 @@ func (c *praxisClient) postItemAction(ctx context.Context, path, itemID string) if err != nil { return nil, err } - setEcosystemHeaders(req, ctx, "X-Praxis-Version") + setEcosystemHeaders(req, ctx, "X-Praxis-Version", c.token) resp, err := c.httpClient.Do(req) if err != nil { return nil, err } defer resp.Body.Close() if resp.StatusCode != 200 { - return nil, fmt.Errorf("praxis %s: %s", path, http.StatusText(resp.StatusCode)) + return nil, httpError("praxis", path, resp.StatusCode) } var out praxisItem if err := json.NewDecoder(resp.Body).Decode(&out); err != nil { @@ -268,14 +349,14 @@ func (c *praxisClient) Pin(ctx context.Context, itemID string, pinned bool) (*pr if err != nil { return nil, err } - setEcosystemHeaders(req, ctx, "X-Praxis-Version") + setEcosystemHeaders(req, ctx, "X-Praxis-Version", c.token) resp, err := c.httpClient.Do(req) if err != nil { return nil, err } defer resp.Body.Close() if resp.StatusCode != 200 { - return nil, fmt.Errorf("praxis pin: %s", http.StatusText(resp.StatusCode)) + return nil, httpError("praxis", "pin", resp.StatusCode) } var out praxisItem if err := json.NewDecoder(resp.Body).Decode(&out); err != nil { @@ -311,7 +392,7 @@ func wireEcosystem(cfg *config.Config) *ecosystemWiring { // Nexus identity service if cfg.Nexus != nil && cfg.Nexus.URL != "" { - w.nexus = newNexusClient(cfg.Nexus.URL) + w.nexus = newNexusClient(cfg.Nexus.URL).withToken(cfg.Nexus.Token) log.Printf("ecosystem: nexus at %s", cfg.Nexus.URL) } else { log.Printf("ecosystem: nexus not configured") @@ -320,6 +401,12 @@ func wireEcosystem(cfg *config.Config) *ecosystemWiring { // Hexis capability service if cfg.Hexis != nil && cfg.Hexis.URL != "" { w.hexis = hexisclient.New(cfg.Hexis.URL) + if cfg.Hexis.Token != "" { + // Upstream hexis grew Client.WithToken, but the copy vendored + // here predates it, so the token cannot be sent yet. Say so + // loudly rather than pretending the call is authenticated. + log.Printf("ecosystem: hexis token configured but the vendored hexis client cannot send it — re-vendor github.com/kami/hexis to enable bearer auth") + } log.Printf("ecosystem: hexis at %s", cfg.Hexis.URL) } else { log.Printf("ecosystem: hexis not configured") @@ -327,7 +414,7 @@ func wireEcosystem(cfg *config.Config) *ecosystemWiring { // Praxis attention service (HTTP tools API — never the DB directly) if cfg.Praxis != nil && cfg.Praxis.URL != "" { - w.praxis = newPraxisClient(cfg.Praxis.URL) + w.praxis = newPraxisClient(cfg.Praxis.URL).withToken(cfg.Praxis.Token) log.Printf("ecosystem: praxis at %s", cfg.Praxis.URL) } else { log.Printf("ecosystem: praxis not configured") diff --git a/cmd/mavend/ecosystem_acts.go b/cmd/mavend/ecosystem_acts.go index e0c1f72..b0a513d 100644 --- a/cmd/mavend/ecosystem_acts.go +++ b/cmd/mavend/ecosystem_acts.go @@ -3,9 +3,11 @@ package main import ( "context" "encoding/json" + "errors" "fmt" "log" "strings" + "time" hexisclient "github.com/kami/hexis/pkg/client" "github.com/kami/maven/internal/ipc" @@ -291,6 +293,14 @@ func (h *reactiveHandler) localFactsForEntity(ctx context.Context, entityID stri return "я помню: " + strings.Join(parts, ", ") } +// mergeFields overlays b onto a and returns a. +func mergeFields(a, b map[string]any) map[string]any { + for k, v := range b { + a[k] = v + } + 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. @@ -302,14 +312,94 @@ func (h *reactiveHandler) recordPraxisTrace(ctx context.Context, operation strin value = operation + " " + string(b) } } - _, _ = h.api.WriteFact(ctx, ipc.WriteFactReq{ + // 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: "system", + 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) + } +} + +// traceStatus classifies an ecosystem call for the trace record. Kept coarse +// on purpose: a trace is read to answer "did this hop work, and how long did +// it take", not to re-derive the error. +const ( + traceOK = "ok" + traceFailed = "failed" + traceRefused = "refused" // the far side answered, and said no + traceAmbig = "ambiguous" + traceNotFound = "not_found" +) + +// 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. +func redactSubject(s string) string { + return fmt.Sprintf("<%d chars>", len([]rune(s))) +} + +// 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. +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 id := correlationIDFromCtx(ctx); id != "" { + details["correlation_id"] = id + } + for k, v := range fields { + details[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 { + log.Printf("ecosystem: record trace %s:%s: %v", service, op, err) + } +} + +// 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 { + fields := map[string]any{} + var ee *ecosystemError + if errors.As(err, &ee) { + fields["http_status"] = ee.Status + switch { + case ee.Unauthorized(): + fields["class"] = "unauthorized" + case ee.ContractMismatch(): + fields["class"] = "contract_mismatch" + case ee.Unreachable(): + fields["class"] = "unreachable" + default: + fields["class"] = "error" + } + return fields + } + fields["class"] = "error" + return fields } // handleHexisAct — resolves entity references through Nexus and executes @@ -320,10 +410,20 @@ func (h *reactiveHandler) handleHexisAct(ctx context.Context, dec router.Decisio return "" } + // Every hop of this action shares one correlation ID, assigned here so + // resolution and discovery are traceable even when execution never + // happens. + if correlationIDFromCtx(ctx) == "" { + ctx = withCorrelationID(ctx, newCorrelationID()) + } + // Resolve the utterance text as an entity reference through Nexus. An // ambiguous match must stop and clarify — never guess a mutation target. + 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, + mergeFields(traceErrorFields(err), map[string]any{"subject": redactSubject(dec.Slots.Text)})) // 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 @@ -331,19 +431,30 @@ func (h *reactiveHandler) handleHexisAct(ctx context.Context, dec router.Decisio return "экосистема недоступна, попробуй ещё раз." } if len(ambiguous) > 0 { + h.recordEcosystemTrace(ctx, "nexus", "resolve", traceAmbig, started, + map[string]any{"candidates": len(ambiguous)}) return "уточни, что именно: " + strings.Join(ambiguous, ", ") + "?" } if entityID == "" { + h.recordEcosystemTrace(ctx, "nexus", "resolve", traceNotFound, started, + map[string]any{"subject": redactSubject(dec.Slots.Text)}) return "" } + h.recordEcosystemTrace(ctx, "nexus", "resolve", traceOK, started, + map[string]any{"entity_id": entityID}) // Discover Hexis capabilities for this entity. A resolved entity with a // genuine Hexis failure must not be treated as "no capabilities" and // fall through to unrelated local execution. + discovered := h.now() caps, err := h.ecosystem.discoverCapabilities(ctx, entityID) if err != nil { + h.recordEcosystemTrace(ctx, "hexis", "capabilities", traceFailed, discovered, + mergeFields(traceErrorFields(err), map[string]any{"entity_id": entityID})) return "экосистема недоступна, попробуй ещё раз." } + h.recordEcosystemTrace(ctx, "hexis", "capabilities", traceOK, discovered, + map[string]any{"entity_id": entityID, "count": len(caps)}) if len(caps) == 0 { return "" } @@ -388,6 +499,8 @@ 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(), + map[string]any{"entity_id": entityID, "capability": matched.Name}) return "выполнить «" + matched.Name + "» для " + displayName + "? скажи «да» или «нет»." } @@ -398,11 +511,21 @@ func (h *reactiveHandler) handleHexisAct(ctx context.Context, dec router.Decisio // the correlation ID. It reports command success, never operational recovery // (Praxis observes recovery independently). func (h *reactiveHandler) execHexis(ctx context.Context, capID, capName, entityID, displayName string) string { + started := h.now() + causationID := correlationIDFromCtx(ctx) correlationID, err := h.ecosystem.executeCapability(ctx, capID, entityID, nil) + 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, + mergeFields(traceErrorFields(err), map[string]any{ + "entity_id": entityID, "capability": capName, "causation_id": causationID, + })) return "не получилось выполнить команду для " + displayName + "." } + 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, diff --git a/cmd/mavend/ecosystem_trace_test.go b/cmd/mavend/ecosystem_trace_test.go new file mode 100644 index 0000000..ee59d19 --- /dev/null +++ b/cmd/mavend/ecosystem_trace_test.go @@ -0,0 +1,277 @@ +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) + } +} diff --git a/internal/config/config.go b/internal/config/config.go index a679351..0124e1a 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -461,18 +461,30 @@ func (c *Config) MCPServers() []mcp.ServerConfig { type PraxisConfig struct { // URL — the Praxis HTTP API base URL (e.g. "http://localhost:9742"). URL string `json:"url,omitempty"` + + // Token — the shared bearer token sent on every request. Empty ⇒ calls + // go out unauthenticated, which is only appropriate on a loopback or + // unix-socket transport. Supports ${VAR} expansion, so the secret lives + // in deploy/telegram.env, not in the committed config. + Token string `json:"token,omitempty"` } // NexusConfig — connection to the Nexus identity service. type NexusConfig struct { // URL — the Nexus HTTP API base URL (e.g. "http://localhost:9740"). URL string `json:"url,omitempty"` + + // Token — shared bearer token; see PraxisConfig.Token. + Token string `json:"token,omitempty"` } // HexisConfig — connection to the Hexis capability execution service. type HexisConfig struct { // URL — the Hexis HTTP API base URL (e.g. "http://localhost:9741"). URL string `json:"url,omitempty"` + + // Token — shared bearer token; see PraxisConfig.Token. + Token string `json:"token,omitempty"` } // RoutineConfig — one scheduled routine. Cron is a standard 5-field expression