From cfbef45feb704fa4ffc89bcf1a264a7c9cee04af Mon Sep 17 00:00:00 2001 From: claude Date: Thu, 6 Aug 2026 01:37:20 +0400 Subject: [PATCH 1/2] ecosystem clients share one JSON transport (V-575) Nexus and Praxis were the same HTTP client written twice: build the request, stamp the headers, send it, check the status, decode the body, and wrap each failure in an ecosystemError. They differ only in the service name and the version header, so both now embed ecosystemHTTP and call getJSON or postJSON. setEcosystemHeaders took a context beside the request it was stamping. It now reads req.Context(), which is the same value, so a caller cannot pass a context the request never carried. No behaviour change. Same headers, same statuses, same error types. One error changed shape: a malformed base URL used to come back from Nexus resolve as a plain wrapped error and is now an ecosystemError like every other failure on that path. Co-Authored-By: Claude Opus 5 --- cmd/mavend/ecosystem.go | 204 +++++++++++++++++++--------------------- 1 file changed, 97 insertions(+), 107 deletions(-) diff --git a/cmd/mavend/ecosystem.go b/cmd/mavend/ecosystem.go index 5d32f6b..2e1a9bf 100644 --- a/cmd/mavend/ecosystem.go +++ b/cmd/mavend/ecosystem.go @@ -42,28 +42,100 @@ const ecosystemAPIVersion = "v1" // 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). +// ecosystemHTTP is the JSON transport every ecosystem client shares: one base +// URL, one bearer token, and the header set the contract requires on each +// request. Nexus and Praxis differ only in the service name and the version +// header, so both embed this rather than repeating build, send and classify. +type ecosystemHTTP struct { + service string // "nexus", "praxis" — the name errors and traces carry + versionHeader string + baseURL string + token string + httpClient *http.Client +} + +func newEcosystemHTTP(service, versionHeader, baseURL string) ecosystemHTTP { + return ecosystemHTTP{ + service: service, + versionHeader: versionHeader, + baseURL: baseURL, + httpClient: &http.Client{Timeout: 10 * time.Second}, + } +} + +// setHeaders stamps the version, requester, auth and correlation headers common +// to every outgoing ecosystem request. The token may be empty, which means the +// transport itself is trusted (loopback or unix socket). // -// The correlation ID is read from the context and never minted here. Minting -// one per request sent the far side an ID that existed nowhere on this side, -// and gave a single multi-hop action as many unrelated IDs as it made calls. -// Callers that start an action assign the ID once (handleHexisAct, +// The correlation ID is read from the request's own context and never minted +// here. Minting one per request sent the far side an ID that existed nowhere on +// this side, and gave a single multi-hop action as many unrelated IDs as it +// made calls. Callers that start an action assign the ID once (handleHexisAct, // handlePraxisAct, resolveEntityReference) and every hop inherits it. -func setEcosystemHeaders(req *http.Request, ctx context.Context, versionHeader, token string) { +func (t *ecosystemHTTP) setHeaders(req *http.Request) { req.Header.Set("Content-Type", "application/json") - req.Header.Set(versionHeader, ecosystemAPIVersion) + req.Header.Set(t.versionHeader, ecosystemAPIVersion) req.Header.Set("Accept", "application/json") req.Header.Set("X-Requested-By", mavenRequester) - if token != "" { - req.Header.Set("Authorization", "Bearer "+token) + if t.token != "" { + req.Header.Set("Authorization", "Bearer "+t.token) } - if id := correlationIDFromCtx(ctx); id != "" { + if id := correlationIDFromCtx(req.Context()); id != "" { req.Header.Set("X-Correlation-ID", id) } } +// call sends one request and decodes the JSON answer into out, which may be nil +// when the body carries nothing worth reading. op is the logical operation name +// for errors and traces: the path carries the query string, and after entity +// scoping that means an entity id in every log line built from the error, next +// to a trace that redacts far less than that. +// +// Every failure is an *ecosystemError, including the transport and decode ones. +// Some of these paths mutate remote state, and the question worth answering +// afterwards is whether the call never left or was refused. +func (t *ecosystemHTTP) call(ctx context.Context, method, op, path string, payload, out any) error { + var body io.Reader + if payload != nil { + data, err := json.Marshal(payload) + if err != nil { + return &ecosystemError{Service: t.service, Op: op, Err: err} + } + body = bytes.NewReader(data) + } + req, err := http.NewRequestWithContext(ctx, method, t.baseURL+path, body) + if err != nil { + return &ecosystemError{Service: t.service, Op: op, Err: err} + } + t.setHeaders(req) + + resp, err := t.httpClient.Do(req) + if err != nil { + return &ecosystemError{Service: t.service, Op: op, Err: err} + } + defer resp.Body.Close() + if resp.StatusCode != 200 { + return httpError(t.service, op, resp.StatusCode) + } + if out == nil { + return nil + } + if err := json.NewDecoder(resp.Body).Decode(out); err != nil { + return &ecosystemError{Service: t.service, Op: op, Status: resp.StatusCode, Err: err} + } + return nil +} + +// getJSON performs a GET and decodes the JSON body into out. +func (t *ecosystemHTTP) getJSON(ctx context.Context, op, path string, out any) error { + return t.call(ctx, http.MethodGet, op, path, nil, out) +} + +// postJSON posts a JSON payload and decodes the JSON answer into out. +func (t *ecosystemHTTP) postJSON(ctx context.Context, op, path string, payload, out any) error { + return t.call(ctx, http.MethodPost, op, path, payload, out) +} + // 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: @@ -108,16 +180,11 @@ func httpError(service, op string, status int) *ecosystemError { } type nexusClient struct { - baseURL string - token string - httpClient *http.Client + ecosystemHTTP } func newNexusClient(url string) *nexusClient { - return &nexusClient{ - baseURL: url, - httpClient: &http.Client{Timeout: 10 * time.Second}, - } + return &nexusClient{newEcosystemHTTP("nexus", "X-Nexus-Version", url)} } // withToken sets the bearer token sent on every request. Returns the client so @@ -176,62 +243,26 @@ func (c *nexusClient) Resolve(ctx context.Context, query string, types []string) body["types"] = types } - data, _ := json.Marshal(body) - req, err := http.NewRequestWithContext(ctx, http.MethodPost, c.baseURL+"/api/v1/resolve", bytes.NewReader(data)) - if err != nil { - return nil, fmt.Errorf("create request: %w", err) - } - setEcosystemHeaders(req, ctx, "X-Nexus-Version", c.token) - - resp, err := c.httpClient.Do(req) - if err != nil { - return nil, &ecosystemError{Service: "nexus", Op: "resolve", Err: err} - } - defer resp.Body.Close() - - bodyBytes, _ := io.ReadAll(resp.Body) - if resp.StatusCode != 200 { - return nil, httpError("nexus", "resolve", resp.StatusCode) - } - var result nexusResolveResult - if err := json.Unmarshal(bodyBytes, &result); err != nil { - return nil, &ecosystemError{Service: "nexus", Op: "resolve", Status: resp.StatusCode, Err: err} + if err := c.postJSON(ctx, "resolve", "/api/v1/resolve", body, &result); err != nil { + return nil, err } return &result, nil } func (c *nexusClient) Health(ctx context.Context) error { - req, err := http.NewRequestWithContext(ctx, http.MethodGet, c.baseURL+"/health", nil) - if err != nil { - return &ecosystemError{Service: "nexus", Op: "health", Err: err} - } - setEcosystemHeaders(req, ctx, "X-Nexus-Version", c.token) - resp, err := c.httpClient.Do(req) - if err != nil { - return &ecosystemError{Service: "nexus", Op: "health", Err: err} - } - resp.Body.Close() - if resp.StatusCode != 200 { - return httpError("nexus", "health", resp.StatusCode) - } - return nil + return c.getJSON(ctx, "health", "/health", nil) } // praxisClient talks to the Praxis HTTP tools API. Maven must not open Praxis's // SQLite store directly (ecosystem invariant: no component reads another's DB), // so attention/changes/lifecycle all go over this HTTP contract against praxisd. type praxisClient struct { - baseURL string - token string - httpClient *http.Client + ecosystemHTTP } func newPraxisClient(url string) *praxisClient { - return &praxisClient{ - baseURL: url, - httpClient: &http.Client{Timeout: 10 * time.Second}, - } + return &praxisClient{newEcosystemHTTP("praxis", "X-Praxis-Version", url)} } func (c *praxisClient) withToken(token string) *praxisClient { @@ -239,30 +270,6 @@ func (c *praxisClient) withToken(token string) *praxisClient { return c } -// getJSON performs a GET and decodes the JSON body into out. op is the logical -// operation name for errors and traces: the path carries the query string, and -// after entity scoping that means an entity id in every log line built from the -// error, next to a trace that redacts far less than that. -func (c *praxisClient) getJSON(ctx context.Context, op, 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", c.token) - resp, err := c.httpClient.Do(req) - if err != nil { - return &ecosystemError{Service: "praxis", Op: op, Err: err} - } - defer resp.Body.Close() - if resp.StatusCode != 200 { - return httpError("praxis", op, resp.StatusCode) - } - if err := json.NewDecoder(resp.Body).Decode(out); err != nil { - return &ecosystemError{Service: "praxis", Op: op, Status: resp.StatusCode, Err: err} - } - return nil -} - // praxisAttention — an attention response in either of the two shapes Praxis // may send (Vikunja #540). // @@ -376,31 +383,14 @@ type praxisItem struct { // postItemAction posts {"item_id": id} to a Praxis tools lifecycle endpoint // and decodes the resulting item. Shared by Surface/Acknowledge/Resolve/Ignore. func (c *praxisClient) postItemAction(ctx context.Context, op, path, itemID string) (*praxisItem, error) { - return c.postJSON(ctx, op, path, map[string]any{"item_id": itemID}) + return c.postItem(ctx, op, path, map[string]any{"item_id": itemID}) } -// postJSON posts a body to a Praxis lifecycle endpoint and decodes the item. -// Every failure is a *ecosystemError, including the transport and decode ones: -// these are the paths that mutate remote state, and the question worth -// answering afterwards is whether the call never left or was refused. -func (c *praxisClient) postJSON(ctx context.Context, op, path string, payload map[string]any) (*praxisItem, error) { - body, _ := json.Marshal(payload) - req, err := http.NewRequestWithContext(ctx, http.MethodPost, c.baseURL+path, bytes.NewReader(body)) - if err != nil { - return nil, &ecosystemError{Service: "praxis", Op: op, Err: err} - } - setEcosystemHeaders(req, ctx, "X-Praxis-Version", c.token) - resp, err := c.httpClient.Do(req) - if err != nil { - return nil, &ecosystemError{Service: "praxis", Op: op, Err: err} - } - defer resp.Body.Close() - if resp.StatusCode != 200 { - return nil, httpError("praxis", op, resp.StatusCode) - } +// postItem posts a body to a Praxis lifecycle endpoint and decodes the item. +func (c *praxisClient) postItem(ctx context.Context, op, path string, payload map[string]any) (*praxisItem, error) { var out praxisItem - if err := json.NewDecoder(resp.Body).Decode(&out); err != nil { - return nil, &ecosystemError{Service: "praxis", Op: op, Status: resp.StatusCode, Err: err} + if err := c.postJSON(ctx, op, path, payload, &out); err != nil { + return nil, err } return &out, nil } @@ -425,7 +415,7 @@ func (c *praxisClient) Ignore(ctx context.Context, itemID string) (*praxisItem, } func (c *praxisClient) Pin(ctx context.Context, itemID string, pinned bool) (*praxisItem, error) { - return c.postJSON(ctx, "pin", "/api/v1/tools/pin", map[string]any{"item_id": itemID, "pinned": pinned}) + return c.postItem(ctx, "pin", "/api/v1/tools/pin", map[string]any{"item_id": itemID, "pinned": pinned}) } func (c *praxisClient) GetItem(ctx context.Context, itemID string) (*praxisItem, error) { From 95eeef13ddb2a87eeb83466110e4b3ae5603f52d Mon Sep 17 00:00:00 2001 From: claude Date: Thu, 6 Aug 2026 01:37:30 +0400 Subject: [PATCH 2/2] every ecosystem degrade path names the gap the same way (V-575) Three call sites wrote the same two lines: a rejected credential says eco_denied, anything else says eco_down, and both name the service. That is ecosystemGap now, so a fourth caller cannot get it half right. The failed Nexus resolve was also traced twice in the same shape, once in the entity attention arm and once in the Hexis act. Both now go through nexusResolveFailed, which keeps the rune count redaction. Surfacing a spoken item was written twice as well. surfaceSpoken calls Surface and nothing else, because reading an item aloud is not an acknowledgement. No behaviour change: same lifecycle verbs, same replies, same trace fields. traceErrorFields lost a duplicated default and returns what it returned before. Co-Authored-By: Claude Opus 5 --- cmd/mavend/ecosystem_acts.go | 98 +++++++++++++++++++----------------- 1 file changed, 53 insertions(+), 45 deletions(-) diff --git a/cmd/mavend/ecosystem_acts.go b/cmd/mavend/ecosystem_acts.go index 1f5438a..4c7a852 100644 --- a/cmd/mavend/ecosystem_acts.go +++ b/cmd/mavend/ecosystem_acts.go @@ -29,6 +29,17 @@ const ( // serviceVars — the one-key map the eco_down and eco_denied lines take. func serviceVars(name string) map[string]string { return map[string]string{"name": name} } +// ecosystemGap names the service that failed. A rejected credential gets its +// own line, because a wrong token looks exactly like an outage to him and +// "try again" is advice that will never work. Every degrade path reads through +// here, so all of them name the service and none of them guesses instead. +func ecosystemGap(service string, err error) string { + if unauthorizedEcosystemError(err) { + return phraser.A(phraser.EcoDenied, serviceVars(service)) + } + return phraser.A(phraser.EcoDown, serviceVars(service)) +} + // praxisCapability is one arm of the Praxis act dispatch. This is an interface // rather than a map[string]func because each arm carries its own state: the // verb aliases it answers to, the trace name it records, and its own reply @@ -199,16 +210,10 @@ func (listAttentionCapability) handle(ctx context.Context, h *reactiveHandler, p } parts = append(parts, s) - // Speaking an item surfaces it, it does not acknowledge it - // (ECOSYSTEM-SPEC.md §2.3: surfaced != acknowledged). Best-effort: - // a failed surface call must not block delivering the digest. - if id, ok := item["id"].(string); ok && id != "" { - // Recorded in the order she says them, and only for items she could - // say: an item skipped above has no position in what he heard (#516). + // Recorded in the order she says them, and only for items she could + // say: an item skipped above has no position in what he heard (#516). + if id := surfaceSpoken(ctx, px, item); id != "" { spoken = append(spoken, id) - if _, err := px.Surface(ctx, id); err != nil { - log.Printf("ecosystem: praxis surface %s: %v", id, err) - } } } h.rememberSurfaced(spoken) @@ -298,12 +303,7 @@ func (entityAttentionCapability) handle(ctx context.Context, h *reactiveHandler, // trace gets. A trace that stores a rune count next to a log line // storing the runes is not redacted at all. log.Printf("ecosystem: entity attention resolve %s: %v", redactSubject(subject), err) - h.recordEcosystemTrace(ctx, "nexus", "resolve", traceStatusForError(err), started, - mergeFields(traceErrorFields(err), map[string]any{"subject": redactSubject(subject)})) - if unauthorizedEcosystemError(err) { - return phraser.A(phraser.EcoDenied, serviceVars(serviceNexus)) - } - return phraser.A(phraser.EcoDown, serviceVars(serviceNexus)) + return h.nexusResolveFailed(ctx, subject, started, err) } if len(ambiguous) > 0 { return phraser.A(phraser.EcoAmbiguous, map[string]string{"items": strings.Join(ambiguous, ", ")}) @@ -345,12 +345,7 @@ func (entityAttentionCapability) handle(ctx context.Context, h *reactiveHandler, continue } parts = append(parts, title) - // Same surfaced != acknowledged rule as the unscoped digest. - if id, ok := item["id"].(string); ok && id != "" { - if _, err := px.Surface(ctx, id); err != nil { - log.Printf("ecosystem: praxis surface %s: %v", id, err) - } - } + surfaceSpoken(ctx, px, item) } if known := h.localFactsForEntity(ctx, entityID); known != "" { parts = append(parts, known) @@ -366,6 +361,22 @@ func (entityAttentionCapability) handle(ctx context.Context, h *reactiveHandler, return phraser.A(phraser.AttentionListEntity, map[string]string{"name": displayName, "items": strings.Join(parts, "; ")}) } +// surfaceSpoken marks an item she just read out as surfaced. Speaking an item +// surfaces it, it does not acknowledge it (ECOSYSTEM-SPEC.md §2.3: surfaced != +// acknowledged), so this calls Surface and nothing else. Best effort: a failed +// surface call must not block delivering the digest. Returns the item id, or "" +// when the item carried none. +func surfaceSpoken(ctx context.Context, px *praxisClient, item map[string]any) string { + id, _ := item["id"].(string) + if id == "" { + return "" + } + if _, err := px.Surface(ctx, id); err != nil { + log.Printf("ecosystem: praxis surface %s: %v", id, err) + } + return id +} + // scopedToEntity drops items that carry an entity_id other than the one asked // about, and reports whether the response can be trusted as scoped at all. An // item without an entity_id is kept only when at least one sibling carries the @@ -474,6 +485,14 @@ func traceStatusForError(err error) string { return traceFailed } +// nexusResolveFailed records a resolve that failed and returns the named gap. +// The subject is his words, so the trace keeps a rune count and not the runes. +func (h *reactiveHandler) nexusResolveFailed(ctx context.Context, subject string, started time.Time, err error) string { + h.recordEcosystemTrace(ctx, "nexus", "resolve", traceStatusForError(err), started, + mergeFields(traceErrorFields(err), map[string]any{"subject": redactSubject(subject)})) + return ecosystemGap(serviceNexus, err) +} + // 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. @@ -535,23 +554,20 @@ func unauthorizedEcosystemError(err error) bool { // 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{} + fields := map[string]any{"class": "error"} 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" - } + if !errors.As(err, &ee) { return fields } - fields["class"] = "error" + 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" + } return fields } @@ -636,16 +652,11 @@ func (h *reactiveHandler) handleHexisAct(ctx context.Context, dec router.Decisio res := h.resolveEntityCandidates(ctx, entityReferences(dec)) subject, entityID, displayName, ambiguous, err := res.subject, res.entityID, res.displayName, res.ambiguous, res.err if err != nil { - h.recordEcosystemTrace(ctx, "nexus", "resolve", traceStatusForError(err), started, - mergeFields(traceErrorFields(err), map[string]any{"subject": redactSubject(subject)})) - if unauthorizedEcosystemError(err) { - return phraser.A(phraser.EcoDenied, serviceVars(serviceNexus)) - } // 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 // independently, never a silent all-clear). - return phraser.A(phraser.EcoDown, serviceVars(serviceNexus)) + return h.nexusResolveFailed(ctx, subject, started, err) } if len(ambiguous) > 0 { h.recordEcosystemTrace(ctx, "nexus", "resolve", traceAmbig, started, @@ -668,10 +679,7 @@ func (h *reactiveHandler) handleHexisAct(ctx context.Context, dec router.Decisio if err != nil { h.recordEcosystemTrace(ctx, "hexis", "capabilities", traceStatusForError(err), discovered, mergeFields(traceErrorFields(err), map[string]any{"entity_id": entityID})) - if unauthorizedEcosystemError(err) { - return phraser.A(phraser.EcoDenied, serviceVars(serviceHexis)) - } - return phraser.A(phraser.EcoDown, serviceVars(serviceHexis)) + return ecosystemGap(serviceHexis, err) } h.recordEcosystemTrace(ctx, "hexis", "capabilities", traceOK, discovered, map[string]any{"entity_id": entityID, "count": len(caps)})