Egress: return large unshared responses without an error, namespace the transport cache per client, pace WebFetch only

.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=

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.
 

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
 }

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,50 +305,42 @@ 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)
+	release, err := g.acquire(ctx, HostKey(u))
 	if err != nil {
-			return nil, 0, err
+		var zero T
+		return zero, err
 	}
 	defer release()
-		v, err := fn()
-		return v, 0, 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
+	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
+	namespace string
 	base      http.RoundTripper
 }
 
@@ -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

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

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 },

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