diff --git a/agent/webfetch.go b/agent/webfetch.go index ad323c8..fe65b46 100644 --- a/agent/webfetch.go +++ b/agent/webfetch.go @@ -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 diff --git a/agent/webfetch_egress_test.go b/agent/webfetch_egress_test.go new file mode 100644 index 0000000..540f967 --- /dev/null +++ b/agent/webfetch_egress_test.go @@ -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()) + } +} diff --git a/agent/webfetch_redirect_test.go b/agent/webfetch_redirect_test.go index 0243aec..0d2bc7a 100644 --- a/agent/webfetch_redirect_test.go +++ b/agent/webfetch_redirect_test.go @@ -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) })) diff --git a/egress/egress.go b/egress/egress.go index 94c3926..b91262b 100644 --- a/egress/egress.go +++ b/egress/egress.go @@ -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 diff --git a/egress/egress_test.go b/egress/egress_test.go index d28e65c..660a400 100644 --- a/egress/egress_test.go +++ b/egress/egress_test.go @@ -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()) } }