Add agentgateway to CI & rename dataplane flag (#1598)

Run agentgateway data plane tests as a part of substrate CI
(non-blocking to start so we can confirm it's not flaky). Also, change
the `--atenet-router` flag to `--atenet-dataplane` to make it clearer
that the flag controls ingress and egress.

I've run the e2es locally across gVisor and microVM plus the MITM
variants for both. The only skip we do for agentgateway is
`TestIngressProtocolDowngrade` because 1. the behavior its testing only
exists on the non-CONNECT atunnel ingress path and agentgateway only
sends CONNECT to atunnel and 2. I'm not sure that we want this to be a
part of the contract that substrate is bound by (e.g. do we really want
to commit to atunnel always parsing HTTP?).

My goal with getting both dataplanes into CI is to start taking steps to
codify the proxy (router + egress PEP) contract for substrate. The
telemetry they emit, atunnel expectations, etc. are all important
contracts to explicitly call out so that they don't become too coupled
to a single dataplane implementation.

> It's a good idea to open an issue first for discussion.

- [X] Tests pass
- [X] Appropriate changes to documentation are included in the PR

---------

Signed-off-by: Keith Mattix II <keithmattix2@gmail.com>

.github/workflows/pr-workflow.yaml

@@ -56,8 +56,20 @@ jobs:
   # One kind cluster exercises BOTH runtimes. Free x86-64 ubuntu-latest runners
   # expose /dev/kvm (with a udev rule), so create-kind-cluster.sh mounts it and the
   # micro-VM (kata + cloud-hypervisor) sandbox class works alongside gVisor.
-  e2e-test:
+  e2e-test-matrix:
+    name: E2E (${{ matrix.dataplane }})
     runs-on: ubuntu-latest
+    continue-on-error: ${{ matrix.experimental }} # TODO: Make AgentGateway required once tests show stability
+    strategy:
+      fail-fast: false
+      matrix:
+        include:
+        - dataplane: envoy
+          experimental: false
+        - dataplane: agentgateway
+          experimental: true
+    env:
+      E2E_ATENET_DATAPLANE: ${{ matrix.dataplane }}
     steps:
     - name: Checkout
       uses: actions/checkout@fbc6f3992d24b796d5a048ff273f7fcc4a7b6c09 # v5.1.0
@@ -86,8 +98,10 @@ jobs:
         sudo udevadm trigger --name-match=kvm
     - name: Create cluster
       run: hack/create-kind-cluster.sh
-    - name: Install Agent Substrate
-      run: hack/install-ate-kind.sh --deploy-ate-system
+    - name: Install Agent Substrate (${{ matrix.dataplane }})
+      # The dataplane selection applies to both the ingress router and egress
+      # gateway.
+      run: hack/install-ate-kind.sh --deploy-ate-system --atenet-dataplane=${{ matrix.dataplane }}
     - name: Enable NFS
       # Load NFS kernel modules so in-cluster NFS server and CSI driver can run.
       run: |
@@ -97,7 +111,10 @@ jobs:
       run: hack/install-ate-kind.sh --setup-csi=nfs
     - name: Deploy micro-VM counter demo
       # Stages the (cached) assets into the cluster's rustfs and deploys the
-      # counter-microvm demo onto the control plane installed above.
+      # counter-microvm demo onto the control plane installed above. The demo
+      # redeploys the control plane, so retain the selected dataplane.
+      env:
+        ATE_ATENET_DATAPLANE: ${{ matrix.dataplane }}
       run: hack/run-microvm-demo-kind.sh
     - name: Deploy gVisor counter demo
       run: hack/install-ate-kind.sh --deploy-demo-counter
@@ -127,7 +144,7 @@ jobs:
       # Cluster-wide, so it must come AFTER the standard lanes: once egress
       # TLS is intercepted, their passthrough assumptions
       # (TestActorEgressHTTPS's end-to-end TLS with the origin) no longer hold.
-      run: hack/install-ate-kind.sh --deploy-atenet --experimental-use-sdsmint
+      run: hack/install-ate-kind.sh --deploy-atenet --atenet-dataplane=${{ matrix.dataplane }} --experimental-use-sdsmint
     - name: Run E2E tests (egress MITM trust)
       # The consumption half of the trust-bundle chain: an actor does TLS with
       # the MITM gateway's minted leaf using ONLY the projected bundle, plus a
@@ -178,3 +195,15 @@ jobs:
         kubectl --context kind-kind get pods -A -l ate.dev/worker-pool \
           -o 'custom-columns=:.metadata.namespace,:.metadata.name' --no-headers 2>/dev/null \
           | while read -r ns name; do dump "$ns" "$name"; done
+  # Preserve the required-check name while the concrete Envoy and AgentGateway
+  # executions run as entries in the shared matrix above.
+  e2e-test:
+    name: e2e-test
+    needs: e2e-test-matrix
+    if: always()
+    runs-on: ubuntu-latest
+    steps:
+    - name: Require E2E matrix success
+      env:
+        MATRIX_RESULT: ${{ needs.e2e-test-matrix.result }}
+      run: test "$MATRIX_RESULT" = success

benchmarking/automation/tests.yaml

@@ -43,7 +43,7 @@
 #                  outrun the default.
 #   ateArgs        Optional. List of strings appended verbatim to
 #                  `hack/install-ate.sh --deploy-ate-system` when this
-#                  test's substrate is deployed (e.g. ["--atenet-router=agentgateway"]
+#                  test's substrate is deployed (e.g. ["--atenet-dataplane=agentgateway"]
 #                  or ["--experimental-use-sdsmint"]).
 #
 # ---- type: locust -----------------------------------------------------------

cmd/ate-setup/commands.md

@@ -22,7 +22,7 @@ a pre-scan pass, so they may appear anywhere on its command line.
 | `ate-setup` | `hack/install-ate.sh` | Notes |
 |---|---|---|
 | `--kind` | `hack/install-ate-kind.sh`, or `ATE_INSTALL_KIND=true` | Kind overlays, the local registry, and host-architecture image builds |
-| `--atenet-router envoy\|agentgateway` | `--atenet-router envoy\|agentgateway` | atenet router dataplane (default `envoy`) |
+| `--atenet-dataplane envoy\|agentgateway` | `--atenet-dataplane envoy\|agentgateway` | atenet ingress and egress dataplane (default `envoy`) |
 | `--rollout-timeout DURATION` | `--rollout-timeout DURATION` | Readiness timeout for workloads (default `60s`). Unlike the shell flag it also governs the podcertificate-controller and CSI waits, which stay at their 120s default until it is passed |
 | `--podcert-workers-per-signer N` | `--podcert-workers-per-signer N` | Concurrent workers per podcertificate-controller signer |
 | `--experimental-use-sdsmint` | `--experimental-use-sdsmint` | Mint TLS certificates on-demand via SDS in atenet egress gateway |

cmd/ate-setup/differences.md

@@ -48,14 +48,14 @@ These were treated as contracts and reproduced exactly:
 | Actions per run | many, one per flag, in command-line order | exactly one subcommand |
 | Argument errors | detected when the dispatch loop reaches the flag, after earlier actions already ran | rejected by cobra before anything runs |
 | Value flags | pre-scanned so they could appear anywhere | positional, per-command, standard flag parsing |
-| Configuration | env vars (`ATE_INSTALL_KIND`, `ATE_ATENET_ROUTER`, `KUBECTL_CONTEXT`, …) | flags, with the env vars still honored as defaults |
+| Configuration | env vars (`ATE_INSTALL_KIND`, `ATE_ATENET_DATAPLANE`, `KUBECTL_CONTEXT`, …) | flags, with the env vars still honored as defaults |
 | Repository root | `git rev-parse --show-toplevel`, then `cd` | walk up for `go.mod`; no `chdir`, all paths absolute |
 
 The one-action-per-run change is the most visible: a line that passed
 `--deploy-ate-system --deploy-demo-counter` becomes two `ate-setup` calls.
 `hack/install-ate.sh` still accepts the combined form.
 
-Invalid input now fails before any cluster mutation. `--atenet-router=nginx`
+Invalid input now fails before any cluster mutation. `--atenet-dataplane=nginx`
 used to be caught by a pre-scan validation pass; `--worker-count 0` was not
 caught at all and surfaced from inside `deploy_locust.sh` after the microvm
 dependencies had already been installed.

cmd/ate-setup/internal/cmd/deploy.go

@@ -43,7 +43,7 @@ not supported here — the ATE_API_POSTGRES_CLOUDSQL_* variables are ignored, so
 use hack/install-ate.sh for a Cloud SQL install (see
 cmd/ate-setup/differences.md).
 
-Shape the install with the global --atenet-router flag.`,
+Shape the install with the global --atenet-dataplane flag.`,
 	Args: cobra.NoArgs,
 	RunE: func(cmd *cobra.Command, _ []string) error {
 		return env.DeployAteSystem(cmd.Context(), deployOpts)

cmd/ate-setup/internal/cmd/root.go

@@ -88,7 +88,7 @@ func init() {
 		"Target a local Kind cluster: use the kind overlays, the local registry, and host-architecture builds")
 	f.StringVar(&opts.Kubeconfig, "kubeconfig", "", "Path to the kubeconfig file")
 	f.StringVar(&opts.Context, "context", "", "Name of the kubeconfig context to use (defaults to KUBECTL_CONTEXT)")
-	f.StringVar(&opts.Router, "atenet-router", "", "atenet router dataplane: envoy or agentgateway (default envoy)")
+	f.StringVar(&opts.Router, "atenet-dataplane", "", "Atenet ingress and egress dataplane: envoy or agentgateway (default envoy)")
 	f.StringVar(&opts.RolloutTimeout, "rollout-timeout", "", "Timeout for workload rollouts as a duration string (e.g. 60s, 5m)")
 	f.IntVar(&opts.PodcertWorkersPerSigner, "podcert-workers-per-signer", 0, "Number of worker goroutines per signer in podcertificate-controller")
 	f.BoolVar(&opts.ExperimentalUseSDSMint, "experimental-use-sdsmint", false, "Deploy egress gateway with dynamic per-SNI certificate minting")

cmd/ate-setup/internal/config/config.go

@@ -237,7 +237,7 @@ func Load(opts Options) (*Config, error) {
 		applyKindDefaults(cfg)
 	}
 
-	cfg.Router = firstNonEmpty(opts.Router, env["ATE_ATENET_ROUTER"], RouterEnvoy)
+	cfg.Router = firstNonEmpty(opts.Router, env["ATE_ATENET_DATAPLANE"], RouterEnvoy)
 
 	if err := validate(cfg); err != nil {
 		return nil, err
@@ -283,7 +283,7 @@ func validate(cfg *Config) error {
 			return fmt.Errorf("--experimental-additional-egress-extproc-service requires --experimental-use-sdsmint")
 		}
 		if cfg.Router != RouterEnvoy {
-			return fmt.Errorf("--experimental-additional-egress-extproc-service requires --atenet-router=envoy")
+			return fmt.Errorf("--experimental-additional-egress-extproc-service requires --atenet-dataplane=envoy")
 		}
 	}
 	return nil

cmd/ate-setup/internal/config/config_test.go

@@ -30,7 +30,7 @@ import (
 //
 // Load deliberately reads the developer's environment, so a test that sets only
 // what it cares about is at the mercy of whatever the shell or CI job happens
-// to export: an ambient PROJECT_ID or ATE_ATENET_ROUTER quietly changes the
+// to export: an ambient PROJECT_ID or ATE_ATENET_DATAPLANE quietly changes the
 // result. Every variable Load consults is blanked here -- empty reads as unset,
 // which is what these tests mean by "not configured" -- and NO_DEV_ENV keeps
 // .ate-dev-env.sh out of it. Tests then set back only what they exercise.
@@ -42,7 +42,7 @@ func loadEnv(t *testing.T) {
 		"ATE_ADDITIONAL_EGRESS_EXTPROC_SERVICE",
 		"ATE_API_POSTGRES_CONNECTION_STRING",
 		"ATE_API_POSTGRES_SCHEMA",
-		"ATE_ATENET_ROUTER",
+		"ATE_ATENET_DATAPLANE",
 		"ATE_EXPERIMENTAL_USE_SDSMINT",
 		"ATE_IMAGE_REPO",
 		"ATE_IMAGE_TAG",
@@ -85,7 +85,7 @@ func TestLoadDefaults(t *testing.T) {
 
 func TestLoadFlagsBeatEnvironment(t *testing.T) {
 	loadEnv(t)
-	t.Setenv("ATE_ATENET_ROUTER", RouterEnvoy)
+	t.Setenv("ATE_ATENET_DATAPLANE", RouterEnvoy)
 	t.Setenv("ATE_INSTALL_ROLLOUT_TIMEOUT", "30s")
 
 	cfg, err := Load(Options{Router: RouterAgentgateway, RolloutTimeout: "120s"})

cmd/ate-setup/internal/steps/overlay.go

@@ -75,7 +75,7 @@ func (e *Env) atenetEgressManifestPath() string {
 func (e *Env) renderAtenetEgressManifest(ctx context.Context) ([]byte, error) {
 	if e.Cfg.Router == config.RouterAgentgateway {
 		if e.Cfg.AdditionalEgressExtprocService != "" {
-			return nil, fmt.Errorf("--experimental-additional-egress-extproc-service requires --atenet-router=envoy")
+			return nil, fmt.Errorf("--experimental-additional-egress-extproc-service requires --atenet-dataplane=envoy")
 		}
 		return e.KustomizeResolve(ctx, installDir+"/agentgateway-egress")
 	}

cmd/atenet/internal/router/README.md

@@ -2,8 +2,8 @@
 
 Router has several responsibilities:
 
-* Serves Envoy xDS configuration when `--atenet-router=envoy` (the default).
-  With `--atenet-router=agentgateway`, the sidecar uses a static ConfigMap and
+* Serves Envoy xDS configuration when `--atenet-dataplane=envoy` (the default).
+  With `--atenet-dataplane=agentgateway`, the sidecar uses a static ConfigMap and
   atenet does not start an xDS server.
 * ext_proc server for the dataplane. To make the deployment and debugging easier, we will run this component together
   with the router, but this will be split later into its own component.
@@ -185,7 +185,7 @@ Ingress and egress are deployed separately today — `atenet-router` fronts the
 ingress dataplane, `atenet-egress` the egress gateway — because the two scale
 independently, not because they need separate binaries.
 
-`--atenet-router` selects the dataplane for both Deployments. Each gateway has
+`--atenet-dataplane` selects the dataplane for both Deployments. Each gateway has
 its own static configuration because ingress and egress scale independently.
 
 ## status page

cmd/atenet/internal/router/cmd.go

@@ -45,7 +45,7 @@ func NewRouterCmd() *cobra.Command {
 	cmd.Flags().StringVar((*string)(&cfg.Mode), "mode", string(ModeAll), fmt.Sprintf("Traffic direction this instance serves: %q (also runs the ingress control plane — the xDS server — for an Envoy dataplane), %q (ext_proc only, needs no Kubernetes access), or %q for both. The ext_proc mux refuses a direction this instance was not started to serve rather than falling back to the other one", ModeIngress, ModeEgress, ModeAll))
 	cmd.Flags().StringVar(&cfg.LogLevel, "log-level", "info", "Log level: debug, info, warn, error")
 	cmd.Flags().StringVar(&cfg.MetricsAddr, "metrics-listen-addr", ":9090", "Address and port the prometheus metrics server should listen on.")
-	cmd.Flags().StringVar(&cfg.AtenetRouter, "atenet-router", string(atenetRouterEnvoy), "Router dataplane: envoy or agentgateway")
+	cmd.Flags().StringVar(&cfg.AtenetRouter, "atenet-dataplane", string(atenetRouterEnvoy), "Atenet ingress and egress dataplane: envoy or agentgateway")
 	cmd.Flags().StringVar(&cfg.Namespace, "namespace", "default", "Target operations namespace")
 	cmd.Flags().StringVar(&cfg.Kubeconfig, "kubeconfig", "", "Absolute path to the kubeconfig configuration file")
 	cmd.Flags().StringVar(&cfg.AteapiAddr, "ateapi-address", "k8s:///api.ate-system.svc:443", "gRPC dial target for the cluster ateapi Control instance.")
@@ -84,7 +84,7 @@ func NewRouterCmd() *cobra.Command {
 	// must propagate to the Service endpoints before the drain starts.
 	cmd.Flags().DurationVar(&cfg.DrainDelay, "drain-delay", 13*time.Second, "How long to keep serving after SIGTERM before starting the drain, covering readiness-probe detection and Service endpoint propagation")
 	cmd.Flags().DurationVar(&cfg.DrainTimeout, "drain-timeout", 0, "Deadline for the ext_proc drain on shutdown; streams still open past it (parked requests included) are forcefully cancelled. 0 (the default) derives --parked-request-budget + the actor route timeout + margin so parked requests always finish normally. Explicit values must be >= --parked-request-budget")
-	cmd.Flags().StringVar(&cfg.EnvoyAdminAddr, "envoy-admin-address", "localhost:9901", "Envoy admin interface the shutdown sequence drives to drain the sidecar (healthcheck/fail, drain_listeners, stats polling). Ignored with --atenet-router=agentgateway")
+	cmd.Flags().StringVar(&cfg.EnvoyAdminAddr, "envoy-admin-address", "localhost:9901", "Envoy admin interface the shutdown sequence drives to drain the sidecar (healthcheck/fail, drain_listeners, stats polling). Ignored with --atenet-dataplane=agentgateway")
 	cmd.Flags().StringVar(&cfg.DrainCompleteFile, "drain-complete-file", defaultDrainCompleteFile, "Marker file created (on a pod-shared emptyDir) once the shutdown drain completes; the dataplane container's preStop hook polls for it so the proxy exits as soon as — and no sooner than — the drain is done. Removed at startup to defuse stale markers. Empty disables the handshake")
 
 	return cmd

cmd/atenet/internal/router/config.go

@@ -216,7 +216,7 @@ func (c routerConfig) validate() error {
 	switch c.atenetRouter() {
 	case atenetRouterEnvoy, atenetRouterAgentgateway:
 	default:
-		return fmt.Errorf("--atenet-router must be %q or %q, got %q", atenetRouterEnvoy, atenetRouterAgentgateway, c.AtenetRouter)
+		return fmt.Errorf("--atenet-dataplane must be %q or %q, got %q", atenetRouterEnvoy, atenetRouterAgentgateway, c.AtenetRouter)
 	}
 	switch c.Mode {
 	case "", ModeIngress, ModeEgress, ModeAll:

cmd/atenet/internal/router/config_test.go

@@ -43,7 +43,7 @@ func TestRouterConfigValidate(t *testing.T) {
 		{
 			name:    "unknown router rejected",
 			cfg:     routerConfig{AtenetRouter: "blah"},
-			wantErr: "--atenet-router must be",
+			wantErr: "--atenet-dataplane must be",
 		},
 		{
 			name:    "negative extproc-max-requests rejected",

demos/egress/README.md

@@ -61,12 +61,12 @@ ActorTemplate, worker pool, test, and manual walkthrough are otherwise the same.
 ./hack/install-ate-kind.sh --deploy-ate-system
 
 # agentgateway
-./hack/install-ate-kind.sh --deploy-ate-system --atenet-router=agentgateway
+./hack/install-ate-kind.sh --deploy-ate-system --atenet-dataplane=agentgateway
 ```
 
 | | Envoy | agentgateway |
 | --- | --- | --- |
-| Select with | `--atenet-router=envoy` (default) | `--atenet-router=agentgateway` |
+| Select with | `--atenet-dataplane=envoy` (default) | `--atenet-dataplane=agentgateway` |
 | Egress routing | Dynamic forward proxy | Dynamic backend from CONNECT authority |
 | Actor authentication | Co-located atenet `ext_proc` | Built-in `substrateEgress` policy |
 | Configuration | Envoy bootstrap in `atenet-egress.yaml` | Static agentgateway ConfigMap overlay |

hack/install-ate.sh

@@ -77,7 +77,7 @@ function usage() {
   echo "                                         a bare --setup-csi means nfs; hostpath is Kind only)"
   echo "  --delete-ate-system                    Delete core system"
   echo "  --delete-all                           Delete core system and all registered demos"
-  echo "  --atenet-router=envoy|agentgateway     Select the ingress and egress dataplane (default: envoy)"
+  echo "  --atenet-dataplane=envoy|agentgateway  Select the atenet ingress and egress dataplane (default: envoy)"
   echo "  --podcert-workers-per-signer N         Concurrent workers per podcertificate-controller signer (default: 1)"
   echo "  --rollout-timeout DURATION             Per-workload readiness wait timeout, kubectl-style Go duration (default: 60s)"
   echo "  --otlp-endpoint URL                    Send all control plane telemetry to URL, not to the cluster default (see benchmarking/telemetry/README.md)"
@@ -218,12 +218,12 @@ run_ko() {
 }
 
 atenet_router() {
-  case "${ATE_ATENET_ROUTER:-envoy}" in
+  case "${ATE_ATENET_DATAPLANE:-envoy}" in
     envoy|agentgateway)
-      echo "${ATE_ATENET_ROUTER:-envoy}"
+      echo "${ATE_ATENET_DATAPLANE:-envoy}"
       ;;
     *)
-      echo "Error: --atenet-router must be envoy or agentgateway, got '${ATE_ATENET_ROUTER}'" >&2
+      echo "Error: --atenet-dataplane must be envoy or agentgateway, got '${ATE_ATENET_DATAPLANE}'" >&2
       exit 1
       ;;
   esac
@@ -362,7 +362,7 @@ render_atenet_egress_manifest() {
     # refuses a non-sdsmint manifest: ignoring the flag would report a
     # successful install of a gateway that has no additional checkpoint on it.
     if additional_egress_extproc_enabled; then
-      echo "Error: --experimental-additional-egress-extproc-service requires --atenet-router=envoy" >&2
+      echo "Error: --experimental-additional-egress-extproc-service requires --atenet-dataplane=envoy" >&2
       return 1
     fi
     local agentgateway_egress="manifests/ate-install/agentgateway-egress"
@@ -1432,13 +1432,13 @@ BENCHMARK_ACTOR_MEMORY=""
 prescan_args=("$@")
 for ((i = 0; i < ${#prescan_args[@]}; i++)); do
   case "${prescan_args[i]}" in
-    --atenet-router=*) ATE_ATENET_ROUTER="${prescan_args[i]#*=}" ;;
-    --atenet-router)
+    --atenet-dataplane=*) ATE_ATENET_DATAPLANE="${prescan_args[i]#*=}" ;;
+    --atenet-dataplane)
       if (( i + 1 >= ${#prescan_args[@]} )); then
-        echo "Error: --atenet-router requires envoy or agentgateway" >&2
+        echo "Error: --atenet-dataplane requires envoy or agentgateway" >&2
         exit 1
       fi
-      ATE_ATENET_ROUTER="${prescan_args[$((i + 1))]}"
+      ATE_ATENET_DATAPLANE="${prescan_args[$((i + 1))]}"
       ;;
     --experimental-use-sdsmint) ATE_EXPERIMENTAL_USE_SDSMINT=true ;;
     --experimental-additional-egress-extproc-service=*)
@@ -1540,14 +1540,14 @@ while [[ "$#" -gt 0 ]]; do
   fi
 
   case $1 in
-    --atenet-router=*) ATE_ATENET_ROUTER="${1#*=}" ;;
-    --atenet-router)
+    --atenet-dataplane=*) ATE_ATENET_DATAPLANE="${1#*=}" ;;
+    --atenet-dataplane)
       shift
       if [[ "$#" -eq 0 ]]; then
-        echo "Error: --atenet-router requires envoy or agentgateway" >&2
+        echo "Error: --atenet-dataplane requires envoy or agentgateway" >&2
         exit 1
       fi
-      ATE_ATENET_ROUTER="$1"
+      ATE_ATENET_DATAPLANE="$1"
       ;;
     # Captured in the pre-scan above; matched here only so the `*)` branch does
     # not reject it as an unknown option.

internal/e2e/atenet_dataplane.go

@@ -0,0 +1,162 @@
+// Copyright 2026 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package e2e
+
+import (
+	"context"
+	"fmt"
+	"net/http"
+	"os"
+	"strconv"
+	"strings"
+	"time"
+)
+
+// AtenetDataplaneEnv selects the dataplane exercised by an e2e lane.
+const AtenetDataplaneEnv = "E2E_ATENET_DATAPLANE"
+
+// AtenetDataplane captures the observable differences between the supported
+// atenet dataplanes. Suites use these operations rather than branching on the
+// selected implementation.
+type AtenetDataplane interface {
+	NewParkingObserver(context.Context) (ParkingObserver, error)
+	IsRetryableParkingBudgetExhaustion(status int, body string) bool
+	ParkingBudgetStatus() int
+	IsEgressPolicyDenied(status int, body string) bool
+	SupportsTLSPassthroughEgressPolicy() bool
+	PlatformMetricPrefixes([]string) []string
+	RouteDurationSeen(context.Context, string) (bool, error)
+	SupportsIngressProtocolDowngrade() bool
+}
+
+// CurrentAtenetDataplane returns the implementation selected for this test
+// process. Envoy remains the default to match installation defaults.
+func CurrentAtenetDataplane() AtenetDataplane {
+	if os.Getenv(AtenetDataplaneEnv) == "agentgateway" {
+		return agentGatewayAtenetDataplane{}
+	}
+	return envoyAtenetDataplane{}
+}
+
+// ParkingObserver waits for the dataplane's active parked-request gauge.
+type ParkingObserver interface {
+	WaitForCount(context.Context, func(int) bool) (int, error)
+	Close()
+}
+
+type envoyAtenetDataplane struct{}
+
+func (envoyAtenetDataplane) NewParkingObserver(ctx context.Context) (ParkingObserver, error) {
+	return NewStatuszClient(ctx)
+}
+
+func (envoyAtenetDataplane) IsRetryableParkingBudgetExhaustion(status int, body string) bool {
+	return status == http.StatusServiceUnavailable && strings.Contains(body, "no free workers available")
+}
+
+func (envoyAtenetDataplane) ParkingBudgetStatus() int { return http.StatusServiceUnavailable }
+
+func (envoyAtenetDataplane) IsEgressPolicyDenied(status int, body string) bool {
+	return (status == http.StatusForbidden && strings.Contains(body, "egress denied")) ||
+		(status == http.StatusBadGateway && strings.Contains(body, "request failed"))
+}
+
+func (envoyAtenetDataplane) SupportsTLSPassthroughEgressPolicy() bool { return true }
+
+func (envoyAtenetDataplane) PlatformMetricPrefixes(prefixes []string) []string { return prefixes }
+
+func (envoyAtenetDataplane) RouteDurationSeen(_ context.Context, collectorScrape string) (bool, error) {
+	return len(MissingPlatformMetrics(collectorScrape, []string{"atenet_router_route_duration"})) == 0, nil
+}
+
+func (envoyAtenetDataplane) SupportsIngressProtocolDowngrade() bool { return true }
+
+type agentGatewayAtenetDataplane struct{}
+
+func (agentGatewayAtenetDataplane) NewParkingObserver(context.Context) (ParkingObserver, error) {
+	return agentGatewayParkingObserver{}, nil
+}
+
+func (agentGatewayAtenetDataplane) IsRetryableParkingBudgetExhaustion(status int, body string) bool {
+	return status == http.StatusGatewayTimeout && strings.Contains(body, "request timed out")
+}
+
+func (agentGatewayAtenetDataplane) ParkingBudgetStatus() int { return http.StatusGatewayTimeout }
+
+func (agentGatewayAtenetDataplane) IsEgressPolicyDenied(status int, body string) bool {
+	return status == http.StatusForbidden && strings.Contains(body, "actor egress policy denied")
+}
+
+// TODO: Apply substrateEgress to TLS passthrough routes in AgentGateway.
+func (agentGatewayAtenetDataplane) SupportsTLSPassthroughEgressPolicy() bool { return false }
+
+func (agentGatewayAtenetDataplane) PlatformMetricPrefixes(prefixes []string) []string {
+	filtered := make([]string, 0, len(prefixes))
+	for _, prefix := range prefixes {
+		if prefix != "atenet_router_route_duration" {
+			filtered = append(filtered, prefix)
+		}
+	}
+	return filtered
+}
+
+func (agentGatewayAtenetDataplane) RouteDurationSeen(ctx context.Context, _ string) (bool, error) {
+	scrape, err := ScrapeAgentGatewayRouterMetrics(ctx)
+	if err != nil {
+		return false, err
+	}
+	return len(MissingPlatformMetrics(scrape, []string{"agentgateway_atenet_router_route_duration_seconds"})) == 0, nil
+}
+
+func (agentGatewayAtenetDataplane) SupportsIngressProtocolDowngrade() bool { return false }
+
+type agentGatewayParkingObserver struct{}
+
+func (agentGatewayParkingObserver) WaitForCount(ctx context.Context, cond func(int) bool) (int, error) {
+	deadline := time.Now().Add(4 * time.Second)
+	var last int
+	for time.Now().Before(deadline) {
+		active, err := agentGatewayParkingCount(ctx)
+		if err == nil {
+			last = active
+			if cond(active) {
+				return active, nil
+			}
+		}
+		time.Sleep(150 * time.Millisecond)
+	}
+	return last, fmt.Errorf("timed out waiting for the parking gauge to satisfy the condition")
+}
+
+func (agentGatewayParkingObserver) Close() {}
+
+func agentGatewayParkingCount(ctx context.Context) (int, error) {
+	scrape, err := ScrapeAgentGatewayRouterMetrics(ctx)
+	if err != nil {
+		return 0, err
+	}
+	for _, line := range strings.Split(scrape, "\n") {
+		fields := strings.Fields(line)
+		if len(fields) != 2 || !strings.HasPrefix(fields[0], "agentgateway_substrate_request_parking_active") {
+			continue
+		}
+		value, err := strconv.ParseFloat(fields[1], 64)
+		if err != nil {
+			return 0, err
+		}
+		return int(value), nil
+	}
+	return 0, fmt.Errorf("AgentGateway parking gauge not found")
+}

internal/e2e/atenet_dataplane_test.go

@@ -0,0 +1,50 @@
+// Copyright 2026 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package e2e
+
+import (
+	"net/http"
+	"testing"
+)
+
+func TestAtenetDataplaneEgressPolicyDenial(t *testing.T) {
+	t.Run("envoy", func(t *testing.T) {
+		t.Setenv(AtenetDataplaneEnv, "")
+		if !CurrentAtenetDataplane().IsEgressPolicyDenied(http.StatusBadGateway, "request failed") {
+			t.Error("Envoy CONNECT refusal was not recognized as an egress-policy denial")
+		}
+	})
+	t.Run("agentgateway", func(t *testing.T) {
+		t.Setenv(AtenetDataplaneEnv, "agentgateway")
+		if !CurrentAtenetDataplane().IsEgressPolicyDenied(http.StatusForbidden, "actor egress policy denied destination") {
+			t.Error("AgentGateway direct policy denial was not recognized")
+		}
+	})
+}
+
+func TestAtenetDataplaneTLSPassthroughEgressPolicy(t *testing.T) {
+	t.Run("envoy", func(t *testing.T) {
+		t.Setenv(AtenetDataplaneEnv, "")
+		if !CurrentAtenetDataplane().SupportsTLSPassthroughEgressPolicy() {
+			t.Error("Envoy TLS passthrough egress policy was not supported")
+		}
+	})
+	t.Run("agentgateway", func(t *testing.T) {
+		t.Setenv(AtenetDataplaneEnv, "agentgateway")
+		if CurrentAtenetDataplane().SupportsTLSPassthroughEgressPolicy() {
+			t.Error("AgentGateway TLS passthrough egress policy unexpectedly reported support")
+		}
+	})
+}

internal/e2e/collector_metrics.go

@@ -31,6 +31,9 @@ const (
 	collectorNamespace = "otel-system"
 	collectorService   = "opentelemetry-collector"
 	collectorPromPort  = 8889
+	// AgentGateway exposes native Prometheus metrics; it does not export these
+	// instruments through the OTLP collector.
+	agentGatewayRouterStatsPort = 15020
 )
 
 // PlatformMetricPrefixes are the Prometheus metric-name prefixes (OTLP dots
@@ -51,6 +54,42 @@ var PlatformMetricPrefixes = []string{
 	"ate_scheduler_eligible_workers",
 }
 
+// ScrapeAgentGatewayRouterMetrics reads the AgentGateway router's native
+// Prometheus stats endpoint. AgentGateway instruments are not OTLP exports.
+func ScrapeAgentGatewayRouterMetrics(ctx context.Context) (string, error) {
+	config, err := ateclient.LoadKubeConfig(KubeConfig, KubeContext)
+	if err != nil {
+		return "", fmt.Errorf("loading kubeconfig: %w", err)
+	}
+	clientset, err := kubernetes.NewForConfig(config)
+	if err != nil {
+		return "", fmt.Errorf("creating k8s client: %w", err)
+	}
+	localPort, stop, err := portforward.ServicePortForward(ctx, config, clientset, routerNamespace, routerService, agentGatewayRouterStatsPort)
+	if err != nil {
+		return "", err
+	}
+	defer stop()
+
+	req, err := http.NewRequestWithContext(ctx, http.MethodGet, fmt.Sprintf("http://127.0.0.1:%d/metrics", localPort), nil)
+	if err != nil {
+		return "", err
+	}
+	resp, err := (&http.Client{Timeout: 30 * time.Second}).Do(req)
+	if err != nil {
+		return "", fmt.Errorf("scraping AgentGateway metrics: %w", err)
+	}
+	defer resp.Body.Close()
+	body, err := io.ReadAll(resp.Body)
+	if err != nil {
+		return "", fmt.Errorf("reading AgentGateway metrics: %w", err)
+	}
+	if resp.StatusCode != http.StatusOK {
+		return "", fmt.Errorf("AgentGateway metrics returned %d: %s", resp.StatusCode, body)
+	}
+	return string(body), nil
+}
+
 // ScrapeCollectorMetrics port-forwards the kind stack's OTel Collector and reads
 // its Prometheus exporter surface, returning the raw exposition text.
 func ScrapeCollectorMetrics(ctx context.Context) (string, error) {

internal/e2e/statusz.go

@@ -94,6 +94,23 @@ func (c *StatuszClient) Parking(ctx context.Context) (*ParkingStatusz, error) {
 	return &dashboard.Parking, nil
 }
 
+// WaitForCount polls the request-parking gauge until cond holds.
+func (c *StatuszClient) WaitForCount(ctx context.Context, cond func(int) bool) (int, error) {
+	deadline := time.Now().Add(4 * time.Second)
+	var last int
+	for time.Now().Before(deadline) {
+		parking, err := c.Parking(ctx)
+		if err == nil {
+			last = parking.Active
+			if cond(last) {
+				return last, nil
+			}
+		}
+		time.Sleep(150 * time.Millisecond)
+	}
+	return last, fmt.Errorf("timed out waiting for the parking gauge to satisfy the condition")
+}
+
 // Close tears down the port-forward.
 func (c *StatuszClient) Close() {
 	if c.stop != nil {

internal/e2e/suites/metrics/metrics_test.go

@@ -64,7 +64,7 @@ func TestPlatformMetricsEmitted(t *testing.T) {
 	// they add the drive steps their instruments need.
 	resume(t, ctx, clients, actorID)
 
-	// Drive request through the router so Envoy ext_proc emits atenet_router_route_duration.
+	// Drive request through the router so the dataplane emits atenet_router_route_duration.
 	rClient, err := e2e.NewRouterClient(ctx)
 	if err != nil {
 		t.Fatalf("NewRouterClient: %v", err)
@@ -88,15 +88,21 @@ func TestPlatformMetricsEmitted(t *testing.T) {
 	triggerActorCrash(t, ctx, clients, actorID)
 
 	deadline := time.Now().Add(2 * time.Minute)
+	dataplane := e2e.CurrentAtenetDataplane()
+	prefixes := dataplane.PlatformMetricPrefixes(e2e.PlatformMetricPrefixes)
 	var missing []string
-	var ateomSeen, controllerSeen bool
+	var ateomSeen, controllerSeen, routeDurationSeen bool
 	var lastLabelErr error
 	for time.Now().Before(deadline) {
 		scrape, err := e2e.ScrapeCollectorMetrics(ctx)
 		if err != nil {
 			t.Fatalf("ScrapeCollectorMetrics: %v", err)
 		}
-		missing = e2e.MissingPlatformMetrics(scrape, e2e.PlatformMetricPrefixes)
+		missing = e2e.MissingPlatformMetrics(scrape, prefixes)
+		routeDurationSeen, err = dataplane.RouteDurationSeen(ctx, scrape)
+		if err != nil {
+			t.Fatalf("checking route-duration metric: %v", err)
+		}
 		ateomSeen = e2e.CollectorHasService(scrape, "ateom-gvisor", "ateom-microvm")
 		// atecontroller bridges controller-runtime's Prometheus registry onto its OTLP
 		// reader, so the reconcile families are what prove the bridge, not just that
@@ -105,7 +111,7 @@ func TestPlatformMetricsEmitted(t *testing.T) {
 		controllerSeen = e2e.CollectorHasService(scrape, "atecontroller") &&
 			strings.Contains(scrape, "controller_runtime_")
 
-		if len(missing) == 0 && ateomSeen && controllerSeen {
+		if len(missing) == 0 && ateomSeen && controllerSeen && routeDurationSeen {
 			var errs []string
 
 			// Verify ate_workerpool_desired_workers carries required namespaced attributes.
@@ -301,8 +307,8 @@ func TestPlatformMetricsEmitted(t *testing.T) {
 		t.Fatalf("platform telemetry validation failed: missing metrics %v, ateom pushed=%v, atecontroller pushed=%v, error detail: %v",
 			missing, ateomSeen, controllerSeen, lastLabelErr)
 	}
-	t.Fatalf("platform telemetry never reached the collector: missing metrics %v, ateom pushed=%v, atecontroller pushed=%v",
-		missing, ateomSeen, controllerSeen)
+	t.Fatalf("platform telemetry validation failed: collector missing metrics %v, AgentGateway route duration seen=%v, ateom pushed=%v, atecontroller pushed=%v",
+		missing, routeDurationSeen, ateomSeen, controllerSeen)
 }
 
 func triggerActorCrash(t *testing.T, ctx context.Context, clients *e2e.Clients, actorID string) {

internal/e2e/suites/networking/egresspolicy_test.go

@@ -55,6 +55,7 @@ func reached(status int, _ []byte) bool { return status == http.StatusOK }
 // the same server.
 func TestActorEgressPolicyDeniesUnlistedHost(t *testing.T) {
 	ctx := context.Background()
+	dataplane := e2e.CurrentAtenetDataplane()
 	origin := egressHTTPTarget()
 	target := e2e.DeployServerPod(t, ctx, origin)
 	allowed := fmt.Sprintf("%s.%s.svc.cluster.local", origin.Name, target.Namespace)
@@ -72,17 +73,16 @@ func TestActorEgressPolicyDeniesUnlistedHost(t *testing.T) {
 
 	url = fmt.Sprintf("http://%s/healthz", target.Address())
 	status, body = fetchThroughEgressActorUntil(t, ctx, router, actorRef, url, notTransient)
-	if status != http.StatusForbidden || !strings.Contains(string(body), "egress denied") {
-		t.Fatalf("fetch of the same origin by address %s returned HTTP %d, want 403 egress denied; body: %s", url, status, body)
+	if !dataplane.IsEgressPolicyDenied(status, string(body)) {
+		t.Fatalf("fetch of the same origin by address %s returned HTTP %d, want an egress-policy denial; body: %s", url, status, body)
 	}
 	t.Logf("fetch by address was denied as expected: %s", body)
 }
 
-// TestActorEgressRequiresPolicy: an actor with no EgressPolicy gets no tunnel.
-// The CONNECT is refused, so the demo app reports a 502, not a 403, on both
-// gateways.
+// TestActorEgressRequiresPolicy: an actor with no EgressPolicy is denied.
 func TestActorEgressRequiresPolicy(t *testing.T) {
 	ctx := context.Background()
+	dataplane := e2e.CurrentAtenetDataplane()
 	target := e2e.DeployServerPod(t, ctx, egressHTTPTarget())
 
 	actorName, _ := createAndResumeActorWithEgress(t, ctx, "egress-nopolicy", egressFixture())
@@ -97,18 +97,19 @@ func TestActorEgressRequiresPolicy(t *testing.T) {
 
 	url := fmt.Sprintf("http://%s/healthz", target.Address())
 	status, body := fetchThroughEgressActorUntil(t, ctx, router, actorRef, url, notTransient)
-	if status != http.StatusBadGateway || !strings.Contains(string(body), "request failed") {
-		t.Fatalf("fetch by an actor with no policy returned HTTP %d, want 502 from a refused tunnel; body: %s", status, body)
+	if !dataplane.IsEgressPolicyDenied(status, string(body)) {
+		t.Fatalf("fetch by an actor with no policy returned HTTP %d, want an egress-policy denial; body: %s", status, body)
 	}
-	t.Logf("tunnel was refused as expected: %s", body)
+	t.Logf("egress was denied as expected: %s", body)
 }
 
 // TestActorEgressPolicyAllowsByAddress: the policy names the origin's address
 // only. Both gateways allow the fetch by address, and the same origin by name,
 // because the request is checked against the address the actor dialed and
-// sent there. A name that resolves to any other address gets no tunnel.
+// sent there. A name that resolves to any other address is denied.
 func TestActorEgressPolicyAllowsByAddress(t *testing.T) {
 	ctx := context.Background()
+	dataplane := e2e.CurrentAtenetDataplane()
 	origin := egressHTTPTarget()
 	target := e2e.DeployServerPod(t, ctx, origin)
 	block := netip.MustParseAddr(target.ClusterIP)
@@ -132,14 +133,13 @@ func TestActorEgressPolicyAllowsByAddress(t *testing.T) {
 	}
 
 	// The API server's ClusterIP is outside the block, and the policy has no
-	// hostname rule that could allow a request inside, so the CONNECT itself
-	// is refused: the demo app reports a 502, not a 403.
+	// hostname rule that could allow a request inside, so the request is denied.
 	url = "http://kubernetes.default.svc.cluster.local/healthz"
 	status, body = fetchThroughEgressActorUntil(t, ctx, router, actorRef, url, notTransient)
-	if status != http.StatusBadGateway || !strings.Contains(string(body), "request failed") {
-		t.Fatalf("fetch of an address outside the policy %s returned HTTP %d, want 502 from a refused tunnel; body: %s", url, status, body)
+	if !dataplane.IsEgressPolicyDenied(status, string(body)) {
+		t.Fatalf("fetch of an address outside the policy %s returned HTTP %d, want an egress-policy denial; body: %s", url, status, body)
 	}
-	t.Logf("tunnel to an address outside the policy was refused as expected: %s", body)
+	t.Logf("egress to an address outside the policy was denied as expected: %s", body)
 }
 
 // hostnamePolicyActor creates an actor whose policy names example.com and
@@ -161,6 +161,7 @@ func TestActorEgressHTTPSByHostnameMITM(t *testing.T) {
 		t.Skip("covers the sdsmint gateway; set E2E_EGRESS_MITM")
 	}
 	ctx := context.Background()
+	dataplane := e2e.CurrentAtenetDataplane()
 	router, actorRef := hostnamePolicyActor(t, ctx)
 
 	status, body := fetchThroughEgressActorUntil(t, ctx, router, actorRef, "https://example.com/", reached)
@@ -168,8 +169,8 @@ func TestActorEgressHTTPSByHostnameMITM(t *testing.T) {
 		t.Fatalf("fetch of the allowed host returned HTTP %d, want 200; body: %s", status, body)
 	}
 	status, body = fetchThroughEgressActorUntil(t, ctx, router, actorRef, "https://example.org/", notTransient)
-	if status != http.StatusForbidden || !strings.Contains(string(body), "egress denied") {
-		t.Fatalf("fetch of a host outside the policy returned HTTP %d, want 403 egress denied; body: %s", status, body)
+	if !dataplane.IsEgressPolicyDenied(status, string(body)) {
+		t.Fatalf("fetch of a host outside the policy returned HTTP %d, want an egress-policy denial; body: %s", status, body)
 	}
 	t.Logf("denied on the decrypted request: %s", body)
 }
@@ -182,6 +183,9 @@ func TestActorEgressHTTPSByHostnamePassthrough(t *testing.T) {
 	if egressMITM() {
 		t.Skip("covers the plain gateway; sdsmint is TestActorEgressHTTPSByHostnameMITM")
 	}
+	if !e2e.CurrentAtenetDataplane().SupportsTLSPassthroughEgressPolicy() {
+		t.Skip("TODO: AgentGateway must enforce substrateEgress for TLS passthrough")
+	}
 	ctx := context.Background()
 	router, actorRef := hostnamePolicyActor(t, ctx)
 

internal/e2e/suites/networking/grpcingress_test.go

@@ -57,6 +57,9 @@ var grpcEchoFixtureManifests = e2e.SubstrateFixtureManifests{
 // TestIngressGRPC below is the positive counterpart: the same path, against an
 // actor that really does speak gRPC.
 func TestIngressProtocolDowngrade(t *testing.T) {
+	if !e2e.CurrentAtenetDataplane().SupportsIngressProtocolDowngrade() {
+		t.Skip("TODO: is HTTP/2-to-HTTP/1 downgrade, and rejecting gRPC for HTTP/1-only actors, an AgentGateway ingress contract?")
+	}
 	ctx := context.Background()
 	actorName, _ := createAndResumeSubstrateActor(t, ctx, "protodowngrade", e2e.SubstrateCounterFixture())
 	actorRef := resources.ActorRef{Atespace: networkingAtespace, Name: actorName}

internal/e2e/suites/networking/networking_test.go

@@ -267,38 +267,46 @@ func postThroughEgressActorUntil(t *testing.T, ctx context.Context, router *e2e.
 	}
 }
 
-// assertEgressGatewayConnect waits for the atenet-egress access log to show a
-// CONNECT to port opened by actorName.
+// assertEgressGatewayConnect waits for the atenet-egress log to show a CONNECT
+// to port opened by actorName. Envoy logs authenticated peer details in its
+// successful CONNECT access record; AgentGateway logs the terminated tunnel.
 func assertEgressGatewayConnect(t *testing.T, ctx context.Context, since metav1.Time, actorName, port string) {
 	t.Helper()
 	want := fmt.Sprintf("a CONNECT to port %s by actor %s", port, actorName)
-	waitForAccessLog(t, ctx, since, want, func(lines []string) (bool, error) {
+	waitForAccessLog(t, ctx, since, want, func(lines []gatewayAccessLogLine) bool {
 		for _, line := range lines {
-			authority, ok := accessLogField(line, "authority")
-			if !ok || !strings.HasSuffix(authority, ":"+port) {
-				continue
+			switch line.container {
+			case "envoy":
+				authority, ok := accessLogField(line.text, "authority")
+				if ok && strings.HasSuffix(authority, ":"+port) && strings.Contains(line.text, "/actor/"+actorName) {
+					t.Logf("egress gateway tunneled the request: %s", line.text)
+					return true
+				}
+			case "agentgateway":
+				if strings.Contains(line.text, "CONNECT tunnel terminated") &&
+					strings.Contains(line.text, "target=") &&
+					strings.Contains(line.text, ":"+port) {
+					t.Logf("egress gateway tunneled the request: %s", line.text)
+					return true
+				}
 			}
-			if !strings.Contains(line, "/actor/"+actorName) {
-				continue
-			}
-			t.Logf("egress gateway tunneled the request: %s", line)
-			return true, nil
 		}
-		return false, nil
+		return false
 	})
 }
 
+type gatewayAccessLogLine struct {
+	container string
+	text      string
+}
+
 // waitForAccessLog polls the atenet-egress access log, across every gateway
 // replica, until predicate accepts the lines written since.
-func waitForAccessLog(t *testing.T, ctx context.Context, since metav1.Time, want string, predicate func(lines []string) (bool, error)) {
+func waitForAccessLog(t *testing.T, ctx context.Context, since metav1.Time, want string, predicate func(lines []gatewayAccessLogLine) bool) {
 	t.Helper()
 	const (
 		gatewayNamespace = "ate-system"
 		gatewaySelector  = "app=atenet-egress"
-		gatewayContainer = "envoy"
-		// The access log's line prefix, from the HttpConnectionManager
-		// text_format_source in manifests/ate-install/atenet-egress.yaml.
-		accessLogPrefix = "[egress] "
 	)
 
 	clients := e2e.GetClients()
@@ -314,32 +322,43 @@ func waitForAccessLog(t *testing.T, ctx context.Context, since metav1.Time, want
 	const timeout = 30 * time.Second
 	deadline := time.Now().Add(timeout)
 	for {
-		var lines []string
+		var lines []gatewayAccessLogLine
 		for _, pod := range pods.Items {
+			container := ""
+			for _, candidate := range pod.Spec.Containers {
+				if candidate.Name == "envoy" || candidate.Name == "agentgateway" {
+					container = candidate.Name
+					break
+				}
+			}
+			if container == "" {
+				t.Fatalf("egress gateway pod %s has neither an Envoy nor AgentGateway container", pod.Name)
+			}
 			logs, err := clients.K8s.CoreV1().Pods(gatewayNamespace).GetLogs(pod.Name, &corev1.PodLogOptions{
-				Container: gatewayContainer,
+				Container: container,
 				SinceTime: &since,
 			}).DoRaw(ctx)
 			if err != nil {
 				t.Fatalf("reading logs of %s/%s: %v", gatewayNamespace, pod.Name, err)
 			}
 			for line := range strings.SplitSeq(string(logs), "\n") {
-				if strings.Contains(line, accessLogPrefix) {
-					lines = append(lines, line)
+				if (container == "envoy" && strings.Contains(line, "[egress] ")) ||
+					(container == "agentgateway" && (strings.Contains(line, "substrate.connect.authority=") ||
+						strings.Contains(line, "CONNECT tunnel terminated"))) {
+					lines = append(lines, gatewayAccessLogLine{container: container, text: line})
 				}
 			}
 		}
 
-		matched, err := predicate(lines)
-		if err != nil {
-			t.Fatalf("looking for %s in the atenet-egress access log: %v", want, err)
-		}
-		if matched {
+		if predicate(lines) {
 			return
 		}
 		if time.Now().After(deadline) {
-			t.Fatalf("no atenet-egress access-log line for %s after %v; lines seen:\n%s",
-				want, timeout, strings.Join(lines, "\n"))
+			seen := make([]string, 0, len(lines))
+			for _, line := range lines {
+				seen = append(seen, line.text)
+			}
+			t.Fatalf("no atenet-egress access-log line for %s after %v; lines seen:\n%s", want, timeout, strings.Join(seen, "\n"))
 		}
 		time.Sleep(1 * time.Second)
 	}

internal/e2e/suites/parking/parking_test.go

@@ -62,11 +62,12 @@ func TestRequestParking(t *testing.T) {
 		t.Fatalf("creating router client: %v", err)
 	}
 	defer router.Close()
-	statusz, err := e2e.NewStatuszClient(ctx)
+	dataplane := e2e.CurrentAtenetDataplane()
+	parking, err := dataplane.NewParkingObserver(ctx)
 	if err != nil {
-		t.Fatalf("creating statusz client: %v", err)
+		t.Fatalf("creating parking observer: %v", err)
 	}
-	defer statusz.Close()
+	defer parking.Close()
 
 	t.Run("ParkThenServed", func(t *testing.T) {
 		// Occupy the only worker with actor A.
@@ -104,8 +105,9 @@ func TestRequestParking(t *testing.T) {
 			}()
 			if attempt == 1 {
 				// Free the worker only once the request is observably parked —
-				// the statusz gauge, not a sleep, is the synchronization point.
-				waitForParkedCount(ctx, t, statusz, func(active int) bool { return active >= 1 })
+				// the dataplane's active-parking gauge, not a sleep, is the
+				// synchronization point.
+				waitForParkedCount(ctx, t, parking, func(active int) bool { return active >= 1 })
 				suspendActor(ctx, t, clients, actorA)
 			}
 			res = <-resCh
@@ -113,9 +115,9 @@ func TestRequestParking(t *testing.T) {
 			if res.err != nil {
 				t.Fatalf("parked request failed transport-level: %v", res.err)
 			}
-			if res.resp.StatusCode == http.StatusServiceUnavailable &&
-				strings.Contains(res.body, "no free workers available") && attempt < 3 {
-				t.Logf("attempt %d budget-exhausted while the worker was still freeing (503 after %v); retrying", attempt, elapsed)
+			retryableBudgetExhaustion := dataplane.IsRetryableParkingBudgetExhaustion(res.resp.StatusCode, res.body)
+			if retryableBudgetExhaustion && attempt < 3 {
+				t.Logf("attempt %d budget-exhausted while the worker was still freeing (HTTP %d after %v); retrying", attempt, res.resp.StatusCode, elapsed)
 				continue
 			}
 			break
@@ -146,7 +148,7 @@ func TestRequestParking(t *testing.T) {
 		}
 
 		// The slot must be released once served.
-		waitForParkedCount(ctx, t, statusz, func(active int) bool { return active == 0 })
+		waitForParkedCount(ctx, t, parking, func(active int) bool { return active == 0 })
 	})
 
 	t.Run("BudgetExhaustion", func(t *testing.T) {
@@ -163,14 +165,17 @@ func TestRequestParking(t *testing.T) {
 		defer resp.Body.Close()
 		body, _ := io.ReadAll(resp.Body)
 
-		if resp.StatusCode != http.StatusServiceUnavailable {
-			t.Fatalf("status = %d (body %q), want 503", resp.StatusCode, string(body))
+		wantStatus := dataplane.ParkingBudgetStatus()
+		if resp.StatusCode != wantStatus {
+			t.Fatalf("status = %d (body %q), want %d", resp.StatusCode, string(body), wantStatus)
 		}
-		if !strings.Contains(string(body), "no free workers available") {
+		if wantStatus == http.StatusServiceUnavailable && !strings.Contains(string(body), "no free workers available") {
 			t.Errorf("body = %q, want the router's capacity verdict", string(body))
 		}
-		if ct := resp.Header.Get("content-type"); ct != "text/plain" {
-			t.Errorf("content-type = %q, want text/plain", ct)
+		if wantStatus == http.StatusServiceUnavailable {
+			if ct := resp.Header.Get("content-type"); ct != "text/plain" {
+				t.Errorf("content-type = %q, want text/plain", ct)
+			}
 		}
 		// Lower bound proves the request parked (fail-fast would answer in
 		// milliseconds); upper bound proves the router's own verdict landed
@@ -181,7 +186,7 @@ func TestRequestParking(t *testing.T) {
 		if elapsed > routerParkBudget+4*time.Second {
 			t.Errorf("503 after %v: too slow, likely an Envoy timeout rather than the router's verdict", elapsed)
 		}
-		t.Logf("budget exhausted after %v", elapsed)
+		t.Logf("budget exhausted after %v with HTTP %d", elapsed, wantStatus)
 	})
 }
 
@@ -265,22 +270,13 @@ func waitForActorState(ctx context.Context, t *testing.T, clients *e2e.Clients,
 	t.Fatalf("timed out waiting for actor %q to reach %v", name, want)
 }
 
-// waitForParkedCount polls the router's statusz parking gauge until cond holds.
+// waitForParkedCount polls the dataplane's active-parking gauge until cond holds.
 // The deadline is short: a parking request becomes visible within its first
 // retry interval (~100ms), and a served one releases its slot immediately.
-func waitForParkedCount(ctx context.Context, t *testing.T, statusz *e2e.StatuszClient, cond func(active int) bool) {
+func waitForParkedCount(ctx context.Context, t *testing.T, parking e2e.ParkingObserver, cond func(active int) bool) {
 	t.Helper()
-	deadline := time.Now().Add(4 * time.Second)
-	var last int
-	for time.Now().Before(deadline) {
-		p, err := statusz.Parking(ctx)
-		if err == nil {
-			last = p.Active
-			if cond(p.Active) {
-				return
-			}
-		}
-		time.Sleep(150 * time.Millisecond)
+	last, err := parking.WaitForCount(ctx, cond)
+	if err != nil {
+		t.Fatalf("waiting for parking gauge (last active=%d): %v", last, err)
 	}
-	t.Fatalf("timed out waiting for the parking gauge to satisfy the condition (last active=%d)", last)
 }

manifests/ate-install/components/agentgateway-egress-mitm/kustomization.yaml

@@ -33,6 +33,13 @@ patches:
             accessLog:
               add:
                 substrate.connect.authority: source.connectHeaders["host"]
+            substrateEgressActorResolution:
+              host: api.ate-system.svc:443
+              policies:
+                backendTLS:
+                  cert: /run/podidentity.podcert.ate.dev/credential-bundle.pem
+                  key: /run/podidentity.podcert.ate.dev/credential-bundle.pem
+                  root: /run/servicedns-ca/trust-bundle.pem
 
           binds:
           - port: 8443
@@ -53,21 +60,11 @@ patches:
                 cert: /run/egress-mitm/tls.crt
                 key: /run/egress-mitm/tls.key
               routes:
-              - policies:
-                  substrateEgress:
-                    host: api.ate-system.svc:443
-                    policies:
-                      backendTLS:
-                        cert: /run/podidentity.podcert.ate.dev/credential-bundle.pem
-                        key: /run/podidentity.podcert.ate.dev/credential-bundle.pem
-                        root: /run/servicedns-ca/trust-bundle.pem
-                backends:
+              - backends:
                 - dynamic: {}
                   policies:
                     backendTLS: {}
-            - protocol: HTTP
-              routes:
-              - policies:
+                policies:
                   substrateEgress:
                     host: api.ate-system.svc:443
                     policies:
@@ -75,9 +72,19 @@ patches:
                         cert: /run/podidentity.podcert.ate.dev/credential-bundle.pem
                         key: /run/podidentity.podcert.ate.dev/credential-bundle.pem
                         root: /run/servicedns-ca/trust-bundle.pem
-                backends:
+            - protocol: HTTP
+              routes:
+              - backends:
                 - dynamic:
                     target: source.connectHeaders["host"]
+                policies:
+                  substrateEgress:
+                    host: api.ate-system.svc:443
+                    policies:
+                      backendTLS:
+                        cert: /run/podidentity.podcert.ate.dev/credential-bundle.pem
+                        key: /run/podidentity.podcert.ate.dev/credential-bundle.pem
+                        root: /run/servicedns-ca/trust-bundle.pem
             - protocol: TCP
               tcpRoutes:
               - backends:

manifests/ate-install/components/agentgateway/configmap.yaml

@@ -21,8 +21,7 @@ data:
   config.yaml: |
     # yaml-language-server: $schema=https://agentgateway.dev/schema/config
     config:
-      backend:
-        poolIdleTimeout: 5s
+      statsAddr: 0.0.0.0:15020
 
     frontendPolicies:
       tracing:
@@ -127,6 +126,13 @@ data:
       accessLog:
         add:
           substrate.connect.authority: source.connectHeaders["host"]
+      substrateEgressActorResolution:
+        host: api.ate-system.svc:443
+        policies:
+          backendTLS:
+            cert: /run/podidentity.podcert.ate.dev/credential-bundle.pem
+            key: /run/podidentity.podcert.ate.dev/credential-bundle.pem
+            root: /run/servicedns-ca/trust-bundle.pem
 
     binds:
     # Authenticate the actor before accepting CONNECT.
@@ -153,7 +159,10 @@ data:
               target: source.connectHeaders["host"]
       - protocol: HTTP
         routes:
-        - policies:
+        - backends:
+          - dynamic:
+              target: source.connectHeaders["host"]
+          policies:
             substrateEgress:
               host: api.ate-system.svc:443
               policies:
@@ -161,9 +170,6 @@ data:
                   cert: /run/podidentity.podcert.ate.dev/credential-bundle.pem
                   key: /run/podidentity.podcert.ate.dev/credential-bundle.pem
                   root: /run/servicedns-ca/trust-bundle.pem
-          backends:
-          - dynamic:
-              target: source.connectHeaders["host"]
       - protocol: TCP
         tcpRoutes:
         - backends:

manifests/ate-install/components/agentgateway/kustomization.yaml

@@ -42,7 +42,7 @@ patches:
         path: /spec/template/spec/containers/0
         value:
           name: agentgateway
-          image: cr.agentgateway.dev/agentgateway:v1.5.0
+          image: ghcr.io/agentgateway/agentgateway:v0.0.0-alpha.9f9744cf
           args:
           - -f
           - /etc/agentgateway/config.yaml
@@ -99,6 +99,9 @@ patches:
         namespace: ate-system
       spec:
         template:
+          metadata:
+            annotations:
+              prometheus.io/port: "15020"
           spec:
             volumes:
             - name: envoy-config
@@ -115,10 +118,13 @@ patches:
         path: /spec/template/spec/containers/0
         value:
           name: agentgateway
-          image: cr.agentgateway.dev/agentgateway:v1.5.0
+          image: ghcr.io/agentgateway/agentgateway:v0.0.0-alpha.9f9744cf
           args:
           - -f
           - /etc/agentgateway/config.yaml
+          env:
+          - name: RUST_LOG
+            value: debug
           ports:
           - name: https
             containerPort: 8443
@@ -170,3 +176,16 @@ patches:
                   matchLabels:
                     podcert.ate.dev/canarying: live
                 path: trust-bundle.pem
+  - target:
+      version: v1
+      kind: Service
+      name: atenet-router
+      namespace: ate-system
+    patch: |-
+      - op: add
+        path: /spec/ports/-
+        value:
+          name: stats
+          port: 15020
+          targetPort: stats
+          protocol: TCP