WebFetch: coalesce only, never cache; pin the gate-off redirect test to the off profile

This commit is contained in:
BaaRai
2026-11-12 16:33:14 +01:00
parent 0b59f4670b
commit 3b5b2c3b88
5 changed files with 72 additions and 28 deletions
+10 -6
View File
@@ -28,14 +28,19 @@ func withWebFetch(ctx context.Context, tools []actool.CoreTool, proxy, caCert st
gated := scoped && scope.Enabled != nil && scope.Enabled()
webFetch := actool.NewWebFetch(actool.WebFetchConfig{Proxy: proxy, CACert: caCert, CheckRedirect: intercept.RedirectCheck("WebFetch"),
DialContext: intercept.PinnedDialer("WebFetch", scoped), DisableDirectRetry: gated})
return append(tools[:len(tools):len(tools)], governedFetch{webFetch, egress.Shared()})
return append(tools[:len(tools):len(tools)], governedFetch{webFetch, egressGovernor()})
}
// egressGovernor is the governor of WebFetch; a test may pin it.
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 calls (WebFetch only sends GET) coalesced and cached. A failed call
// is not cached. WebFetch ignores tc, so a shared result fits every caller.
// 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.
// Norma owns the HTTP client, so redirect hops are not paced separately and a
// 429/503 is not retried.
type governedFetch struct {
@@ -54,9 +59,8 @@ 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, bool, error) {
r, err := f.CoreTool.Call(ctx, input, tc)
return r, err == nil && !r.IsError, err
r, err := egress.Do(f.g, ctx, u, string(input), 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
+42
View File
@@ -0,0 +1,42 @@
package agent
import (
"encoding/json"
"net/http"
"sync/atomic"
"testing"
"github.com/Autumn-27/artex/egress"
"github.com/Autumn-27/artex/intercept"
actool "github.com/Autumn-27/norma/tool"
)
// pinNoiseOff runs t with the off noise profile, whatever ARTEX_NOISE_PROFILE says.
func pinNoiseOff(t *testing.T) {
old := egressGovernor
egressGovernor = func() *egress.Governor { return egress.New(egress.Profile{Name: "off"}) }
t.Cleanup(func() { egressGovernor = old })
}
// Under a profile, two WebFetch calls on the same URL in one task both run the
// tool and its redirect gate: WebFetch results are never cached.
func TestWebFetchProfileDoesNotCache(t *testing.T) {
pair := newRedirectPair(t)
var checks atomic.Int64
inner := actool.NewWebFetch(actool.WebFetchConfig{CheckRedirect: func(*http.Request, []*http.Request) error {
checks.Add(1)
return nil
}})
tool := governedFetch{inner, egress.New(egress.Profiles["balanced"])}
ctx := intercept.WithTargetScope(t.Context(), intercept.TargetScope{TaskID: "7"})
for range 2 {
res, err := tool.Call(ctx, json.RawMessage(fetchInputJSON(pair.start)), &actool.ToolContext{})
if err != nil || res.IsError {
t.Fatalf("WebFetch: err=%v result=%q", err, res.Flatten())
}
}
if checks.Load() != 2 || pair.startHits.Load() != 2 || pair.finalHits.Load() != 2 {
t.Fatalf("redirect checks %d, start hits %d, final hits %d; want 2 each (no cache hit)",
checks.Load(), pair.startHits.Load(), pair.finalHits.Load())
}
}
+1
View File
@@ -193,6 +193,7 @@ func TestWebFetchGatedScopeHasNoDirectRetry(t *testing.T) {
// with agentcore's own tool: to the end of the chain, and the same error after
// 10 hops. No database.
func TestWebFetchRedirectGateOffUnchanged(t *testing.T) {
pinNoiseOff(t) // a gate-off regression test: independent of ARTEX_NOISE_PROFILE
loop := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
http.Redirect(w, r, r.URL.Path+"x", http.StatusFound)
}))
+8 -11
View File
@@ -305,13 +305,13 @@ 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 and caching under id within the task
// (keep reports whether the value may be cached). 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() (val T, keep bool, err error)) (T, error) {
// 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) {
if g.off() {
v, _, err := fn()
return v, err
return fn()
}
host := HostKey(u)
run := func() (any, time.Duration, error) {
@@ -320,11 +320,8 @@ func Do[T any](g *Governor, ctx context.Context, u *url.URL, id string, fn func(
return nil, 0, err
}
defer release()
v, keep, err := fn()
if !keep {
return v, 0, err
}
return v, g.p.CacheTTL, err
v, err := fn()
return v, 0, err
}
var v any
var err error
+11 -11
View File
@@ -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, bool, error) { calls.Add(1); return 0, true, nil })
_, _ = Do(g, task("t1"), u, "x", 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,18 @@ 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 and caches what fn keeps.
// WebFetch path: Do coalesces concurrent calls and caches nothing.
var calls atomic.Int64
u, _ := url.Parse(s.URL)
fn := func(keep bool) func() (string, bool, error) {
return func() (string, bool, error) { calls.Add(1); time.Sleep(50 * time.Millisecond); return "r", keep, nil }
}
parallel(3, func(int) { _, _ = Do(g, task("t1"), u, "a", fn(true)) })
_, _ = Do(g, task("t1"), u, "a", fn(true))
_, _ = Do(g, task("t1"), u, "b", fn(false))
_, _ = Do(g, task("t1"), u, "b", fn(false))
if calls.Load() != 3 {
t.Fatalf("Do calls = %d, want 3", calls.Load())
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" {
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())
}
}