33c9c39676
Add an optional remote atomization backend (the Anypreta "Issue Atomizer"
API) for the Subdivide flow, selected by ATOMIZER_REMOTE_BASE_URL /
ATOMIZER_REMOTE_API_KEY. When set, POST /v1/atomize is served by the remote
/v1/atomize-issue endpoint; its two-level issue→features→tasks model is mapped
onto our one-level task→children model by treating each Feature as a child and
its `estimation` as the effort coefficient (renormalized to sum to 1, even
split as a reported fallback when estimations are absent).
Extend and Summarize have no counterpart in the remote API and always stay on
ATOMIZER_BASE_URL, so the §5.1 service must keep running; readiness now probes
both backends. The mandatory `system` object is derived from the customer,
with a generic component tree (an empty tree makes the service return zero
features).
extsvc also learns the remote's flat error envelope ({error_code,message} and
FastAPI {detail}) and never swallows an unrecognized 4xx body.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
172 lines
4.6 KiB
Go
172 lines
4.6 KiB
Go
package extsvc
|
||
|
||
import (
|
||
"bytes"
|
||
"context"
|
||
"encoding/json"
|
||
"errors"
|
||
"fmt"
|
||
"io"
|
||
"math/rand/v2"
|
||
"net/http"
|
||
"time"
|
||
)
|
||
|
||
// ErrBreakerOpen is returned without touching the network when the circuit
|
||
// is open.
|
||
var ErrBreakerOpen = errors.New("circuit breaker open")
|
||
|
||
// ServiceError is a non-2xx response from the external service, carrying
|
||
// its §5 error envelope when present.
|
||
type ServiceError struct {
|
||
Status int
|
||
Code string
|
||
Message string
|
||
}
|
||
|
||
func (e *ServiceError) Error() string {
|
||
return fmt.Sprintf("service responded %d: %s %s", e.Status, e.Code, e.Message)
|
||
}
|
||
|
||
// Client is a bearer-token JSON client with retry (×2 on 5xx/network with
|
||
// jitter, §12) and a circuit breaker, shared by the two service clients.
|
||
type Client struct {
|
||
Name string
|
||
BaseURL func() string // resolved per call (admin override support)
|
||
Token string
|
||
HTTP *http.Client
|
||
Breaker *Breaker
|
||
}
|
||
|
||
func NewClient(name string, baseURL func() string, token string, timeout time.Duration) *Client {
|
||
return &Client{
|
||
Name: name,
|
||
BaseURL: baseURL,
|
||
Token: token,
|
||
HTTP: &http.Client{Timeout: timeout},
|
||
Breaker: NewBreaker(),
|
||
}
|
||
}
|
||
|
||
func (c *Client) BreakerState() string { return c.Breaker.State() }
|
||
|
||
// Healthy probes /healthz without breaker accounting (used by readyz and the
|
||
// admin panel).
|
||
func (c *Client) Healthy(ctx context.Context) error {
|
||
req, err := http.NewRequestWithContext(ctx, http.MethodGet, c.BaseURL()+"/healthz", nil)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
resp, err := c.HTTP.Do(req)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
defer resp.Body.Close()
|
||
if resp.StatusCode != http.StatusOK {
|
||
return fmt.Errorf("%s healthz: status %d", c.Name, resp.StatusCode)
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// Call POSTs/GETs JSON through the breaker with retries. A nil out skips
|
||
// decoding.
|
||
func (c *Client) Call(ctx context.Context, method, path string, in, out any) error {
|
||
if !c.Breaker.Allow() {
|
||
return fmt.Errorf("%s: %w", c.Name, ErrBreakerOpen)
|
||
}
|
||
err := c.callOnce(ctx, method, path, in, out)
|
||
c.Breaker.Record(err)
|
||
return err
|
||
}
|
||
|
||
func (c *Client) callOnce(ctx context.Context, method, path string, in, out any) error {
|
||
var body []byte
|
||
if in != nil {
|
||
var err error
|
||
if body, err = json.Marshal(in); err != nil {
|
||
return fmt.Errorf("encode request: %w", err)
|
||
}
|
||
}
|
||
|
||
var lastErr error
|
||
for attempt := 0; attempt < 3; attempt++ {
|
||
if attempt > 0 {
|
||
select {
|
||
case <-ctx.Done():
|
||
return ctx.Err()
|
||
case <-time.After(time.Duration(attempt)*time.Second + time.Duration(rand.IntN(500))*time.Millisecond):
|
||
}
|
||
}
|
||
req, err := http.NewRequestWithContext(ctx, method, c.BaseURL()+path, bytes.NewReader(body))
|
||
if err != nil {
|
||
return err
|
||
}
|
||
req.Header.Set("Content-Type", "application/json")
|
||
req.Header.Set("Authorization", "Bearer "+c.Token)
|
||
resp, err := c.HTTP.Do(req)
|
||
if err != nil {
|
||
lastErr = fmt.Errorf("%s %s: %w", c.Name, path, err)
|
||
continue // network error → retry
|
||
}
|
||
data, err := io.ReadAll(io.LimitReader(resp.Body, 10<<20))
|
||
resp.Body.Close()
|
||
if err != nil {
|
||
lastErr = fmt.Errorf("%s %s: read body: %w", c.Name, path, err)
|
||
continue
|
||
}
|
||
if resp.StatusCode >= 500 {
|
||
lastErr = &ServiceError{Status: resp.StatusCode, Message: string(truncate(data, 200))}
|
||
continue // 5xx → retry
|
||
}
|
||
if resp.StatusCode >= 400 {
|
||
se := &ServiceError{Status: resp.StatusCode}
|
||
var env struct {
|
||
// §5 envelope: {"error":{"code":…,"message":…}}.
|
||
Error struct {
|
||
Code string `json:"code"`
|
||
Message string `json:"message"`
|
||
} `json:"error"`
|
||
// Flat envelope, used by the Anypreta atomizer:
|
||
// {"error_code":…,"message":…}. FastAPI's own validation
|
||
// errors carry {"detail":…} instead.
|
||
ErrorCode string `json:"error_code"`
|
||
Message string `json:"message"`
|
||
Detail json.RawMessage `json:"detail"`
|
||
}
|
||
if json.Unmarshal(data, &env) == nil {
|
||
se.Code, se.Message = env.Error.Code, env.Error.Message
|
||
if se.Code == "" {
|
||
se.Code = env.ErrorCode
|
||
}
|
||
if se.Message == "" {
|
||
se.Message = env.Message
|
||
}
|
||
if se.Message == "" && len(env.Detail) > 0 {
|
||
se.Message = string(truncate(env.Detail, 200))
|
||
}
|
||
}
|
||
if se.Message == "" {
|
||
// Never swallow the failure: an unrecognized envelope still
|
||
// has to reach the logs and the consultant's error toast.
|
||
se.Message = string(truncate(data, 200))
|
||
}
|
||
return se // 4xx is not retryable
|
||
}
|
||
if out == nil {
|
||
return nil
|
||
}
|
||
if err := json.Unmarshal(data, out); err != nil {
|
||
return fmt.Errorf("%s %s: decode response: %w", c.Name, path, err)
|
||
}
|
||
return nil
|
||
}
|
||
return lastErr
|
||
}
|
||
|
||
func truncate(b []byte, n int) []byte {
|
||
if len(b) <= n {
|
||
return b
|
||
}
|
||
return b[:n]
|
||
}
|