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) { 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)})