diff --git a/.env.example b/.env.example index 6e15b99..f129a37 100644 --- a/.env.example +++ b/.env.example @@ -28,7 +28,7 @@ ARTEX_LLM_PROXY= # probe). Default off: no limit, requests unchanged. quiet = 2 req/s per host, burst 4, # 1 in flight; balanced = 5 req/s, burst 10, 2 in flight; aggressive = 20 req/s, burst 40, # 4 in flight. A profile also adds jitter, backs off on 429/503 and merges identical GETs -# of one task (short cache, not for WebFetch). Bash-launched tools are not throttled. +# of one task per tool (short cache; WebFetch is paced only). Bash-launched tools are not throttled. # ARTEX_NOISE_PROFILE=balanced # User-Agent set on target requests that have none, only under a profile. Empty = unchanged. # ARTEX_TARGET_USER_AGENT= diff --git a/README.md b/README.md index 38d1133..3cb0c55 100644 --- a/README.md +++ b/README.md @@ -264,7 +264,7 @@ Optional: `ARTEX_LLM_PROVIDER` (anthropic | openai | openai-responses | ollama) **Untrusted-output boundary (on by default)**: tool results for the worker, planner, chat and main agent are framed between nonce-tagged markers so the model treats them as data, not instructions. Unset or empty, the boundary is on; set `ARTEX_UNTRUSTED_BOUNDARY=0` (or `false` / `off` / `no`) to disable it. Measured cost: about 154 bytes per tool result plus one system note. -**Noise profile (off by default)**: `ARTEX_NOISE_PROFILE` paces the requests ARTEX itself sends to targets (WebFetch, http custom tools, the enrich probe). Unset or `off`, nothing changes: no limit, no cache. Set `quiet` (2 req/s per host, burst 4, 1 in flight), `balanced` (5 req/s, burst 10, 2 in flight) or `aggressive` (20 req/s, burst 40, 4 in flight), e.g. `ARTEX_NOISE_PROFILE=balanced` in `.env` or the environment. A profile also adds a small jitter, backs off on 429/503 (honouring `Retry-After`, bounded retries, GET/HEAD/OPTIONS only), merges identical in-flight GETs of one task and, for the http custom tools and the enrich probe, caches them briefly (60/30/10 s; WebFetch results are never cached; never POST, PUT, PATCH or DELETE), and logs a per-host noise line each minute. `ARTEX_TARGET_USER_AGENT` sets a User-Agent on target requests that carry none, under a profile; empty keeps the current one. Tools launched through Bash are not paced yet. +**Noise profile (off by default)**: `ARTEX_NOISE_PROFILE` paces the requests ARTEX itself sends to targets (WebFetch, http custom tools, the enrich probe). Unset or `off`, nothing changes: no limit, no cache. Set `quiet` (2 req/s per host, burst 4, 1 in flight), `balanced` (5 req/s, burst 10, 2 in flight) or `aggressive` (20 req/s, burst 40, 4 in flight), e.g. `ARTEX_NOISE_PROFILE=balanced` in `.env` or the environment. A profile also adds a small jitter, backs off on 429/503 (honouring `Retry-After`, bounded retries, GET/HEAD/OPTIONS only), and, for the http custom tools and the enrich probe, merges identical in-flight GETs of one task (per tool) and caches them briefly (60/30/10 s; never POST, PUT, PATCH or DELETE; WebFetch is paced only), and logs a per-host noise line each minute. `ARTEX_TARGET_USER_AGENT` sets a User-Agent on target requests that carry none, under a profile; empty keeps the current one. Tools launched through Bash are not paced yet. **Common flags**: `./start.sh -addr :8787 -proxy :8788` (`-addr` frontend+API, `-proxy` traffic recording proxy). The startup script passes arguments through to `artex` verbatim. diff --git a/agent/webfetch.go b/agent/webfetch.go index fe65b46..3d3923e 100644 --- a/agent/webfetch.go +++ b/agent/webfetch.go @@ -4,7 +4,6 @@ import ( "context" "encoding/json" "net/url" - "slices" "github.com/Autumn-27/artex/egress" "github.com/Autumn-27/artex/intercept" @@ -35,12 +34,10 @@ func withWebFetch(ctx context.Context, tools []actool.CoreTool, proxy, caCert st var egressGovernor = egress.Shared // governedFetch runs each WebFetch call as one send under the egress governor -// (off by default; then the call runs unchanged), under a noise profile: -// the rate, jitter and concurrency of the URL's host, and, within one task, -// identical concurrent calls (WebFetch only sends GET) coalesced. Results are -// never cached: the key does not hold the tool's proxy or gate config, so each -// later call runs the tool and its redirect and dial gate again. WebFetch -// ignores tc, so a coalesced result fits every caller. +// (off by default; then the call runs unchanged). Under a noise profile it +// paces the call by the rate, jitter and concurrency of the URL's host, and +// nothing else: WebFetch calls are neither coalesced nor cached, since each +// tool carries its own proxy and redirect and dial gate (see egress.Do). // Norma owns the HTTP client, so redirect hops are not paced separately and a // 429/503 is not retried. type governedFetch struct { @@ -59,9 +56,7 @@ func (f governedFetch) Call(ctx context.Context, input json.RawMessage, tc *acto if err != nil || u.Host == "" { return f.CoreTool.Call(ctx, input, tc) // the tool reports the bad URL } - r, err := egress.Do(f.g, ctx, u, string(input), func() (actool.Result, error) { + return egress.Do(f.g, ctx, u, func() (actool.Result, error) { return f.CoreTool.Call(ctx, input, tc) }) - r.Content = slices.Clone(r.Content) // each caller gets its own slice - return r, err } diff --git a/egress/egress.go b/egress/egress.go index b91262b..4812db6 100644 --- a/egress/egress.go +++ b/egress/egress.go @@ -245,7 +245,8 @@ var errNotShared = errors.New("egress: response not shared") // share runs fn once for the concurrent callers of key and keeps its value for // the ttl fn returns. A caller that waited on a flight that failed for a reason // of the leader's own (its context, a response too large to share) runs fn -// itself, uncached. +// itself, uncached; fn paces its own send (transport.send goes through +// acquire), so that request is paced like any other. func (g *Governor) share(ctx context.Context, host, key string, fn func() (any, time.Duration, error)) (any, error) { g.mu.Lock() f := g.flights[key] @@ -304,51 +305,43 @@ func taskKey(ctx context.Context) string { } // Do runs fn as one send to u for a caller whose HTTP client is not ours -// (WebFetch): the rate, jitter and concurrency of u's host and, when id is not -// empty and ctx has a task, coalescing of concurrent calls with the same id -// within the task. Nothing is cached: each later call runs fn, and with it the -// caller's own checks. fn hides the response, so a 429/503 is not retried -// here. Off, fn just runs. -func Do[T any](g *Governor, ctx context.Context, u *url.URL, id string, fn func() (T, error)) (T, error) { +// (WebFetch): only the rate, jitter and concurrency of u's host. Nothing is +// coalesced or cached: each WebFetch call carries its own proxy and its own +// redirect and dial gate, which a shared key would have to fingerprint +// completely; running every call keeps every gate decision, at the cost of a +// duplicate request now and then. fn hides the response, so a 429/503 is not +// retried here. Off, fn just runs. +func Do[T any](g *Governor, ctx context.Context, u *url.URL, fn func() (T, error)) (T, error) { if g.off() { return fn() } - host := HostKey(u) - run := func() (any, time.Duration, error) { - release, err := g.acquire(ctx, host) - if err != nil { - return nil, 0, err - } - defer release() - v, err := fn() - return v, 0, err + release, err := g.acquire(ctx, HostKey(u)) + if err != nil { + var zero T + return zero, err } - var v any - var err error - if task := taskKey(ctx); task != "" && id != "" { - v, err = g.share(ctx, host, task+"\x00"+id, run) - } else { - v, _, err = run() - } - t, _ := v.(T) - return t, err + defer release() + return fn() } // Transport returns base (nil: http.DefaultTransport) governed by g; off, base -// itself. -func (g *Governor) Transport(base http.RoundTripper) http.RoundTripper { +// itself. namespace names the client (a custom tool's key, "enrich"): it is +// part of the cache key, so clients with other proxies, gates or pins never +// share a response. +func (g *Governor) Transport(namespace string, base http.RoundTripper) http.RoundTripper { if base == nil { base = http.DefaultTransport } if g.off() { return base } - return &transport{g, base} + return &transport{g, namespace, base} } type transport struct { - g *Governor - base http.RoundTripper + g *Governor + namespace string + base http.RoundTripper } const maxSharedBody = 2 << 20 // a larger response is neither coalesced nor cached @@ -366,7 +359,8 @@ func (t *transport) RoundTrip(req *http.Request) (*http.Response, error) { var hdr bytes.Buffer _ = req.Header.Write(&hdr) var own *http.Response // the leader's response when too large to share - v, err := t.g.share(req.Context(), HostKey(req.URL), task+"\x00GET "+req.URL.String()+"\x00"+hdr.String(), func() (any, time.Duration, error) { + key := t.namespace + "\x00" + task + "\x00GET " + req.URL.String() + "\x00" + hdr.String() + v, err := t.g.share(req.Context(), HostKey(req.URL), key, func() (any, time.Duration, error) { resp, err := t.send(req) if err != nil { return nil, 0, err @@ -387,8 +381,11 @@ func (t *transport) RoundTrip(req *http.Request) (*http.Response, error) { } return &stored{resp, body}, ttl, nil }) - if own != nil || err != nil { - return own, err + if own != nil { + return own, nil // too large to share: this caller's own response + } + if err != nil { + return nil, err } s := v.(*stored) r := *s.resp diff --git a/egress/egress_test.go b/egress/egress_test.go index 660a400..11a16ec 100644 --- a/egress/egress_test.go +++ b/egress/egress_test.go @@ -48,7 +48,7 @@ func newServer(t *testing.T, delay time.Duration, h func(n int64, w http.Respons func do(t *testing.T, g *Governor, ctx context.Context, method, u string) (int, string) { req, _ := http.NewRequestWithContext(ctx, method, u, nil) - resp, err := (&http.Client{Transport: g.Transport(nil)}).Do(req) + resp, err := (&http.Client{Transport: g.Transport("test", nil)}).Do(req) if err != nil { t.Errorf("%s %s: %v", method, u, err) return 0, "" @@ -69,7 +69,7 @@ func parallel(n int, f func(i int)) { func TestOffIsStrictPassthrough(t *testing.T) { g := New(Profile{Name: "off"}) - if g.Transport(http.DefaultTransport) != http.DefaultTransport { + if g.Transport("test", http.DefaultTransport) != http.DefaultTransport { t.Fatal("off: Transport must return the base transport itself") } s := newServer(t, 50*time.Millisecond, nil) @@ -81,7 +81,7 @@ func TestOffIsStrictPassthrough(t *testing.T) { var calls atomic.Int64 u, _ := url.Parse(s.URL) for range 2 { - _, _ = Do(g, task("t1"), u, "x", func() (int, error) { calls.Add(1); return 0, nil }) + _, _ = Do(g, task("t1"), u, func() (int, error) { calls.Add(1); return 0, nil }) } if calls.Load() != 2 || len(g.report()) != 0 { t.Fatalf("off: Do calls = %d, report %v; want 2 calls, no counter", calls.Load(), g.report()) @@ -176,18 +176,49 @@ func TestDedupeAndCache(t *testing.T) { if r := g.report(); len(r) != 1 || !strings.Contains(r[0], "rpm=12 ") || !strings.Contains(r[0], "duplicates_avoided=5") { t.Fatalf("noise counter: %v", r) } - // WebFetch path: Do coalesces concurrent calls and caches nothing. + // WebFetch path: Do paces, and neither coalesces nor caches. var calls atomic.Int64 u, _ := url.Parse(s.URL) fn := func() (string, error) { calls.Add(1); time.Sleep(50 * time.Millisecond); return "r", nil } parallel(3, func(int) { - if v, _ := Do(g, task("t1"), u, "a", fn); v != "r" { + if v, _ := Do(g, task("t1"), u, fn); v != "r" { t.Errorf("value = %q", v) } }) - _, _ = Do(g, task("t1"), u, "a", fn) - if calls.Load() != 2 { - t.Fatalf("Do calls = %d, want 2 (one coalesced flight, then no cache)", calls.Load()) + _, _ = Do(g, task("t1"), u, fn) + if calls.Load() != 4 { + t.Fatalf("Do calls = %d, want 4", calls.Load()) + } + // Two clients (namespaces) with the same task, URL and headers never share + // a response; the same namespace again is a cache hit. + for _, ns := range []string{"tool-a", "tool-b", "tool-a"} { + req, _ := http.NewRequestWithContext(task("t9"), "GET", s.URL+"/ns", nil) + if resp, err := (&http.Client{Transport: g.Transport(ns, nil)}).Do(req); err == nil { + resp.Body.Close() + } + } + if s.hits.Load() != 14 { + t.Fatalf("namespaces: hits %d, want 14 (one per namespace, then a cache hit)", s.hits.Load()) + } +} + +// A response too large to share reaches its caller in full and frees its +// slots: with one slot per host, three GETs in a row each complete. +func TestLargeResponseNotSharedReleasesSlots(t *testing.T) { + p := open + p.MaxPerHost = 1 + g := New(p) + big := strings.Repeat("x", 3<<20) + s := newServer(t, 0, func(_ int64, w http.ResponseWriter, _ *http.Request) { _, _ = io.WriteString(w, big) }) + for i := range 3 { + ctx, cancel := context.WithTimeout(task("t1"), 5*time.Second) + if _, body := do(t, g, ctx, "GET", s.URL); len(body) != len(big) { + t.Fatalf("GET %d: body %d bytes, want %d", i+1, len(body), len(big)) + } + cancel() + } + if s.hits.Load() != 3 { + t.Fatalf("hits %d, want 3", s.hits.Load()) } } diff --git a/enrich/enrich.go b/enrich/enrich.go index e3821cf..2bbaee3 100644 --- a/enrich/enrich.go +++ b/enrich/enrich.go @@ -120,7 +120,7 @@ func buildClient(proxy func() string) *http.Client { } } return &http.Client{ - Transport: egress.Shared().Transport(tr), + Transport: egress.Shared().Transport("enrich", tr), Timeout: httpTimeout, // cap redirects; keep them within scope by re-checking at probe time CheckRedirect: func(*http.Request, []*http.Request) error { return http.ErrUseLastResponse }, diff --git a/server/customtool.go b/server/customtool.go index d39acfa..18e58b3 100644 --- a/server/customtool.go +++ b/server/customtool.go @@ -464,7 +464,7 @@ func (s *Server) runHTTPTool(ctx context.Context, key string, execRaw json.RawMe if tr := s.httpToolTransport(ctx, key, spec); tr != nil { base = tr } - client.Transport = egress.Shared().Transport(base) + client.Transport = egress.Shared().Transport(key, base) resp, err := client.Do(req) if err != nil { return actool.Errorf("request failed: " + err.Error()), nil