kami/Maven
1
0
Fork 0
Code Issues 1 Pull Requests 1 Actions Packages Projects Releases Wiki Activity

Version, authenticate and fully trace ecosystem calls (#273)

Browse Source 
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.
This commit is contained in:
kami
2026-08-01 06:57:52 +04:00
parent 08f3db318f
commit 927e46bca3
4 changed files with 523 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cmd/mavend/ecosystem.go
+108 -21
View File
@@ -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")