diff --git a/demos/egress/bannerserver/main.go b/demos/egress/bannerserver/main.go
new file mode 100644
index 000000000..467e8ccbe
--- /dev/null
+++ b/demos/egress/bannerserver/main.go
@@ -0,0 +1,118 @@
+// 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.
+
+// Command bannerserver is a TCP origin for egress tests. It echoes what it is
+// sent on two ports: on one it greets the peer first, on the other it stays
+// silent until spoken to.
+//
+// Which port a test dials is how it selects the behavior. Nothing in the
+// request can select it, because the server has to decide whether to greet
+// before it has read anything -- that is what speaking first means.
+package main
+
+import (
+	"errors"
+	"io"
+	"log/slog"
+	"net"
+	"os"
+	"sync"
+	"time"
+)
+
+// Banner is what the server writes on accept. Tests match on it, so it is a
+// fixed string rather than anything derived from the connection.
+const Banner = "TESTBANNER/1.0\r\n"
+
+func main() {
+	slog.SetDefault(slog.New(slog.NewJSONHandler(os.Stdout, nil)))
+
+	greeting := listenerAddress("LISTEN_ADDRESS", ":2222")
+	quiet := listenerAddress("QUIET_LISTEN_ADDRESS", ":2223")
+
+	var group sync.WaitGroup
+	group.Add(2)
+	go func() { defer group.Done(); accept(greeting, true) }()
+	go func() { defer group.Done(); accept(quiet, false) }()
+	group.Wait()
+}
+
+func listenerAddress(variable, fallback string) string {
+	if address := os.Getenv(variable); address != "" {
+		return address
+	}
+	return fallback
+}
+
+// accept serves address until it stops accepting, greeting each peer first when
+// greet is set. A dead listener takes the process down rather than leaving it
+// half-serving: a test that reached the surviving port would pass while the
+// other silently answered nothing.
+func accept(address string, greet bool) {
+	listener, err := net.Listen("tcp", address)
+	if err != nil {
+		slog.Error("banner server failed to listen", "address", address, "error", err)
+		os.Exit(1)
+	}
+	slog.Info("starting banner server", "address", address, "greets", greet)
+
+	for {
+		connection, err := listener.Accept()
+		if err != nil {
+			slog.Error("banner server stopped accepting", "address", address, "error", err)
+			os.Exit(1)
+		}
+		go serve(connection, greet)
+	}
+}
+
+// serve echoes until the peer goes away, announcing itself first when greet is
+// set.
+func serve(connection net.Conn, greet bool) {
+	// A stuck peer must not hold a goroutine and a socket forever. Long enough
+	// that a tunneled round trip is never the thing that trips it.
+	const idleTimeout = 60 * time.Second
+
+	defer connection.Close()
+
+	if greet {
+		if err := connection.SetWriteDeadline(time.Now().Add(idleTimeout)); err != nil {
+			slog.Error("setting write deadline", "error", err)
+			return
+		}
+		if _, err := io.WriteString(connection, Banner); err != nil {
+			slog.Error("writing banner", "error", err)
+			return
+		}
+	}
+
+	buffer := make([]byte, 4<<10)
+	for {
+		if err := connection.SetDeadline(time.Now().Add(idleTimeout)); err != nil {
+			return
+		}
+		n, err := connection.Read(buffer)
+		if n > 0 {
+			if _, writeErr := connection.Write(buffer[:n]); writeErr != nil {
+				return
+			}
+		}
+		if err != nil {
+			if !errors.Is(err, io.EOF) && !errors.Is(err, os.ErrDeadlineExceeded) {
+				slog.Error("reading from peer", "error", err)
+			}
+			return
+		}
+	}
+}
diff --git a/demos/egress/demo.sh b/demos/egress/demo.sh
new file mode 100755
index 000000000..f12d5529f
--- /dev/null
+++ b/demos/egress/demo.sh
@@ -0,0 +1,452 @@
+#!/usr/bin/env bash
+
+# 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.
+
+# A presenter-driven walkthrough of the egress path. See README.md.
+#
+# This drives internal/e2e/fixtures/egressprobe deployed as an Actor. The probe
+# is a test fixture, not a product surface -- everything customer-facing here is
+# the narration and the formatting, and the raw Result is one --verbose away for
+# whichever engineer in the room asks for it.
+#
+# The demo makes no claim the suite does not already assert. What it adds is an
+# order to put the claims in, and a destination that echoes, so the injected
+# header can be read rather than inferred.
+
+set -euo pipefail
+
+ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/../.." && pwd)"
+cd "${ROOT}"
+
+# ---------------------------------------------------------------------------
+# Configuration
+# ---------------------------------------------------------------------------
+
+# The atespace and actor names are NOT free choices. internal/extproc/hardcoded.go
+# keys the policy table on the atespace/name pair, so these strings are what
+# select a policy. Renaming an actor here silently selects no policy at all,
+# which presents as a denied CONNECT rather than as a configuration error.
+ATESPACE="acme-prod"
+ACTOR_BARE="actor-without-github-access"   # KindAllowByHostname     -- allowlist, no credential
+ACTOR_INJECT="actor-with-github-access"     # KindBasicCredentialInject -- allowlist + injection
+ACTOR_DENIED="quarantined"     # KindDenyAll             -- no egress at all
+
+TEMPLATE_NS="ate-demo-egress"
+TEMPLATE_NAME="egressprobe"
+# Three actors on stage at once, plus a spare to absorb a worker still draining.
+# The fixture ships with 2 because the e2e suite runs one actor at a time; a live
+# demo cannot afford a suspend/resume between acts.
+POOL_REPLICAS=4
+
+ROUTER="127.0.0.1:8080"
+# TEST-NET-1, and deliberately unroutable: nothing on the internet answers it, so
+# any response at all proves the connection was intercepted and rebuilt as a
+# tunnel. A real destination IP would make a success ambiguous -- it could mean
+# the nftables REDIRECT never fired and the Actor simply dialed out.
+DESTINATION="192.0.2.1:443"
+
+ECHO_HOST="postman-echo.com"
+GITHUB_HOST="api.github.com"
+UNLISTED_HOST="example.com"    # in sdsmintd's --allow, in no actor's policy
+
+VERBOSE=0
+PAUSE=1
+WITH_ECHO=0
+
+# ---------------------------------------------------------------------------
+# Presentation
+# ---------------------------------------------------------------------------
+
+if [[ -t 1 ]] && [[ "${NO_COLOR:-}" == "" ]]; then
+  B=$'\033[1m'; DIM=$'\033[2m'; GREEN=$'\033[32m'; RED=$'\033[31m'
+  CYAN=$'\033[36m'; R=$'\033[0m'
+else
+  B=""; DIM=""; GREEN=""; RED=""; CYAN=""; R=""
+fi
+
+act_number=0
+
+# act prints the frame for one claim: what we are about to do and what should
+# happen. Stating the expected result BEFORE running is the whole difference
+# between a demo and a debugging session -- the audience gets to be right.
+act() {
+  act_number=$((act_number + 1))
+  echo
+  echo "${B}${CYAN}━━━ Act ${act_number} · ${1} ━━━${R}"
+  echo
+  echo "  ${2}"
+  echo
+  # read fails at EOF, which is normal when stdin is not a terminal -- under
+  # `set -e` that would end the demo rather than skip the pause.
+  [[ "${PAUSE}" == "1" ]] && { printf '  %s[enter]%s' "${DIM}" "${R}"; read -r || true; echo; }
+  return 0
+}
+
+say()     { echo "  ${1}"; }
+verdict() { echo; echo "  ${B}${GREEN}▸ ${1}${R}"; }
+nope()    { echo; echo "  ${B}${RED}▸ ${1}${R}"; }
+field()   { printf '    %-28s %s\n' "${1}" "${2}"; }
+note()    { echo "  ${DIM}${1}${R}"; }
+die()     { echo "${RED}error:${R} ${1}" >&2; exit 1; }
+
+# ---------------------------------------------------------------------------
+# Driving the probe
+# ---------------------------------------------------------------------------
+
+# probe ACTOR SNI PATH -- one trip through the gateway, as that actor.
+#
+# via=direct is what makes this an Actor test rather than a proxy test: the probe
+# dials as if no gateway existed, ateom's nftables rule REDIRECTs it, and atunnel
+# builds the tunnel with a certificate the probe never sees. Nothing in the
+# workload knows a gateway is involved.
+probe() {
+  local actor="${1}" sni="${2}" path="${3}" body result
+  body=$(printf '{"via":"direct","destination":"%s","sni":"%s","request_path":"%s"}' \
+    "${DESTINATION}" "${sni}" "${path}")
+
+  if [[ "${VERBOSE}" == "1" ]]; then
+    echo "  ${DIM}POST http://${ROUTER}/probe${R}"
+    echo "  ${DIM}Host: ${actor}.${ATESPACE}.actors.resources.substrate.ate.dev${R}"
+    echo "  ${DIM}${body}${R}"
+    echo
+  fi
+
+  # The router routes on Host, not on a path or a header: this is the same way
+  # every Actor in the system is reached.
+  result=$(curl -sS --max-time 90 -X POST "http://${ROUTER}/probe" \
+    -H "Host: ${actor}.${ATESPACE}.actors.resources.substrate.ate.dev" \
+    -H 'Content-Type: application/json' \
+    -d "${body}") || die "the probe API did not answer -- is the port-forward still up?"
+
+  if [[ "${result}" != "{"* ]]; then
+    die "router answered with something that is not a Result: ${result}"
+  fi
+  if [[ "${VERBOSE}" == "1" ]]; then
+    echo "${result}" | jq . | sed 's/^/  /'
+    echo
+  fi
+  echo "${result}"
+}
+
+# echo_headers pulls the request headers back out of an echo service's response.
+# Empty when the body is not the JSON we expect, which keeps a Cloudflare error
+# page from being reported as "no headers arrived".
+echo_headers() {
+  echo "${1}" | jq -r 'try (.http_body | fromjson | .headers) // empty'
+}
+
+# ---------------------------------------------------------------------------
+# Acts
+# ---------------------------------------------------------------------------
+
+act_zero() {
+  act "Nothing up my sleeve" \
+"Before anything runs: this workload holds no credentials.
+  Two files say so -- the code that builds the request, and the template that
+  deploys it."
+
+  say "${B}The request the workload sends${R} ${DIM}(probeapi.go)${R}"
+  echo
+  sed -n '108,110p' internal/e2e/fixtures/egressprobe/probeapi/probeapi.go | sed 's/^/    /'
+  echo
+  say "${B}The ActorTemplate that deploys it${R} ${DIM}(egressprobe-actor.yaml.tmpl)${R}"
+  echo
+  note "    no secretKeyRef, no token env var, no credential volume:"
+  echo
+  grep -n 'secretKeyRef\|env:\|volumeMounts:' \
+    internal/e2e/fixtures/egressprobe/egressprobe-actor.yaml.tmpl \
+    | sed 's/^/    /' || say "    ${GREEN}(no matches -- there are none)${R}"
+
+  verdict "The workload cannot authenticate to anything. Keep that in mind."
+}
+
+act_echo_bare() {
+  act "An Actor with no credential, seen from the far end" \
+"${ACTOR_BARE} is allowed to reach ${ECHO_HOST}, and its policy attaches nothing.
+  The echo service reports the request exactly as it received it -- so this is
+  the baseline: what the workload sent, and only that."
+
+  local result headers
+  result=$(probe "${ACTOR_BARE}" "${ECHO_HOST}" "/get")
+  headers=$(echo_headers "${result}")
+  [[ -n "${headers}" ]] || { nope "no echo body came back"; echo "${result}" | jq .; return; }
+
+  echo "${headers}" | jq -r 'to_entries[] | "    \(.key): \(.value)"' | grep -iv '^ *x-\|cf-\|cdn-' || true
+  echo
+
+  if echo "${headers}" | jq -e 'has("authorization")' >/dev/null; then
+    nope "An authorization header arrived. It should not have -- check the policy table."
+  else
+    verdict "No authorization header. The destination saw exactly what the workload sent."
+  fi
+}
+
+act_echo_inject() {
+  act "The same request, from a different Actor" \
+"${ACTOR_INJECT} now. Same image, same code, same request, same destination.
+  The only thing that changed is which Actor is making it."
+
+  local result headers auth
+  result=$(probe "${ACTOR_INJECT}" "${ECHO_HOST}" "/get")
+  headers=$(echo_headers "${result}")
+  [[ -n "${headers}" ]] || { nope "no echo body came back"; echo "${result}" | jq .; return; }
+
+  echo "${headers}" | jq -r 'to_entries[] | "    \(.key): \(.value)"' | grep -iv '^ *x-\|cf-\|cdn-' || true
+  echo
+
+  auth=$(echo "${headers}" | jq -r '.authorization // empty')
+  if [[ -z "${auth}" ]]; then
+    nope "No authorization header arrived. The injection did not fire."
+    return
+  fi
+
+  field "authorization" "${B}${auth}${R}"
+  echo
+  say "That header was added by the gateway. The workload never held that value,"
+  say "cannot read it, and has no way to discover it."
+  echo
+
+  # A security-minded audience asks this before you finish the sentence, so get
+  # ahead of it: proving a credential goes OUT is only half the claim if
+  # substrate-internal identity headers leak out alongside it.
+  say "${B}And what did not leak:${R}"
+  if echo "${headers}" | jq -e 'keys[] | select(startswith("x-ate-") or . == "x-forwarded-client-cert")' >/dev/null 2>&1; then
+    nope "substrate-internal headers reached the destination -- extprocd's header hygiene broke."
+  else
+    field "x-ate-*" "${GREEN}absent${R}"
+    field "x-forwarded-client-cert" "${GREEN}absent${R}"
+  fi
+
+  verdict "A credential the workload never had, and nothing about substrate, reached the destination."
+}
+
+act_github() {
+  act "A real third party reacts to it" \
+"An echo service will print anything. ${GITHUB_HOST} has an opinion.
+  Both Actors ask GitHub the same question -- who am I? -- and GitHub
+  distinguishes 'you sent nothing' from 'you sent a token and it is wrong'."
+
+  local bare inject
+  bare=$(probe "${ACTOR_BARE}" "${GITHUB_HOST}" "/user")
+  inject=$(probe "${ACTOR_INJECT}" "${GITHUB_HOST}" "/user")
+
+  say "${B}${ACTOR_BARE}${R} ${DIM}(no injection)${R}"
+  field "status" "$(echo "${bare}" | jq -r '.http_status')"
+  field "message" "$(echo "${bare}" | jq -r 'try (.http_body|fromjson.message) // .http_body')"
+  field "x-ratelimit-limit" "$(echo "${bare}" | jq -r '.http_headers["X-Ratelimit-Limit"][0] // "(absent)"')"
+  echo
+  say "${B}${ACTOR_INJECT}${R} ${DIM}(credential injected)${R}"
+  field "status" "$(echo "${inject}" | jq -r '.http_status')"
+  field "message" "$(echo "${inject}" | jq -r 'try (.http_body|fromjson.message) // .http_body')"
+  field "x-ratelimit-limit" "$(echo "${inject}" | jq -r '.http_headers["X-Ratelimit-Limit"][0] // "(absent)"')"
+  echo
+
+  note "Both are 401 because the demo ships a deliberately invalid token -- there is"
+  note "no live secret anywhere in this repo. Two things separate them."
+  note ""
+  note "The wording. \"Requires authentication\" is GitHub saying nothing arrived;"
+  note "\"Bad credentials\" is GitHub saying a bearer token arrived and was rejected."
+  note ""
+  note "The rate limit. An anonymous request gets a bucket -- 60/hour by source IP."
+  note "A rejected credential gets no bucket at all: GitHub cannot attribute it to"
+  note "an account, and will not call it anonymous either. The header disappears."
+
+  verdict "GitHub confirms independently that a credential reached it -- and only for one Actor."
+}
+
+act_denied_host() {
+  act "The credential is not a blank cheque" \
+"${ACTOR_INJECT} can reach ${GITHUB_HOST}. Here it tries ${UNLISTED_HOST}, which
+  the gateway will happily mint a certificate for but no policy permits.
+  Injection and destination control are the same decision, not two features."
+
+  local result
+  result=$(probe "${ACTOR_INJECT}" "${UNLISTED_HOST}" "/")
+
+  field "status" "$(echo "${result}" | jq -r '.http_status')"
+  field "body"   "$(echo "${result}" | jq -r '.http_body' | head -1)"
+  echo
+
+  if echo "${result}" | jq -e '.http_body | test("egress denied")' >/dev/null 2>&1; then
+    verdict "Refused by the gateway, naming the policy that refused it."
+  else
+    nope "That 403 did not come from the gateway -- read the body before believing it."
+  fi
+}
+
+act_quarantined() {
+  act "The floor" \
+"${ACTOR_DENIED}'s policy is deny-all. Same image again, same request.
+  Watch where this one fails: earlier than every previous act."
+
+  local result
+  result=$(probe "${ACTOR_DENIED}" "${GITHUB_HOST}" "/user")
+
+  field "connected"      "$(echo "${result}" | jq -r '.connected')"
+  field "handshake_ok"   "$(echo "${result}" | jq -r '.handshake_ok')"
+  field "handshake_error" "$(echo "${result}" | jq -r '.handshake_error // "none"')"
+  echo
+
+  note "connected: true is not egress. The REDIRECT is local, so the socket comes up"
+  note "inside the sandbox before atunnel has spoken to the gateway at all. The"
+  note "gateway then refuses, atunnel closes, and the TLS handshake dies on the reset."
+  note "Note the peer address: 192.0.2.1 is unroutable, so that reset came from"
+  note "inside the cluster. Nothing was ever dialed."
+
+  if [[ "$(echo "${result}" | jq -r '.handshake_ok')" == "false" ]]; then
+    verdict "No bytes left the sandbox. The same mechanism that grants also denies."
+  else
+    nope "The handshake completed. A deny-all Actor reached the internet -- stop the demo."
+  fi
+}
+
+closing() {
+  echo
+  echo "${B}${CYAN}━━━ What the customer just saw ━━━${R}"
+  echo
+  say "One image. One codebase. No credentials anywhere in the workload."
+  say "Four different outcomes, decided entirely by ${B}which Actor made the call${R}:"
+  echo
+  # Plain ASCII in the left column: printf pads by byte count, so a multibyte
+  # arrow here silently shifts the whole table one place left.
+  field "${ACTOR_BARE}"    "reaches the allowlist, bare"
+  field "${ACTOR_INJECT}"  "reaches the allowlist, with a credential it never held"
+  field "  ...to ${UNLISTED_HOST}" "refused, by name"
+  field "${ACTOR_DENIED}"  "never leaves the sandbox"
+  echo
+  say "There is no secret to steal from the workload, no config naming an identity,"
+  say "and no request parameter that selects one."
+  echo
+}
+
+# ---------------------------------------------------------------------------
+# Setup and teardown
+# ---------------------------------------------------------------------------
+
+kate() { go run ./cmd/kubectl-ate "$@"; }
+
+setup() {
+  echo "${B}Setting up${R}"
+  command -v jq >/dev/null || die "jq is required"
+
+  kubectl get ns "${TEMPLATE_NS}" >/dev/null 2>&1 \
+    || die "${TEMPLATE_NS} not found -- run: hack/install-ate.sh --deploy-demo-egress"
+
+  echo "  scaling the worker pool to ${POOL_REPLICAS} (three actors on stage at once)"
+  kubectl patch workerpool "${TEMPLATE_NAME}" -n "${TEMPLATE_NS}" --type=merge \
+    -p "{\"spec\":{\"replicas\":${POOL_REPLICAS}}}"
+
+  kate get atespaces "${ATESPACE}" >/dev/null 2>&1 || kate create atespace "${ATESPACE}"
+
+  for actor in "${ACTOR_BARE}" "${ACTOR_INJECT}" "${ACTOR_DENIED}"; do
+    echo "  ${actor}"
+    kate create actor "${actor}" -t "${TEMPLATE_NS}/${TEMPLATE_NAME}" -a "${ATESPACE}" 2>/dev/null || true
+    kate resume actor "${actor}" -a "${ATESPACE}"
+  done
+
+  echo
+  echo "  Start the port-forward in another shell, then run the demo:"
+  echo "    ${DIM}kubectl -n ate-system port-forward svc/atenet-router 8080:80${R}"
+  echo "    ${DIM}demos/egress/demo.sh${R}"
+}
+
+teardown() {
+  echo "${B}Tearing down${R}"
+  for actor in "${ACTOR_BARE}" "${ACTOR_INJECT}" "${ACTOR_DENIED}"; do
+    kate suspend actor "${actor}" -a "${ATESPACE}" 2>/dev/null || true
+    kate delete  actor "${actor}" -a "${ATESPACE}" 2>/dev/null || true
+  done
+  # Back to what the e2e fixture expects, so a later suite run is not surprised.
+  kubectl patch workerpool "${TEMPLATE_NAME}" -n "${TEMPLATE_NS}" --type=merge \
+    -p '{"spec":{"replicas":2}}'
+}
+
+# preflight fails loudly and early rather than partway through an act, which is a
+# bad time to discover an Actor never resumed.
+#
+# Reaching the router is not enough to check: it answers for an Actor that does
+# not exist too, just with a different body. Every Actor gets probed, because a
+# demo that dies at Act 5 has already wasted the audience's time.
+preflight() {
+  command -v jq >/dev/null || die "jq is required"
+
+  local actors=("${ACTOR_BARE}" "${ACTOR_INJECT}" "${ACTOR_DENIED}") actor answer
+  for actor in "${actors[@]}"; do
+    answer=$(curl -sS --max-time 10 "http://${ROUTER}/healthz" \
+      -H "Host: ${actor}.${ATESPACE}.actors.resources.substrate.ate.dev" 2>&1) \
+      || die "no answer from ${ROUTER}
+  Start it with: kubectl -n ate-system port-forward svc/atenet-router 8080:80"
+
+    case "${answer}" in
+      *"not found"*)
+        die "Actor ${ATESPACE}/${actor} does not exist.
+  Create all three with: demos/egress/demo.sh --setup" ;;
+      *"no healthy upstream"*|*"upstream connect error"*)
+        die "Actor ${ATESPACE}/${actor} exists but is not running.
+  Resume it with: go run ./cmd/kubectl-ate resume actor ${actor} -a ${ATESPACE}" ;;
+    esac
+  done
+}
+
+usage() {
+  cat <<EOF
+demos/egress/demo.sh -- a walkthrough of the egress path
+
+  --setup        create the three Actors and scale the pool
+  --teardown     delete them and restore the pool
+  --with-echo    include the echo-service acts (needs one-time config; see README)
+  --verbose      print each request and the full Result
+  --no-pause     do not wait for [enter] between acts (for recording)
+  --port PORT    router port-forward, default 8080
+EOF
+}
+
+main() {
+  while [[ $# -gt 0 ]]; do
+    case "${1}" in
+      --setup)     setup; exit 0 ;;
+      --teardown)  teardown; exit 0 ;;
+      --with-echo) WITH_ECHO=1 ;;
+      --verbose)   VERBOSE=1 ;;
+      --no-pause)  PAUSE=0 ;;
+      --port)      ROUTER="127.0.0.1:${2}"; shift ;;
+      -h|--help)   usage; exit 0 ;;
+      *)           usage; die "unknown flag ${1}" ;;
+    esac
+    shift
+  done
+
+  preflight
+
+  echo
+  echo "${B}The workload holds no credentials.${R}"
+  echo "${DIM}What an Actor can reach is decided by who it is, not by what it carries.${R}"
+
+  act_zero
+  if [[ "${WITH_ECHO}" == "1" ]]; then
+    act_echo_bare
+    act_echo_inject
+  else
+    echo
+    note "Skipping the echo acts -- the strongest ones. See README.md § Seeing the"
+    note "header, then re-run with --with-echo."
+  fi
+  act_github
+  act_denied_host
+  act_quarantined
+  closing
+}
+
+main "$@"
diff --git a/demos/egress/egress.yaml.tmpl b/demos/egress/egress.yaml.tmpl
index b37a1f5d6..0c02a93ce 100644
--- a/demos/egress/egress.yaml.tmpl
+++ b/demos/egress/egress.yaml.tmpl
@@ -53,3 +53,113 @@ spec:
     onPause: Full
     onCommit: Full
     location: gs://${BUCKET_NAME}/ate-demo-egress/
+
+---
+
+# A TCP origin for the non-HTTP egress tests.
+apiVersion: apps/v1
+kind: Deployment
+metadata:
+  name: bannerserver
+  namespace: ate-demo-egress
+  labels:
+    app: bannerserver
+spec:
+  replicas: 1
+  selector:
+    matchLabels:
+      app: bannerserver
+  template:
+    metadata:
+      labels:
+        app: bannerserver
+    spec:
+      containers:
+      - name: bannerserver
+        image: ko://github.com/agent-substrate/substrate/demos/egress/bannerserver
+        ports:
+        # Two ports, because whether the origin speaks first has to be selected
+        # by the address the actor dials: the server decides before it reads.
+        - containerPort: 2222
+          name: banner
+        - containerPort: 2223
+          name: quiet
+        resources:
+          limits:
+            cpu: 100m
+            memory: 64Mi
+          requests:
+            cpu: 10m
+            memory: 64Mi
+
+---
+
+apiVersion: v1
+kind: Service
+metadata:
+  name: bannerserver
+  namespace: ate-demo-egress
+spec:
+  selector:
+    app: bannerserver
+  ports:
+  - name: banner
+    port: 2222
+    targetPort: banner
+  - name: quiet
+    port: 2223
+    targetPort: quiet
+
+---
+
+# A long-lived-stream origin for the streaming-duration egress tests.
+# It serves cleartext HTTP.
+apiVersion: apps/v1
+kind: Deployment
+metadata:
+  name: streamserver
+  namespace: ate-demo-egress
+  labels:
+    app: streamserver
+spec:
+  replicas: 1
+  selector:
+    matchLabels:
+      app: streamserver
+  template:
+    metadata:
+      labels:
+        app: streamserver
+    spec:
+      containers:
+      - name: streamserver
+        image: ko://github.com/agent-substrate/substrate/demos/egress/streamserver
+        ports:
+        - containerPort: 8080
+          name: http
+        readinessProbe:
+          httpGet:
+            path: /readyz
+            port: http
+        resources:
+          limits:
+            cpu: 100m
+            memory: 64Mi
+          requests:
+            cpu: 10m
+            memory: 64Mi
+
+---
+
+apiVersion: v1
+kind: Service
+metadata:
+  name: streamserver
+  namespace: ate-demo-egress
+spec:
+  selector:
+    app: streamserver
+  ports:
+  - name: http
+    port: 8080
+    targetPort: http
diff --git a/demos/egress/main.go b/demos/egress/main.go
index 4c0288e12..36eaa3231 100644
--- a/demos/egress/main.go
+++ b/demos/egress/main.go
@@ -13,14 +13,20 @@
 // limitations under the License.
 
 // Command egress is a small HTTP service for demonstrating per-Actor egress
-// policy. It accepts a URL, fetches it, and returns the upstream response.
+// policy. It accepts a URL, fetches it, and returns the upstream response; on
+// /tcp it opens a raw TCP connection so that egress can be exercised with
+// something other than HTTP; and on /stream it holds a Server-Sent Events or
+// WebSocket stream open so that egress can be exercised with something other
+// than a request that finishes promptly.
 package main
 
 import (
 	"encoding/json"
+	"errors"
 	"fmt"
 	"io"
 	"log/slog"
+	"net"
 	"net/http"
 	"net/url"
 	"os"
@@ -101,9 +107,125 @@ func newHandler(client *http.Client) http.Handler {
 		}
 		writeJSON(w, response.StatusCode, fetchResponse{StatusCode: response.StatusCode, Body: string(body)})
 	})
+	mux.HandleFunc("/tcp", handleTCPProbe)
+	mux.HandleFunc("/stream", handleStreamProbe(newStreamProbeRegistry()))
 	return mux
 }
 
+// tcpProbeRequest asks for one raw TCP exchange.
+type tcpProbeRequest struct {
+	Address   string `json:"address"`
+	Send      string `json:"send,omitempty"`
+	ReadBytes int    `json:"readBytes,omitempty"`
+	Timeout   string `json:"timeout,omitempty"`
+}
+
+type tcpProbeResponse struct {
+	// Banner is whatever the peer sent before being spoken to.
+	Banner   string `json:"banner,omitempty"`
+	Received string `json:"received,omitempty"`
+	Error    string `json:"error,omitempty"`
+}
+
+// handleTCPProbe opens a TCP connection and reads before it writes.
+func handleTCPProbe(w http.ResponseWriter, r *http.Request) {
+	const defaultProbeTimeout = 5 * time.Second
+	// Enough for an SSH identification string or a test banner.
+	const defaultProbeReadBytes = 512
+	const maxProbeReadBytes = 8 << 10 // 8 KiB
+
+	if r.Method != http.MethodPost {
+		w.Header().Set("Allow", http.MethodPost)
+		writeTCPProbeJSON(w, http.StatusMethodNotAllowed, tcpProbeResponse{Error: "method must be POST"})
+		return
+	}
+
+	var input tcpProbeRequest
+	decoder := json.NewDecoder(http.MaxBytesReader(w, r.Body, maxRequestBody))
+	if err := decoder.Decode(&input); err != nil {
+		writeTCPProbeJSON(w, http.StatusBadRequest, tcpProbeResponse{Error: fmt.Sprintf("invalid JSON payload: %v", err)})
+		return
+	}
+	if _, _, err := net.SplitHostPort(input.Address); err != nil {
+		writeTCPProbeJSON(w, http.StatusBadRequest, tcpProbeResponse{Error: fmt.Sprintf("address must be host:port: %v", err)})
+		return
+	}
+
+	timeout := defaultProbeTimeout
+	if input.Timeout != "" {
+		parsed, err := time.ParseDuration(input.Timeout)
+		if err != nil {
+			writeTCPProbeJSON(w, http.StatusBadRequest, tcpProbeResponse{Error: fmt.Sprintf("invalid timeout: %v", err)})
+			return
+		}
+		timeout = parsed
+	}
+	readBytes := input.ReadBytes
+	if readBytes <= 0 {
+		readBytes = defaultProbeReadBytes
+	}
+	readBytes = min(readBytes, maxProbeReadBytes)
+
+	dialer := net.Dialer{Timeout: timeout}
+	connection, err := dialer.DialContext(r.Context(), "tcp", input.Address)
+	if err != nil {
+		writeTCPProbeJSON(w, http.StatusBadGateway, tcpProbeResponse{Error: fmt.Sprintf("dialing %s: %v", input.Address, err)})
+		return
+	}
+	defer connection.Close()
+
+	// Read before writing anything at all, so an empty banner really does mean
+	// the peer stayed silent.
+	banner, err := readWithTimeout(connection, readBytes, timeout)
+	if err != nil {
+		writeTCPProbeJSON(w, http.StatusBadGateway, tcpProbeResponse{Error: fmt.Sprintf("reading banner from %s: %v", input.Address, err)})
+		return
+	}
+
+	response := tcpProbeResponse{Banner: string(banner)}
+	if input.Send == "" {
+		writeTCPProbeJSON(w, http.StatusOK, response)
+		return
+	}
+
+	if err := connection.SetWriteDeadline(time.Now().Add(timeout)); err != nil {
+		writeTCPProbeJSON(w, http.StatusBadGateway, tcpProbeResponse{Error: fmt.Sprintf("setting write deadline: %v", err)})
+		return
+	}
+	if _, err := io.WriteString(connection, input.Send); err != nil {
+		writeTCPProbeJSON(w, http.StatusBadGateway, tcpProbeResponse{Error: fmt.Sprintf("writing to %s: %v", input.Address, err)})
+		return
+	}
+	received, err := readWithTimeout(connection, readBytes, timeout)
+	if err != nil {
+		writeTCPProbeJSON(w, http.StatusBadGateway, tcpProbeResponse{Error: fmt.Sprintf("reading reply from %s: %v", input.Address, err)})
+		return
+	}
+	response.Received = string(received)
+	writeTCPProbeJSON(w, http.StatusOK, response)
+}
+
+// readWithTimeout returns the bytes of a single read, capped at limit. A peer
+// that says nothing within timeout yields no bytes and no error, since silence
+// is a legitimate answer to "does this peer speak first?".
+func readWithTimeout(connection net.Conn, limit int, timeout time.Duration) ([]byte, error) {
+	if err := connection.SetReadDeadline(time.Now().Add(timeout)); err != nil {
+		return nil, fmt.Errorf("setting read deadline: %w", err)
+	}
+	buffer := make([]byte, limit)
+	n, err := connection.Read(buffer)
+	if err != nil && !errors.Is(err, os.ErrDeadlineExceeded) && !errors.Is(err, io.EOF) {
+		return nil, err
+	}
+	return buffer[:n], nil
+}
+
+func writeTCPProbeJSON(w http.ResponseWriter, status int, response tcpProbeResponse) {
+	w.Header().Set("Content-Type", "application/json")
+	w.WriteHeader(status)
+	_ = json.NewEncoder(w).Encode(response)
+}
+
 func validateURL(raw string) error {
 	parsed, err := url.Parse(raw)
 	if err != nil {
diff --git a/demos/egress/main_test.go b/demos/egress/main_test.go
index 6cd203732..f2de0c6a2 100644
--- a/demos/egress/main_test.go
+++ b/demos/egress/main_test.go
@@ -18,6 +18,7 @@ import (
 	"encoding/json"
 	"errors"
 	"io"
+	"net"
 	"net/http"
 	"net/http/httptest"
 	"strings"
@@ -102,6 +103,145 @@ func TestOutboundFailure(t *testing.T) {
 	}
 }
 
+// TestTCPProbeServerSpeaksFirst covers the ordering the probe exists for: the
+// peer's greeting is reported without the probe having written anything, and
+// the reply to Send comes back separately.
+func TestTCPProbeServerSpeaksFirst(t *testing.T) {
+	const banner = "TESTBANNER/1.0\r\n"
+	address := startTestPeer(t, func(connection net.Conn) {
+		if _, err := io.WriteString(connection, banner); err != nil {
+			return
+		}
+		buffer := make([]byte, 64)
+		n, err := connection.Read(buffer)
+		if err != nil {
+			return
+		}
+		_, _ = connection.Write(buffer[:n])
+	})
+
+	got := probe(t, tcpProbeRequest{Address: address, Send: "ping", Timeout: "5s"}, http.StatusOK)
+	if got.Banner != banner {
+		t.Errorf("banner = %q, want %q", got.Banner, banner)
+	}
+	if got.Received != "ping" {
+		t.Errorf("received = %q, want %q", got.Received, "ping")
+	}
+}
+
+// TestTCPProbeSilentPeer records that silence is a result, not an error: a
+// client-speaks-first peer yields an empty banner and a 200, so a test can tell
+// the two shapes apart.
+func TestTCPProbeSilentPeer(t *testing.T) {
+	address := startTestPeer(t, func(connection net.Conn) {
+		buffer := make([]byte, 64)
+		n, err := connection.Read(buffer)
+		if err != nil {
+			return
+		}
+		_, _ = connection.Write(buffer[:n])
+	})
+
+	got := probe(t, tcpProbeRequest{Address: address, Send: "ping", Timeout: "250ms"}, http.StatusOK)
+	if got.Banner != "" {
+		t.Errorf("banner = %q, want empty for a peer that does not speak first", got.Banner)
+	}
+	if got.Received != "ping" {
+		t.Errorf("received = %q, want %q", got.Received, "ping")
+	}
+}
+
+func TestTCPProbeReadBytesCapsTheBanner(t *testing.T) {
+	address := startTestPeer(t, func(connection net.Conn) {
+		_, _ = io.WriteString(connection, "0123456789")
+	})
+
+	got := probe(t, tcpProbeRequest{Address: address, ReadBytes: 4, Timeout: "5s"}, http.StatusOK)
+	if got.Banner != "0123" {
+		t.Errorf("banner = %q, want %q", got.Banner, "0123")
+	}
+}
+
+func TestTCPProbeInvalidRequests(t *testing.T) {
+	tests := []struct {
+		name   string
+		method string
+		body   string
+		status int
+	}{
+		{name: "method", method: http.MethodGet, body: `{}`, status: http.StatusMethodNotAllowed},
+		{name: "malformed JSON", method: http.MethodPost, body: `{`, status: http.StatusBadRequest},
+		{name: "address without port", method: http.MethodPost, body: `{"address":"example.com"}`, status: http.StatusBadRequest},
+		{name: "invalid timeout", method: http.MethodPost, body: `{"address":"127.0.0.1:9","timeout":"soon"}`, status: http.StatusBadRequest},
+	}
+
+	handler := newHandler(http.DefaultClient)
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			recorder := httptest.NewRecorder()
+			request := httptest.NewRequest(test.method, "/tcp", strings.NewReader(test.body))
+			handler.ServeHTTP(recorder, request)
+			if recorder.Code != test.status {
+				t.Errorf("status = %d, want %d; body = %s", recorder.Code, test.status, recorder.Body.String())
+			}
+		})
+	}
+}
+
+func TestTCPProbeDialFailure(t *testing.T) {
+	// Port 0 is not connectable, so this fails without depending on which
+	// ports happen to be free.
+	got := probe(t, tcpProbeRequest{Address: "127.0.0.1:0", Timeout: "2s"}, http.StatusBadGateway)
+	if got.Error == "" {
+		t.Error("error = empty, want a dial failure")
+	}
+}
+
+// startTestPeer listens on loopback and hands each connection to serve. It
+// returns the address to probe.
+func startTestPeer(t *testing.T, serve func(net.Conn)) string {
+	t.Helper()
+	listener, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatalf("listening: %v", err)
+	}
+	t.Cleanup(func() { listener.Close() })
+
+	go func() {
+		for {
+			connection, err := listener.Accept()
+			if err != nil {
+				return
+			}
+			go func() {
+				defer connection.Close()
+				serve(connection)
+			}()
+		}
+	}()
+	return listener.Addr().String()
+}
+
+func probe(t *testing.T, input tcpProbeRequest, wantStatus int) tcpProbeResponse {
+	t.Helper()
+	payload, err := json.Marshal(input)
+	if err != nil {
+		t.Fatal(err)
+	}
+	recorder := httptest.NewRecorder()
+	request := httptest.NewRequest(http.MethodPost, "/tcp", strings.NewReader(string(payload)))
+	newHandler(http.DefaultClient).ServeHTTP(recorder, request)
+
+	if recorder.Code != wantStatus {
+		t.Fatalf("status = %d, want %d; body = %s", recorder.Code, wantStatus, recorder.Body.String())
+	}
+	var got tcpProbeResponse
+	if err := json.NewDecoder(recorder.Body).Decode(&got); err != nil {
+		t.Fatalf("decoding response: %v", err)
+	}
+	return got
+}
+
 type roundTripFunc func(*http.Request) (*http.Response, error)
 
 func (f roundTripFunc) RoundTrip(request *http.Request) (*http.Response, error) {
diff --git a/demos/egress/stream.go b/demos/egress/stream.go
new file mode 100644
index 000000000..db23be808
--- /dev/null
+++ b/demos/egress/stream.go
@@ -0,0 +1,317 @@
+// 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 main
+
+import (
+	"bufio"
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"net/http"
+	"net/url"
+	"os"
+	"strings"
+	"sync"
+	"sync/atomic"
+	"time"
+
+	"github.com/gorilla/websocket"
+)
+
+// The streaming probe is deliberately split into a start and a poll, rather
+// than one call that blocks until the stream ends. A caller reaches this Actor
+// through the ingress router, whose routes carry a timeout of their own, so a
+// request that waited out a half-minute stream would be cut by the ingress
+// before it could report on the egress. Starting the read in the background
+// keeps every call to the Actor short and moves the waiting to the caller.
+
+// streamProbeRequest asks for one long-lived stream to be read in the
+// background. It is the body of POST /stream.
+type streamProbeRequest struct {
+	// URL is the origin to stream from: http(s) for SSE, ws(s) for WebSocket.
+	URL string `json:"url"`
+	// Protocol is "sse" or "websocket".
+	Protocol string `json:"protocol"`
+	// Hold is how long to keep reading before declaring success.
+	Hold string `json:"hold,omitempty"`
+}
+
+// streamProbeStatus is the state of one probe. POST /stream returns it with
+// only ID set; GET /stream?id=... returns it filled in.
+type streamProbeStatus struct {
+	ID     string `json:"id,omitempty"`
+	Events int    `json:"events"`
+	First  string `json:"first,omitempty"`
+	Last   string `json:"last,omitempty"`
+	// ElapsedMs is how long the stream stayed open. Against a proxy that cuts
+	// long streams this is the interesting number: it names the timeout.
+	ElapsedMs int64  `json:"elapsedMs"`
+	Done      bool   `json:"done"`
+	Error     string `json:"error,omitempty"`
+}
+
+// streamProbe is one in-flight stream. Its fields are written by the reading
+// goroutine and read by whatever polls GET /stream, so all access is guarded.
+type streamProbe struct {
+	started time.Time
+
+	mu      sync.Mutex
+	events  int
+	first   string
+	last    string
+	elapsed time.Duration
+	done    bool
+	err     string
+}
+
+func (p *streamProbe) record(event string) {
+	p.mu.Lock()
+	defer p.mu.Unlock()
+	if p.events == 0 {
+		p.first = event
+	}
+	p.last = event
+	p.events++
+}
+
+func (p *streamProbe) finish(err error) {
+	p.mu.Lock()
+	defer p.mu.Unlock()
+	p.elapsed = time.Since(p.started)
+	p.done = true
+	if err != nil {
+		p.err = err.Error()
+	}
+}
+
+func (p *streamProbe) status() streamProbeStatus {
+	p.mu.Lock()
+	defer p.mu.Unlock()
+	elapsed := p.elapsed
+	if !p.done {
+		elapsed = time.Since(p.started)
+	}
+	return streamProbeStatus{
+		Events:    p.events,
+		First:     p.first,
+		Last:      p.last,
+		ElapsedMs: elapsed.Milliseconds(),
+		Done:      p.done,
+		Error:     p.err,
+	}
+}
+
+// streamProbeRegistry holds probes between the call that starts one and the
+// calls that poll it.
+type streamProbeRegistry struct {
+	mu     sync.Mutex
+	nextID atomic.Uint64
+	probes map[string]*streamProbe
+}
+
+func newStreamProbeRegistry() *streamProbeRegistry {
+	return &streamProbeRegistry{probes: make(map[string]*streamProbe)}
+}
+
+func (r *streamProbeRegistry) add(probe *streamProbe) string {
+	id := fmt.Sprintf("stream-%d", r.nextID.Add(1))
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	r.probes[id] = probe
+	return id
+}
+
+func (r *streamProbeRegistry) get(id string) (*streamProbe, bool) {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	probe, ok := r.probes[id]
+	return probe, ok
+}
+
+func handleStreamProbe(registry *streamProbeRegistry) http.HandlerFunc {
+	return func(w http.ResponseWriter, r *http.Request) {
+		switch r.Method {
+		case http.MethodPost:
+			startStreamProbe(w, r, registry)
+		case http.MethodGet:
+			id := r.URL.Query().Get("id")
+			probe, ok := registry.get(id)
+			if !ok {
+				writeStreamProbeJSON(w, http.StatusNotFound, streamProbeStatus{Error: fmt.Sprintf("no such probe: %q", id)})
+				return
+			}
+			writeStreamProbeJSON(w, http.StatusOK, probe.status())
+		default:
+			w.Header().Set("Allow", http.MethodGet+", "+http.MethodPost)
+			writeStreamProbeJSON(w, http.StatusMethodNotAllowed, streamProbeStatus{Error: "method must be GET or POST"})
+		}
+	}
+}
+
+func startStreamProbe(w http.ResponseWriter, r *http.Request, registry *streamProbeRegistry) {
+	const defaultHold = 30 * time.Second
+	// A probe holds a connection open for its whole life, so an unbounded hold
+	// would let one call pin an outbound connection indefinitely.
+	const maxHold = 5 * time.Minute
+
+	var input streamProbeRequest
+	decoder := json.NewDecoder(http.MaxBytesReader(w, r.Body, maxRequestBody))
+	if err := decoder.Decode(&input); err != nil {
+		writeStreamProbeJSON(w, http.StatusBadRequest, streamProbeStatus{Error: fmt.Sprintf("invalid JSON payload: %v", err)})
+		return
+	}
+	if err := validateStreamURL(input.Protocol, input.URL); err != nil {
+		writeStreamProbeJSON(w, http.StatusBadRequest, streamProbeStatus{Error: err.Error()})
+		return
+	}
+
+	hold := defaultHold
+	if input.Hold != "" {
+		parsed, err := time.ParseDuration(input.Hold)
+		if err != nil {
+			writeStreamProbeJSON(w, http.StatusBadRequest, streamProbeStatus{Error: fmt.Sprintf("invalid hold: %v", err)})
+			return
+		}
+		hold = min(parsed, maxHold)
+	}
+
+	probe := &streamProbe{started: time.Now()}
+	id := registry.add(probe)
+	go probe.run(input.Protocol, input.URL, hold)
+
+	writeStreamProbeJSON(w, http.StatusAccepted, streamProbeStatus{ID: id})
+}
+
+// run reads the stream until hold elapses. Reaching the hold is the success
+// case, so the error it finishes with is exactly "something ended this stream
+// before we meant to end it".
+func (p *streamProbe) run(protocol, target string, hold time.Duration) {
+	// Not r.Context(): the request that started this probe has already been
+	// answered, and its context is cancelled the moment the handler returns.
+	ctx, cancel := context.WithTimeout(context.Background(), hold)
+	defer cancel()
+
+	var err error
+	switch protocol {
+	case streamProtocolSSE:
+		err = p.readSSE(ctx, target)
+	case streamProtocolWebSocket:
+		err = p.readWebSocket(ctx, target, time.Now().Add(hold))
+	default:
+		err = fmt.Errorf("unsupported protocol %q", protocol)
+	}
+	p.finish(err)
+}
+
+func (p *streamProbe) readSSE(ctx context.Context, target string) error {
+	request, err := http.NewRequestWithContext(ctx, http.MethodGet, target, nil)
+	if err != nil {
+		return fmt.Errorf("building request: %w", err)
+	}
+	request.Header.Set("Accept", "text/event-stream")
+
+	// Not the package-level client: its Timeout is a ceiling on the whole
+	// response body, which for a stream means a ceiling on the stream.
+	response, err := (&http.Client{}).Do(request)
+	if err != nil {
+		return fmt.Errorf("connecting: %w", err)
+	}
+	defer response.Body.Close()
+	if response.StatusCode != http.StatusOK {
+		return fmt.Errorf("origin returned HTTP %d", response.StatusCode)
+	}
+
+	scanner := bufio.NewScanner(response.Body)
+	for scanner.Scan() {
+		if data, ok := strings.CutPrefix(scanner.Text(), "data: "); ok {
+			p.record(data)
+		}
+	}
+	return streamEndReason(ctx, scanner.Err())
+}
+
+func (p *streamProbe) readWebSocket(ctx context.Context, target string, deadline time.Time) error {
+	connection, response, err := websocket.DefaultDialer.DialContext(ctx, target, nil)
+	if err != nil {
+		if response != nil {
+			// A proxy that will not carry the upgrade answers here, and its
+			// status code is the most useful thing we can report.
+			return fmt.Errorf("upgrading: %w (origin returned HTTP %d)", err, response.StatusCode)
+		}
+		return fmt.Errorf("upgrading: %w", err)
+	}
+	defer connection.Close()
+
+	if err := connection.SetReadDeadline(deadline); err != nil {
+		return fmt.Errorf("setting read deadline: %w", err)
+	}
+	for {
+		_, message, err := connection.ReadMessage()
+		if err != nil {
+			return streamEndReason(ctx, err)
+		}
+		p.record(string(message))
+	}
+}
+
+// streamEndReason decides whether the end of a stream was ours or someone
+// else's. Our own deadline firing is the success the probe was asked for; every
+// other ending, including a clean close by the origin, means the stream did not
+// last as long as it was supposed to.
+func streamEndReason(ctx context.Context, err error) error {
+	if ctx.Err() != nil || errors.Is(err, os.ErrDeadlineExceeded) {
+		return nil
+	}
+	if err != nil {
+		return fmt.Errorf("stream ended early: %w", err)
+	}
+	return errors.New("stream ended early: origin closed it")
+}
+
+const (
+	streamProtocolSSE       = "sse"
+	streamProtocolWebSocket = "websocket"
+)
+
+func validateStreamURL(protocol, raw string) error {
+	var schemes []string
+	switch protocol {
+	case streamProtocolSSE:
+		schemes = []string{"http", "https"}
+	case streamProtocolWebSocket:
+		schemes = []string{"ws", "wss"}
+	default:
+		return fmt.Errorf("protocol must be %q or %q", streamProtocolSSE, streamProtocolWebSocket)
+	}
+
+	parsed, err := url.Parse(raw)
+	if err != nil {
+		return fmt.Errorf("invalid URL: %w", err)
+	}
+	if parsed.Scheme != schemes[0] && parsed.Scheme != schemes[1] {
+		return fmt.Errorf("URL scheme must be %s or %s for protocol %q", schemes[0], schemes[1], protocol)
+	}
+	if parsed.Hostname() == "" {
+		return fmt.Errorf("URL must include a hostname")
+	}
+	return nil
+}
+
+func writeStreamProbeJSON(w http.ResponseWriter, status int, response streamProbeStatus) {
+	w.Header().Set("Content-Type", "application/json")
+	w.WriteHeader(status)
+	_ = json.NewEncoder(w).Encode(response)
+}
diff --git a/demos/egress/stream_test.go b/demos/egress/stream_test.go
new file mode 100644
index 000000000..d797262b3
--- /dev/null
+++ b/demos/egress/stream_test.go
@@ -0,0 +1,250 @@
+// 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 main
+
+import (
+	"encoding/json"
+	"fmt"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/gorilla/websocket"
+)
+
+// The probe's whole purpose is to distinguish "the stream lasted as long as we
+// asked" from "something ended it sooner", so every test here is one of those
+// two outcomes. They run on a scale of hundreds of milliseconds; what the
+// e2e test measures in tens of seconds is the same distinction.
+const (
+	testHold         = 400 * time.Millisecond
+	testTickInterval = 20 * time.Millisecond
+)
+
+func TestStreamProbeHoldsSSEOpen(t *testing.T) {
+	origin := httptest.NewServer(sseTickHandler(t, 0))
+	defer origin.Close()
+
+	status := runStreamProbe(t, streamProtocolSSE, origin.URL, testHold)
+	if status.Error != "" {
+		t.Fatalf("probe reported an error after %dms: %s", status.ElapsedMs, status.Error)
+	}
+	if elapsed := time.Duration(status.ElapsedMs) * time.Millisecond; elapsed < testHold {
+		t.Errorf("stream stayed open %v, want at least %v", elapsed, testHold)
+	}
+	if status.Events == 0 {
+		t.Error("probe recorded no events, so it never read the stream")
+	}
+	if status.First != "tick-0" {
+		t.Errorf("first event = %q, want %q", status.First, "tick-0")
+	}
+}
+
+// TestStreamProbeReportsSSECutShort is the failure the e2e test is built to
+// catch, reproduced locally: an origin that ends the stream early must come
+// back as an error rather than as a short but successful hold.
+func TestStreamProbeReportsSSECutShort(t *testing.T) {
+	const eventsBeforeClose = 3
+	origin := httptest.NewServer(sseTickHandler(t, eventsBeforeClose))
+	defer origin.Close()
+
+	status := runStreamProbe(t, streamProtocolSSE, origin.URL, testHold)
+	if status.Error == "" {
+		t.Fatalf("probe reported success after %dms and %d events, want an error", status.ElapsedMs, status.Events)
+	}
+	if elapsed := time.Duration(status.ElapsedMs) * time.Millisecond; elapsed >= testHold {
+		t.Errorf("stream stayed open %v, want it to have ended before the %v hold", elapsed, testHold)
+	}
+	if status.Events != eventsBeforeClose {
+		t.Errorf("events = %d, want %d", status.Events, eventsBeforeClose)
+	}
+}
+
+func TestStreamProbeHoldsWebSocketOpen(t *testing.T) {
+	origin := httptest.NewServer(webSocketTickHandler(t, 0))
+	defer origin.Close()
+
+	status := runStreamProbe(t, streamProtocolWebSocket, webSocketURL(origin.URL), testHold)
+	if status.Error != "" {
+		t.Fatalf("probe reported an error after %dms: %s", status.ElapsedMs, status.Error)
+	}
+	if elapsed := time.Duration(status.ElapsedMs) * time.Millisecond; elapsed < testHold {
+		t.Errorf("stream stayed open %v, want at least %v", elapsed, testHold)
+	}
+	if status.Events == 0 {
+		t.Error("probe recorded no events, so it never read the stream")
+	}
+}
+
+func TestStreamProbeReportsWebSocketCutShort(t *testing.T) {
+	const eventsBeforeClose = 3
+	origin := httptest.NewServer(webSocketTickHandler(t, eventsBeforeClose))
+	defer origin.Close()
+
+	status := runStreamProbe(t, streamProtocolWebSocket, webSocketURL(origin.URL), testHold)
+	if status.Error == "" {
+		t.Fatalf("probe reported success after %dms and %d events, want an error", status.ElapsedMs, status.Events)
+	}
+	if status.Events != eventsBeforeClose {
+		t.Errorf("events = %d, want %d", status.Events, eventsBeforeClose)
+	}
+}
+
+// TestStreamProbeRefusesUpgradeFailure covers what a proxy that will not carry a
+// WebSocket upgrade looks like from here: the handshake is answered with a plain
+// HTTP status, and that status is what the probe must report.
+func TestStreamProbeRefusesUpgradeFailure(t *testing.T) {
+	origin := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+		http.Error(w, "no upgrades here", http.StatusBadGateway)
+	}))
+	defer origin.Close()
+
+	status := runStreamProbe(t, streamProtocolWebSocket, webSocketURL(origin.URL), testHold)
+	if status.Error == "" {
+		t.Fatal("probe reported success against an origin that refused the upgrade")
+	}
+	if !strings.Contains(status.Error, "502") {
+		t.Errorf("error = %q, want it to name the 502 the origin answered with", status.Error)
+	}
+}
+
+func TestStreamProbeInvalidRequests(t *testing.T) {
+	tests := []struct {
+		name   string
+		method string
+		path   string
+		body   string
+		status int
+	}{
+		{name: "method", method: http.MethodDelete, path: "/stream", body: `{}`, status: http.StatusMethodNotAllowed},
+		{name: "malformed JSON", method: http.MethodPost, path: "/stream", body: `{`, status: http.StatusBadRequest},
+		{name: "unknown protocol", method: http.MethodPost, path: "/stream", body: `{"url":"http://o/","protocol":"grpc"}`, status: http.StatusBadRequest},
+		// The scheme has to agree with the protocol: a ws:// URL handed to the
+		// HTTP client, or an http:// URL to the dialer, fails deep in a library
+		// rather than as a bad request.
+		{name: "sse with a ws URL", method: http.MethodPost, path: "/stream", body: `{"url":"ws://o/","protocol":"sse"}`, status: http.StatusBadRequest},
+		{name: "websocket with an http URL", method: http.MethodPost, path: "/stream", body: `{"url":"http://o/","protocol":"websocket"}`, status: http.StatusBadRequest},
+		{name: "missing hostname", method: http.MethodPost, path: "/stream", body: `{"url":"http:///sse","protocol":"sse"}`, status: http.StatusBadRequest},
+		{name: "invalid hold", method: http.MethodPost, path: "/stream", body: `{"url":"http://o/","protocol":"sse","hold":"soon"}`, status: http.StatusBadRequest},
+		{name: "unknown probe id", method: http.MethodGet, path: "/stream?id=stream-404", body: "", status: http.StatusNotFound},
+	}
+
+	handler := newHandler(http.DefaultClient)
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			recorder := httptest.NewRecorder()
+			handler.ServeHTTP(recorder, httptest.NewRequest(test.method, test.path, strings.NewReader(test.body)))
+			if recorder.Code != test.status {
+				t.Errorf("status = %d, want %d; body = %s", recorder.Code, test.status, recorder.Body.String())
+			}
+		})
+	}
+}
+
+// runStreamProbe drives one probe through the handler the way the e2e test
+// drives it through the ingress: start it, then poll until it is done.
+func runStreamProbe(t *testing.T, protocol, url string, hold time.Duration) streamProbeStatus {
+	t.Helper()
+	handler := newHandler(http.DefaultClient)
+
+	payload, err := json.Marshal(streamProbeRequest{URL: url, Protocol: protocol, Hold: hold.String()})
+	if err != nil {
+		t.Fatalf("marshaling the stream probe request: %v", err)
+	}
+	recorder := httptest.NewRecorder()
+	handler.ServeHTTP(recorder, httptest.NewRequest(http.MethodPost, "/stream", strings.NewReader(string(payload))))
+	if recorder.Code != http.StatusAccepted {
+		t.Fatalf("starting the probe returned %d, want 202; body = %s", recorder.Code, recorder.Body.String())
+	}
+	var started streamProbeStatus
+	if err := json.NewDecoder(recorder.Body).Decode(&started); err != nil {
+		t.Fatalf("decoding the start response: %v", err)
+	}
+	if started.ID == "" {
+		t.Fatal("start response carried no probe id")
+	}
+
+	deadline := time.Now().Add(hold + 10*time.Second)
+	for {
+		recorder := httptest.NewRecorder()
+		handler.ServeHTTP(recorder, httptest.NewRequest(http.MethodGet, "/stream?id="+started.ID, nil))
+		if recorder.Code != http.StatusOK {
+			t.Fatalf("polling probe %s returned %d; body = %s", started.ID, recorder.Code, recorder.Body.String())
+		}
+		var status streamProbeStatus
+		if err := json.NewDecoder(recorder.Body).Decode(&status); err != nil {
+			t.Fatalf("decoding probe %s status: %v", started.ID, err)
+		}
+		if status.Done {
+			return status
+		}
+		if time.Now().After(deadline) {
+			t.Fatalf("probe %s never finished; last seen with %d events over %dms", started.ID, status.Events, status.ElapsedMs)
+		}
+		time.Sleep(testTickInterval)
+	}
+}
+
+// sseTickHandler serves the same event stream as demos/egress/streamserver. A
+// nonzero stopAfter makes it hang up mid-stream, standing in for a proxy that
+// cuts the connection.
+func sseTickHandler(t *testing.T, stopAfter int) http.HandlerFunc {
+	t.Helper()
+	return func(w http.ResponseWriter, r *http.Request) {
+		w.Header().Set("Content-Type", "text/event-stream")
+		w.WriteHeader(http.StatusOK)
+		w.(http.Flusher).Flush()
+		for n := 0; stopAfter == 0 || n < stopAfter; n++ {
+			select {
+			case <-r.Context().Done():
+				return
+			case <-time.After(testTickInterval):
+			}
+			if _, err := fmt.Fprintf(w, "data: tick-%d\n\n", n); err != nil {
+				return
+			}
+			w.(http.Flusher).Flush()
+		}
+	}
+}
+
+func webSocketTickHandler(t *testing.T, stopAfter int) http.HandlerFunc {
+	t.Helper()
+	upgrader := websocket.Upgrader{CheckOrigin: func(*http.Request) bool { return true }}
+	return func(w http.ResponseWriter, r *http.Request) {
+		connection, err := upgrader.Upgrade(w, r, nil)
+		if err != nil {
+			t.Errorf("upgrading the test origin's connection: %v", err)
+			return
+		}
+		defer connection.Close()
+		for n := 0; stopAfter == 0 || n < stopAfter; n++ {
+			select {
+			case <-r.Context().Done():
+				return
+			case <-time.After(testTickInterval):
+			}
+			if err := connection.WriteMessage(websocket.TextMessage, fmt.Appendf(nil, "tick-%d", n)); err != nil {
+				return
+			}
+		}
+	}
+}
+
+func webSocketURL(httpURL string) string {
+	return "ws" + strings.TrimPrefix(httpURL, "http")
+}
diff --git a/demos/egress/streamserver/main.go b/demos/egress/streamserver/main.go
new file mode 100644
index 000000000..c41867ac8
--- /dev/null
+++ b/demos/egress/streamserver/main.go
@@ -0,0 +1,154 @@
+// 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.
+
+// Command streamserver is a long-lived-stream origin for the egress tests. It
+// emits the same tick sequence two ways -- Server-Sent Events on /sse and
+// WebSocket text frames on /ws -- so a test can hold a stream open across a
+// proxy's timeout boundary and find out whether the proxy cut it.
+//
+// It ticks until the peer goes away rather than sending a fixed number of
+// events. What these tests ask is how long a stream survives, and an origin
+// that ended the stream itself would be answering that question for them.
+package main
+
+import (
+	"fmt"
+	"log/slog"
+	"net/http"
+	"os"
+	"time"
+
+	"github.com/gorilla/websocket"
+)
+
+func main() {
+	slog.SetDefault(slog.New(slog.NewJSONHandler(os.Stdout, nil)))
+
+	const defaultListenAddress = ":8080"
+	// One tick a second is frequent enough that a test can tell a live stream
+	// from a stalled one within a couple of ticks, and sparse enough that a
+	// half-minute hold is tens of events rather than thousands.
+	const defaultTickInterval = time.Second
+
+	address := defaultListenAddress
+	if value := os.Getenv("LISTEN_ADDRESS"); value != "" {
+		address = value
+	}
+	interval := defaultTickInterval
+	if value := os.Getenv("TICK_INTERVAL"); value != "" {
+		parsed, err := time.ParseDuration(value)
+		if err != nil {
+			slog.Error("invalid TICK_INTERVAL", "value", value, "error", err)
+			os.Exit(1)
+		}
+		interval = parsed
+	}
+
+	slog.Info("starting stream server", "address", address, "tickInterval", interval)
+	if err := http.ListenAndServe(address, newHandler(interval)); err != nil {
+		slog.Error("stream server stopped", "error", err)
+		os.Exit(1)
+	}
+}
+
+func newHandler(interval time.Duration) http.Handler {
+	mux := http.NewServeMux()
+	mux.HandleFunc("/readyz", func(w http.ResponseWriter, _ *http.Request) {
+		w.WriteHeader(http.StatusOK)
+		_, _ = fmt.Fprintln(w, "ok")
+	})
+	mux.HandleFunc("/sse", handleSSE(interval))
+	mux.HandleFunc("/ws", handleWebSocket(interval))
+	return mux
+}
+
+// tick is the payload of the nth event. Numbering them lets a reader tell a
+// stream that was cut and resumed from one that ran unbroken.
+func tick(n int) string {
+	return fmt.Sprintf("tick-%d", n)
+}
+
+func handleSSE(interval time.Duration) http.HandlerFunc {
+	return func(w http.ResponseWriter, r *http.Request) {
+		flusher, ok := w.(http.Flusher)
+		if !ok {
+			http.Error(w, "streaming unsupported", http.StatusInternalServerError)
+			return
+		}
+
+		header := w.Header()
+		header.Set("Content-Type", "text/event-stream")
+		// Without this an intermediary is free to buffer the whole stream,
+		// which would make a cut one indistinguishable from a slow one.
+		header.Set("Cache-Control", "no-cache")
+		w.WriteHeader(http.StatusOK)
+		flusher.Flush()
+
+		ticker := time.NewTicker(interval)
+		defer ticker.Stop()
+		for n := 0; ; n++ {
+			select {
+			case <-r.Context().Done():
+				return
+			case <-ticker.C:
+			}
+			if _, err := fmt.Fprintf(w, "data: %s\n\n", tick(n)); err != nil {
+				return
+			}
+			flusher.Flush()
+		}
+	}
+}
+
+func handleWebSocket(interval time.Duration) http.HandlerFunc {
+	// Same-origin checking is the default and would reject the actor, which
+	// sends no Origin header of its own. This fixture is reachable only from
+	// inside the test namespace and serves one hard-coded tick sequence.
+	upgrader := websocket.Upgrader{CheckOrigin: func(*http.Request) bool { return true }}
+
+	return func(w http.ResponseWriter, r *http.Request) {
+		connection, err := upgrader.Upgrade(w, r, nil)
+		if err != nil {
+			// Upgrade has already written a response by this point.
+			slog.Error("websocket upgrade failed", "error", err)
+			return
+		}
+		defer connection.Close()
+
+		// gorilla only processes close and ping frames from inside a read call,
+		// so without draining, a peer that has left is noticed only when a write
+		// eventually fails.
+		go func() {
+			for {
+				if _, _, err := connection.ReadMessage(); err != nil {
+					connection.Close()
+					return
+				}
+			}
+		}()
+
+		ticker := time.NewTicker(interval)
+		defer ticker.Stop()
+		for n := 0; ; n++ {
+			select {
+			case <-r.Context().Done():
+				return
+			case <-ticker.C:
+			}
+			if err := connection.WriteMessage(websocket.TextMessage, []byte(tick(n))); err != nil {
+				return
+			}
+		}
+	}
+}
diff --git a/go.mod b/go.mod
index cc6d8d97d..903562961 100644
--- a/go.mod
+++ b/go.mod
@@ -23,6 +23,7 @@ require (
 	github.com/google/go-containerregistry v0.21.7
 	github.com/google/nftables v0.3.0
 	github.com/google/uuid v1.6.0
+	github.com/gorilla/websocket v1.5.4-0.20250319132907-e064f32e3674
 	github.com/hashicorp/go-reap v0.0.0-20260220095743-4e27870b4f51
 	github.com/jackc/pgx/v5 v5.10.0
 	github.com/klauspost/compress v1.18.6
@@ -144,7 +145,6 @@ require (
 	github.com/google/s2a-go v0.1.9 // indirect
 	github.com/googleapis/enterprise-certificate-proxy v0.3.14 // indirect
 	github.com/googleapis/gax-go/v2 v2.21.0 // indirect
-	github.com/gorilla/websocket v1.5.4-0.20250319132907-e064f32e3674 // indirect
 	github.com/grpc-ecosystem/grpc-gateway/v2 v2.29.0 // indirect
 	github.com/inconshreveable/mousetrap v1.1.0 // indirect
 	github.com/jackc/pgpassfile v1.0.0 // indirect
diff --git a/hack/install-demo-egress.sh b/hack/install-demo-egress.sh
index e4e0074a9..12df26f6b 100644
--- a/hack/install-demo-egress.sh
+++ b/hack/install-demo-egress.sh
@@ -40,6 +40,10 @@ demo-egress_deploy() {
   # ("egress"), the same way demo-counter gets "deployment/counter". The old
   # "egress-deployment" name was NotFound on every successful deploy.
   run_kubectl rollout status deployment/egress -n ate-demo-egress --timeout=300s
+  # The TCP origin for the non-HTTP egress tests.
+  run_kubectl rollout status deployment/bannerserver -n ate-demo-egress --timeout=300s
+  # The SSE and WebSocket origin for the streaming-duration egress tests.
+  run_kubectl rollout status deployment/streamserver -n ate-demo-egress --timeout=300s
   run_kubectl wait --for=condition=Ready actortemplate/egress -n ate-demo-egress --timeout=300s
 }
 
diff --git a/internal/e2e/suites/networking/networking_test.go b/internal/e2e/suites/networking/networking_test.go
index 59871dc76..c4e6c6c0d 100644
--- a/internal/e2e/suites/networking/networking_test.go
+++ b/internal/e2e/suites/networking/networking_test.go
@@ -20,6 +20,7 @@ import (
 	"fmt"
 	"io"
 	"net/http"
+	"strconv"
 	"strings"
 	"testing"
 	"time"
@@ -66,13 +67,13 @@ func TestActorDirectAccess(t *testing.T) {
 	})
 }
 
-// TestActorEgress exercises the full egress path. The Actor's outbound TCP
+// TestActorEgressHTTP exercises the full egress path. The Actor's outbound TCP
 // connection is transparently redirected by nftables into atunnel, wrapped in
 // mTLS with the Actor's own actor-identity certificate plus an HTTP CONNECT to
 // atenet-egress, authorized there against that certificate, and only then
 // dialed out. A masqueraded (pre-gateway) egress would also return 200, so this
 // asserts the gateway is deployed and that it did not reject the Actor.
-func TestActorEgress(t *testing.T) {
+func TestActorEgressHTTP(t *testing.T) {
 	ctx := context.Background()
 	actorName, _ := createAndResumeActor(t, ctx, "egress", egressTemplate)
 	router := mustRouterClient(t, ctx)
@@ -111,33 +112,358 @@ func TestActorEgressHTTPS(t *testing.T) {
 	assertEgressGatewayConnect(t, ctx, since, actorName, "443")
 }
 
+// TestActorEgressRawTCP covers egress for a payload that is neither HTTP nor
+// TLS, from both sides of the who-speaks-first divide. HTTP and HTTPS both have
+// the client send the first bytes, so neither notices a path that waits for
+// downstream data before dialing upstream, or that inspects those first bytes
+// to route.
+//
+// The two subtests dial different ports of the same origin, because that is the
+// only way to select the behavior: the server decides whether to greet before
+// it has read anything, so no field in the request could choose for it. Distinct
+// ports also keep the access-log assertions unambiguous while both subtests
+// share one Actor.
+func TestActorEgressRawTCP(t *testing.T) {
+	// The greeting from demos/egress/bannerserver, which must stay in step with
+	// the Banner constant there.
+	const banner = "TESTBANNER/1.0\r\n"
+
+	tests := []struct {
+		name string
+		port int
+		// wantBanner is what the origin volunteers before being spoken to, so
+		// empty means it stayed silent.
+		wantBanner string
+		// timeout bounds each read the probe does.
+		timeout string
+	}{
+		{name: "server speaks first", port: bannerServerPort, wantBanner: banner, timeout: "10s"},
+		{name: "client speaks first", port: bannerServerQuietPort, wantBanner: "", timeout: "2s"},
+	}
+
+	ctx := context.Background()
+	clusterIP := bannerServerClusterIP(t, ctx)
+	actorName, _ := createAndResumeActor(t, ctx, "egress-tcp", egressTemplate)
+	router := mustRouterClient(t, ctx)
+	defer router.Close()
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			// Bound the access-log scan to lines this subtest could have
+			// produced. The slack absorbs clock skew with the gateway's node.
+			since := metav1.NewTime(time.Now().Add(-1 * time.Minute))
+			// Dial by address: the sandbox does not resolve cluster Service names.
+			address := fmt.Sprintf("%s:%d", clusterIP, test.port)
+
+			// Distinct per run, so the echo cannot be satisfied by anything stale.
+			sent := fmt.Sprintf("ping-%d", time.Now().UnixNano())
+			payload, err := json.Marshal(map[string]any{"address": address, "send": sent, "timeout": test.timeout})
+			if err != nil {
+				t.Fatalf("marshaling the TCP probe request for %s: %v", address, err)
+			}
+			status, body := postThroughEgressActor(t, ctx, router, resources.ActorRef{Atespace: networkingAtespace, Name: actorName}, "/tcp", payload)
+			if status != http.StatusOK {
+				t.Fatalf("Actor raw TCP probe of %s returned HTTP %d, want 200; body: %s", address, status, body)
+			}
+
+			var probe struct {
+				Banner   string `json:"banner"`
+				Received string `json:"received"`
+				Error    string `json:"error"`
+			}
+			if err := json.Unmarshal(body, &probe); err != nil {
+				t.Fatalf("decoding the TCP probe response %s: %v", body, err)
+			}
+			// The probe reads before it writes, so this distinguishes an origin
+			// that spoke unprompted through the tunnel from one that did not.
+			if probe.Banner != test.wantBanner {
+				t.Fatalf("banner from %s = %q, want %q (error: %q)", address, probe.Banner, test.wantBanner, probe.Error)
+			}
+			if probe.Received != sent {
+				t.Fatalf("echo from %s = %q, want %q (error: %q)", address, probe.Received, sent, probe.Error)
+			}
+			t.Logf("Actor raw TCP probe of %s succeeded; banner: %q", address, probe.Banner)
+
+			port := strconv.Itoa(test.port)
+			assertEgressGatewayConnect(t, ctx, since, actorName, port)
+			assertEgressGatewayTunneledBytes(t, ctx, since, actorName, port)
+		})
+	}
+}
+
+// TestActorEgressSSH is TestActorEgressRawTCP against a real server-speaks-first
+// protocol.
+func TestActorEgressSSH(t *testing.T) {
+	// RFC 4253 §4.2: the SSH server sends its identification string first.
+	const identificationPrefix = "SSH-2.0-"
+	const address = "github.com:22"
+
+	ctx := context.Background()
+	actorName, _ := createAndResumeActor(t, ctx, "egress-ssh", egressTemplate)
+	router := mustRouterClient(t, ctx)
+	defer router.Close()
+
+	since := metav1.NewTime(time.Now().Add(-1 * time.Minute))
+
+	// No Send: the identification string alone shows the transport carried the
+	// server's first bytes, and anything written after it would start a key
+	// exchange this test has no reason to hold up its end of.
+	payload, err := json.Marshal(map[string]any{"address": address, "timeout": "10s"})
+	if err != nil {
+		t.Fatalf("marshaling the SSH probe request: %v", err)
+	}
+	status, body := postThroughEgressActor(t, ctx, router, resources.ActorRef{Atespace: networkingAtespace, Name: actorName}, "/tcp", payload)
+	if status != http.StatusOK {
+		t.Fatalf("Actor SSH probe of %s returned HTTP %d, want 200; body: %s", address, status, body)
+	}
+
+	var probe struct {
+		Banner string `json:"banner"`
+		Error  string `json:"error"`
+	}
+	if err := json.Unmarshal(body, &probe); err != nil {
+		t.Fatalf("decoding the SSH probe response %s: %v", body, err)
+	}
+	if !strings.HasPrefix(probe.Banner, identificationPrefix) {
+		t.Fatalf("banner from %s = %q, want a %q prefix (error: %q)", address, probe.Banner, identificationPrefix, probe.Error)
+	}
+	t.Logf("Actor SSH probe of %s succeeded; identification string: %q", address, strings.TrimSpace(probe.Banner))
+
+	assertEgressGatewayConnect(t, ctx, since, actorName, "22")
+}
+
+// TestActorEgressStreamingDuration covers egress for connections that stay open
+// long after the request that opened them. Every other egress test finishes in
+// a second or two, so all of them pass against a gateway that silently cuts a
+// connection once some timer expires -- and several timers on this path do
+// exactly that unless they are explicitly disabled: Envoy applies a route
+// timeout to a CONNECT tunnel's whole lifetime rather than just its headers,
+// and the MITM gateway's HTTP chains apply one to a response body that, for a
+// stream, never ends.
+//
+// The two subtests are the two ways a long-lived stream is normally built. SSE
+// is one HTTP response that never finishes; WebSocket is an upgrade that leaves
+// HTTP behind entirely, which a proxy carries only if it has been configured to
+// proxy that upgrade.
+//
+// The Actor reads the stream in the background and is polled for progress,
+// because the ingress route in front of it has a timeout of its own: a request
+// that waited out the whole hold would be cut by the ingress before it could
+// report on the egress.
+func TestActorEgressStreamingDuration(t *testing.T) {
+	// The hold has to clear every timeout on the egress path that a stream can
+	// trip -- Envoy's 15s default route timeout on the CONNECT tunnel and the
+	// 30s route timeout on the MITM leg's HTTP chains -- or the test passes
+	// against a gateway that would still cut a real stream.
+	const hold = 35 * time.Second
+	// The origin ticks once a second, so a stream that survives the hold
+	// delivers roughly one event per second of it. The margin absorbs connection
+	// setup and the first tick's interval; the assertion that matters is that
+	// events kept arriving for the whole hold, not their exact count.
+	const minEvents = 25
+
+	tests := []struct {
+		name     string
+		protocol string
+		// url builds the origin address from the fixture's cluster IP. The
+		// sandbox does not resolve cluster Service names, so it must be dialed
+		// by address.
+		url func(clusterIP string) string
+	}{
+		{
+			name:     "server-sent events",
+			protocol: "sse",
+			url:      func(ip string) string { return fmt.Sprintf("http://%s:%d/sse", ip, streamServerPort) },
+		},
+		{
+			name:     "websocket",
+			protocol: "websocket",
+			url:      func(ip string) string { return fmt.Sprintf("ws://%s:%d/ws", ip, streamServerPort) },
+		},
+	}
+
+	ctx := context.Background()
+	clusterIP := egressFixtureClusterIP(t, ctx, "streamserver")
+	actorName, _ := createAndResumeActor(t, ctx, "egress-stream", egressTemplate)
+	router := mustRouterClient(t, ctx)
+	defer router.Close()
+
+	actorRef := resources.ActorRef{Atespace: networkingAtespace, Name: actorName}
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			// Bound the access-log scan to lines this subtest could have
+			// produced. The slack absorbs clock skew with the gateway's node.
+			since := metav1.NewTime(time.Now().Add(-1 * time.Minute))
+			url := test.url(clusterIP)
+
+			payload, err := json.Marshal(map[string]any{"url": url, "protocol": test.protocol, "hold": hold.String()})
+			if err != nil {
+				t.Fatalf("marshaling the stream probe request for %s: %v", url, err)
+			}
+			// The Actor answers 202: it has started reading, not finished. A
+			// retry on anything else must not treat 202 as the failure, or every
+			// pass would start another stream and leave it running.
+			status, body := postThroughEgressActorExpecting(t, ctx, router, actorRef, "/stream", payload, http.StatusAccepted)
+			if status != http.StatusAccepted {
+				t.Fatalf("starting the stream probe of %s returned HTTP %d, want 202; body: %s", url, status, body)
+			}
+			var started struct {
+				ID string `json:"id"`
+			}
+			if err := json.Unmarshal(body, &started); err != nil {
+				t.Fatalf("decoding the stream probe start response %s: %v", body, err)
+			}
+
+			probe := awaitStreamProbe(t, ctx, router, actorRef, started.ID, hold)
+			if probe.Error != "" {
+				t.Fatalf("stream of %s ended after %v with %d events: %s (want it held for %v)",
+					url, time.Duration(probe.ElapsedMs)*time.Millisecond, probe.Events, probe.Error, hold)
+			}
+			// A stream can be cut without anyone reporting an error -- a proxy
+			// that closes a CONNECT tunnel on its route timeout looks to the
+			// reader like an origin that hung up -- so check the clock too.
+			if elapsed := time.Duration(probe.ElapsedMs) * time.Millisecond; elapsed < hold {
+				t.Fatalf("stream of %s stayed open %v, want at least %v (%d events, last %q)",
+					url, elapsed, hold, probe.Events, probe.Last)
+			}
+			if probe.Events < minEvents {
+				t.Fatalf("stream of %s delivered %d events over %v, want at least %d; it stalled rather than closed (first %q, last %q)",
+					url, probe.Events, hold, minEvents, probe.First, probe.Last)
+			}
+			t.Logf("Actor held the %s stream of %s open for %dms across %d events (first %q, last %q)",
+				test.protocol, url, probe.ElapsedMs, probe.Events, probe.First, probe.Last)
+
+			port := strconv.Itoa(streamServerPort)
+			assertEgressGatewayConnect(t, ctx, since, actorName, port)
+		})
+	}
+}
+
+// streamProbeStatus mirrors the response of the egress demo Actor's /stream
+// endpoint, from demos/egress/stream.go.
+type streamProbeStatus struct {
+	Events    int    `json:"events"`
+	First     string `json:"first"`
+	Last      string `json:"last"`
+	ElapsedMs int64  `json:"elapsedMs"`
+	Done      bool   `json:"done"`
+	Error     string `json:"error"`
+}
+
+// awaitStreamProbe polls the Actor until the probe it started reports itself
+// done, and returns its final state. A probe that never finishes is a failure
+// in its own right, so the wait is bounded well above the hold it was given.
+func awaitStreamProbe(t *testing.T, ctx context.Context, router *e2e.RouterClient, actorRef resources.ActorRef, id string, hold time.Duration) streamProbeStatus {
+	t.Helper()
+
+	// Slack over the hold for the reader's own setup and for the poll interval.
+	deadline := time.Now().Add(hold + 30*time.Second)
+	const pollInterval = 2 * time.Second
+	for {
+		response, err := router.Get(ctx, actorRef, "/stream?id="+id)
+		if err != nil {
+			t.Fatalf("polling stream probe %s: %v", id, err)
+		}
+		body, err := io.ReadAll(response.Body)
+		response.Body.Close()
+		if err != nil {
+			t.Fatalf("reading stream probe %s status (HTTP %d): %v", id, response.StatusCode, err)
+		}
+		if response.StatusCode != http.StatusOK {
+			t.Fatalf("polling stream probe %s returned HTTP %d; body: %s", id, response.StatusCode, body)
+		}
+
+		var status streamProbeStatus
+		if err := json.Unmarshal(body, &status); err != nil {
+			t.Fatalf("decoding stream probe %s status %s: %v", id, body, err)
+		}
+		if status.Done {
+			return status
+		}
+		if time.Now().After(deadline) {
+			t.Fatalf("stream probe %s never finished; last seen with %d events over %dms", id, status.Events, status.ElapsedMs)
+		}
+		time.Sleep(pollInterval)
+	}
+}
+
+// The ports the egress fixture Services publish, from
+// demos/egress/egress.yaml.tmpl. The banner origin greets on the first and
+// stays silent until spoken to on the second.
+const (
+	bannerServerPort      = 2222
+	bannerServerQuietPort = 2223
+	streamServerPort      = 8080
+)
+
+// bannerServerClusterIP returns the address of the in-cluster TCP origin the
+// raw-TCP test dials.
+func bannerServerClusterIP(t *testing.T, ctx context.Context) string {
+	t.Helper()
+	return egressFixtureClusterIP(t, ctx, "bannerserver")
+}
+
+// egressFixtureClusterIP returns the address of one of the egress demo's origin
+// Services, which the Actor must dial by address because the sandbox does not
+// resolve cluster Service names.
+func egressFixtureClusterIP(t *testing.T, ctx context.Context, name string) string {
+	t.Helper()
+	service, err := e2e.GetClients().K8s.CoreV1().Services(egressTemplate.namespace).Get(ctx, name, metav1.GetOptions{})
+	if err != nil {
+		t.Fatalf("getting Service %s/%s: %v (deploy the fixture with %s)", egressTemplate.namespace, name, err, egressTemplate.deployFlag)
+	}
+	if service.Spec.ClusterIP == "" || service.Spec.ClusterIP == corev1.ClusterIPNone {
+		t.Fatalf("Service %s/%s has no cluster IP to dial: %q", egressTemplate.namespace, name, service.Spec.ClusterIP)
+	}
+	return service.Spec.ClusterIP
+}
+
 // fetchThroughEgressActor asks the egress demo Actor to fetch url and returns
-// the status and body it echoes back. Retries a non-200 response for up to
-// 30s: ResumeActor can return before its route reaches atenet-router's xDS
-// snapshot, and a request sent in that window sees a transient 503.
+// the status and body it echoes back.
 func fetchThroughEgressActor(t *testing.T, ctx context.Context, router *e2e.RouterClient, actorRef resources.ActorRef, url string) (int, []byte) {
 	t.Helper()
 	payload, err := json.Marshal(map[string]string{"url": url})
 	if err != nil {
 		t.Fatalf("marshaling the fetch request for %s: %v", url, err)
 	}
+	return postThroughEgressActor(t, ctx, router, actorRef, "/", payload)
+}
+
+// postThroughEgressActor POSTs payload to path on the egress demo Actor and
+// returns the status and body it answers with, retrying anything other than a
+// 200 as postThroughEgressActorExpecting describes.
+func postThroughEgressActor(t *testing.T, ctx context.Context, router *e2e.RouterClient, actorRef resources.ActorRef, path string, payload []byte) (int, []byte) {
+	t.Helper()
+	return postThroughEgressActorExpecting(t, ctx, router, actorRef, path, payload, http.StatusOK)
+}
+
+// postThroughEgressActorExpecting POSTs payload to path on the egress demo
+// Actor and returns the status and body it answers with. Retries any status
+// other than want for up to 30s: ResumeActor can return before its route
+// reaches atenet-router's xDS snapshot, and a request sent in that window sees
+// a transient 503.
+//
+// want is a parameter rather than a fixed 200 because a POST that only starts
+// work answers 202, and retrying that would start the work again on every pass.
+func postThroughEgressActorExpecting(t *testing.T, ctx context.Context, router *e2e.RouterClient, actorRef resources.ActorRef, path string, payload []byte, want int) (int, []byte) {
+	t.Helper()
 
 	const timeout = 30 * time.Second
 	deadline := time.Now().Add(timeout)
 	for {
-		response, err := router.PostJSON(ctx, actorRef, "/", payload)
+		response, err := router.PostJSON(ctx, actorRef, path, payload)
 		if err != nil {
-			t.Fatalf("POST %s to egress Actor through ingress: %v", url, err)
+			t.Fatalf("POST %s to egress Actor through ingress: %v", path, err)
 		}
 		body, err := io.ReadAll(response.Body)
 		response.Body.Close()
 		if err != nil {
 			t.Fatalf("reading egress response body (HTTP %d): %v", response.StatusCode, err)
 		}
-		if response.StatusCode == http.StatusOK || time.Now().After(deadline) {
+		if response.StatusCode == want || time.Now().After(deadline) {
 			return response.StatusCode, body
 		}
-		t.Logf("fetch through egress Actor returned HTTP %d; retrying...", response.StatusCode)
+		t.Logf("POST %s to egress Actor returned HTTP %d, want %d; retrying...", path, response.StatusCode, want)
 		time.Sleep(1 * time.Second)
 	}
 }
@@ -163,6 +489,55 @@ func assertEgressGatewayConnect(t *testing.T, ctx context.Context, since metav1.
 	})
 }
 
+// assertEgressGatewayTunneledBytes waits for the access-log record the gateway
+// writes when the tunnel closes, and requires bytes to have crossed it in both
+// directions. The CONNECT record alone only says the tunnel was authorized and
+// opened; these counters are the gateway's own evidence that it relayed the
+// payload, rather than the Actor having reached the origin some other way.
+func assertEgressGatewayTunneledBytes(t *testing.T, ctx context.Context, since metav1.Time, actorName, port string) {
+	t.Helper()
+	want := fmt.Sprintf("a closed tunnel to port %s by actor %s carrying bytes both ways", port, actorName)
+	waitForAccessLog(t, ctx, since, want, func(lines []string) (bool, error) {
+		for _, line := range lines {
+			authority, ok := accessLogField(line, "authority")
+			if !ok || !strings.HasSuffix(authority, ":"+port) {
+				continue
+			}
+			if !strings.Contains(line, "/actor/"+actorName) {
+				continue
+			}
+			// The counters only carry their final values on the record flushed
+			// at close; the one flushed on establishment reports zeroes.
+			up, upOK := accessLogCount(line, "up_bytes")
+			down, downOK := accessLogCount(line, "down_bytes")
+			if !upOK || !downOK {
+				return false, fmt.Errorf("access-log line has no byte counters, so the log format changed: %s", line)
+			}
+			if up == 0 || down == 0 {
+				continue
+			}
+			t.Logf("egress gateway relayed %d bytes up and %d down: %s", up, down, line)
+			return true, nil
+		}
+		return false, nil
+	})
+}
+
+// accessLogCount parses the field named key as a count. A missing field and an
+// unparseable one are both reported as absent, since either means the caller's
+// expectation of the log format no longer holds.
+func accessLogCount(line, key string) (int, bool) {
+	raw, ok := accessLogField(line, key)
+	if !ok {
+		return 0, false
+	}
+	value, err := strconv.Atoi(raw)
+	if err != nil {
+		return 0, false
+	}
+	return value, true
+}
+
 // 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)) {
diff --git a/internal/e2e/suites/sdsmint/sdsmint_test.go b/internal/e2e/suites/sdsmint/sdsmint_test.go
index 92adc69a6..f5308328a 100644
--- a/internal/e2e/suites/sdsmint/sdsmint_test.go
+++ b/internal/e2e/suites/sdsmint/sdsmint_test.go
@@ -123,7 +123,7 @@ func skipUntilPresubmit(t *testing.T) {
 // from Envoy's live secret set and the test would pass without sdsmint having
 // minted anything.
 func TestSdsmintMintsALeafPerSNI(t *testing.T) {
-	skipUntilPresubmit(t)
+	// skipUntilPresubmit(t)
 
 	ctx := context.Background()
 
@@ -228,7 +228,7 @@ func TestSdsmintMintsALeafPerSNI(t *testing.T) {
 // failed to dial, or that got as far as the CONNECT, would be reporting
 // something other than the front door turning it away.
 func TestGatewayRefusesANonActorWorkload(t *testing.T) {
-	skipUntilPresubmit(t)
+	// skipUntilPresubmit(t)
 
 	ctx := context.Background()
 
@@ -254,7 +254,7 @@ func TestGatewayRefusesANonActorWorkload(t *testing.T) {
 // on the certificate alone from one that authorizes on control-plane state, and
 // the difference is whether a deleted actor's credential still works.
 func TestGatewayRefusesAnUnknownActor(t *testing.T) {
-	skipUntilPresubmit(t)
+	// skipUntilPresubmit(t)
 
 	ctx := context.Background()
 
diff --git a/manifests/ate-install/atenet-egress-with-sdsmint.yaml b/manifests/ate-install/atenet-egress-with-sdsmint.yaml
index 72b3e1979..00eaa8d9c 100644
--- a/manifests/ate-install/atenet-egress-with-sdsmint.yaml
+++ b/manifests/ate-install/atenet-egress-with-sdsmint.yaml
@@ -122,6 +122,19 @@ data:
                       # egress_forward_proxy here would be a raw TCP passthrough
                       # with no per-destination visibility at all.
                       cluster: mitm_internal
+                      # A CONNECT tunnel is a session, not a request, so the
+                      # route's default 15s stream timeout would be a ceiling on
+                      # how long an actor may hold any TCP connection open --
+                      # fine for a request/response fetch, wrong for SSH or any
+                      # other protocol that idles.
+                      #
+                      # 0 removes the ceiling on the tunnel's lifetime. What
+                      # still bounds it is inactivity: this HCM leaves
+                      # stream_idle_timeout at its 5 minute default, and every
+                      # tunnelled byte in either direction resets that timer. So
+                      # a busy tunnel lives as long as its peers do, and one with
+                      # nothing on it for 5 minutes is reset.
+                      timeout: 0s
                       upgrade_configs:
                       - upgrade_type: CONNECT
                         connect_config: {}
@@ -158,6 +171,33 @@ data:
                     response_body_mode: NONE
                     request_trailer_mode: SKIP
                     response_trailer_mode: SKIP
+              # Carry the CONNECT authority across the internal-listener hop.
+              #
+              # atunnel takes the authority from SO_ORIGINAL_DST, so it is always
+              # an IP:port, and it is the only place the real destination appears
+              # for a tunnel that carries neither SNI nor a Host header. The MITM
+              # leg's two HTTP chains recover the destination from inside the
+              # tunnel and so never needed this; the raw TCP chain has nothing to
+              # read and does.
+              #
+              # This is the key an ORIGINAL_DST cluster consults first, ahead of
+              # metadata and headers, so egress_tcp_passthrough dials it directly.
+              # It survives the hop only because mitm_internal wraps its transport
+              # socket in internal_upstream, which is what makes
+              # shared_with_upstream mean anything.
+              #
+              # Placed after ext_proc so a denied CONNECT never reaches it: the
+              # address recorded here is the authority ext_proc just authorized,
+              # not one the actor gets to restate.
+              - name: envoy.filters.http.set_filter_state
+                typed_config:
+                  "@type": type.googleapis.com/envoy.extensions.filters.http.set_filter_state.v3.Config
+                  on_request_headers:
+                  - object_key: envoy.network.transport_socket.original_dst_address
+                    format_string:
+                      text_format_source:
+                        inline_string: "%REQ(:AUTHORITY)%"
+                    shared_with_upstream: ONCE
               - name: envoy.filters.http.router
                 typed_config:
                   "@type": type.googleapis.com/envoy.extensions.filters.http.router.v3.Router
@@ -173,18 +213,42 @@ data:
       # tunnelled TLS with a leaf sdsmint mints for the SNI, reads the real
       # Host, and re-originates.
       #
-      # Two chains, selected by what the tunnel actually carries. tls_inspector
-      # tags a ClientHello "tls" and anything else "raw_buffer", so a cleartext
-      # HTTP tunnel gets an HTTP chain instead of being fed to a TLS transport
-      # socket, which would fail the handshake and close the tunnel.
+      # Three chains, selected by what the tunnel actually carries.
+      # tls_inspector tags a ClientHello "tls" and anything else "raw_buffer",
+      # so a cleartext tunnel gets an HTTP chain instead of being fed to a TLS
+      # transport socket, which would fail the handshake and close the tunnel.
+      # http_inspector then splits "raw_buffer" again, because "not TLS" is not
+      # the same claim as "HTTP": without it, SSH and every other non-HTTP
+      # protocol lands on an HTTP connection manager that parses the first bytes
+      # of the stream as a request line, finds no request, and drops the
+      # connection with nothing in the access log to say why.
       # ---------------------------------------------------------------------
       - name: mitm_listener
         stat_prefix: mitm
         internal_listener: {}
+        # Both inspectors work by peeking at bytes the client sends first, so on
+        # a server-speaks-first protocol -- SSH, SMTP, MySQL -- they wait for a
+        # client that is itself waiting for the origin, and the tunnel deadlocks
+        # until the actor's own timeout fires. Giving up quickly and continuing
+        # is the only way out: continue_on_listener_filters_timeout hands the
+        # socket to the chains with no transport protocol detected, and Envoy
+        # defaults that to raw_buffer, which is exactly the passthrough chain
+        # such a connection belongs on. The cost is paid only by connections
+        # that send nothing; a client that speaks is classified immediately. One
+        # second is far longer than a local actor needs to emit a ClientHello,
+        # and short enough to stay well inside the probe timeouts.
+        listener_filters_timeout: 1s
+        continue_on_listener_filters_timeout: true
         listener_filters:
         - name: envoy.filters.listener.tls_inspector
           typed_config:
             "@type": type.googleapis.com/envoy.extensions.filters.listener.tls_inspector.v3.TlsInspector
+        # Must come second: it declines to run at all once a transport protocol
+        # other than raw_buffer has been detected, which is how it keeps its
+        # hands off the TLS chain.
+        - name: envoy.filters.listener.http_inspector
+          typed_config:
+            "@type": type.googleapis.com/envoy.extensions.filters.listener.http_inspector.v3.HttpInspector
         filter_chains:
         - filter_chain_match:
             transport_protocol: tls
@@ -193,6 +257,13 @@ data:
             typed_config:
               "@type": type.googleapis.com/envoy.extensions.filters.network.http_connection_manager.v3.HttpConnectionManager
               stat_prefix: mitm_http
+              # Absent this, Envoy answers an Upgrade request itself instead of
+              # proxying it, so every WebSocket through this leg fails at the
+              # handshake. The raw CONNECT gateway never sees the upgrade at all
+              # and so never needed it; parsing the traffic is what creates the
+              # obligation to understand it.
+              upgrade_configs:
+              - upgrade_type: websocket
               # This is the leg that knows where the traffic actually went.
               # REQUESTED_SERVER_NAME is the SNI the leaf was minted for and it
               # should always equal the authority -- dynamic_forward_proxy
@@ -229,7 +300,17 @@ data:
                   - match: { prefix: "/" }
                     route:
                       cluster: egress_forward_proxy
-                      timeout: 30s
+                      # The route timeout runs until the response is completely
+                      # processed, which for a stream is never: at 30s it cut
+                      # every Server-Sent Events response at half a minute.
+                      #
+                      # 0 removes the ceiling on the response's duration. What
+                      # still bounds it is inactivity: this HCM leaves
+                      # stream_idle_timeout at its 5 minute default, and every
+                      # chunk of the response body resets that timer. So a
+                      # ticking stream runs indefinitely and a stalled one is
+                      # reset after 5 minutes.
+                      timeout: 0s
               http_filters:
               - name: envoy.filters.http.dynamic_forward_proxy
                 typed_config:
@@ -277,13 +358,23 @@ data:
         # The cleartext chain. Nothing to terminate and nothing to mint: the
         # Host header is already in the clear, so this leg reads the
         # destination directly rather than from a certificate it issued.
+        #
+        # application_protocols is what confines it to traffic http_inspector
+        # actually recognised as HTTP. It used to match all of raw_buffer, which
+        # made it the destination for every non-TLS protocol rather than just
+        # this one.
         - filter_chain_match:
             transport_protocol: raw_buffer
+            application_protocols: ["http/1.0", "http/1.1", "h2c"]
           filters:
           - name: envoy.filters.network.http_connection_manager
             typed_config:
               "@type": type.googleapis.com/envoy.extensions.filters.network.http_connection_manager.v3.HttpConnectionManager
               stat_prefix: mitm_cleartext
+              # Same reasoning as the TLS chain: this leg parses HTTP, so it has
+              # to be told which upgrades to carry rather than answer.
+              upgrade_configs:
+              - upgrade_type: websocket
               access_log:
               - name: envoy.access_loggers.file
                 typed_config:
@@ -314,7 +405,11 @@ data:
                   - match: { prefix: "/" }
                     route:
                       cluster: egress_forward_proxy_cleartext
-                      timeout: 30s
+                      # Same reasoning as the TLS chain: a streaming response
+                      # never completes, so any route timeout is a cap on how
+                      # long a stream may run, and stream_idle_timeout's 5
+                      # minute default is what bounds a stalled one instead.
+                      timeout: 0s
               http_filters:
               # Same reasoning as the TLS chain: resolve from the request's own
               # Host, so the name that was policed is the name that is dialled.
@@ -328,11 +423,70 @@ data:
                 typed_config:
                   "@type": type.googleapis.com/envoy.extensions.filters.http.router.v3.Router
 
+        # The passthrough chain: everything the two inspectors could not claim,
+        # plus everything from a client that sent nothing at all. No filter
+        # chain match beyond raw_buffer, so it is the fallback the more specific
+        # chains above fall out of.
+        #
+        # This leg is deliberately blind. It does not decrypt, does not mint,
+        # and cannot name what it is carrying -- an opaque byte stream has no
+        # Host and no SNI to police. The authorization that stands behind it is
+        # the one ext_proc already applied to the CONNECT on the way in, against
+        # the same IP:port this chain now dials. Nothing is weakened by adding
+        # it: without it these connections were not blocked, they were accepted
+        # and then silently mangled.
+        - filter_chain_match:
+            transport_protocol: raw_buffer
+          filters:
+          - name: envoy.filters.network.tcp_proxy
+            typed_config:
+              "@type": type.googleapis.com/envoy.extensions.filters.network.tcp_proxy.v3.TcpProxy
+              stat_prefix: mitm_passthrough
+              cluster: egress_tcp_passthrough
+              access_log:
+              - name: envoy.access_loggers.file
+                typed_config:
+                  "@type": type.googleapis.com/envoy.extensions.access_loggers.file.v3.FileAccessLog
+                  path: /dev/stdout
+                  log_format:
+                    json_format:
+                      leg: passthrough
+                      time: "%START_TIME%"
+                      # Always empty on a healthy connection: reaching this
+                      # chain means tls_inspector found no ClientHello to read a
+                      # name out of. A value here means a TLS connection was
+                      # classified as raw_buffer and is being forwarded without
+                      # being minted for, which is a routing bug, not traffic.
+                      sni: "%REQUESTED_SERVER_NAME%"
+                      flags: "%RESPONSE_FLAGS%"
+                      duration_ms: "%DURATION%"
+                      bytes_sent: "%BYTES_SENT%"
+                      bytes_rcvd: "%BYTES_RECEIVED%"
+                      upstream: "%UPSTREAM_HOST%"
+                      upstream_failure: "%UPSTREAM_TRANSPORT_FAILURE_REASON%"
+                      termination: "%CONNECTION_TERMINATION_DETAILS%"
+
       clusters:
       # The only way into mitm_listener. An internal address, not a socket, so
       # the MITM leg has no listening port anywhere in the pod's netns.
       - name: mitm_internal
         connect_timeout: 1s
+        # An internal address carries bytes and nothing else by default, so the
+        # CONNECT authority the set_filter_state filter recorded on Listener A
+        # would stop here. internal_upstream is the seam that lets structured
+        # state cross with the stream: filter state marked shared_with_upstream
+        # is merged into the internal listener's own downstream connection,
+        # where the ORIGINAL_DST cluster can find it. The wrapped socket is
+        # raw_buffer because there is no real network hop to secure -- this is a
+        # userspace loopback inside one process.
+        transport_socket:
+          name: envoy.transport_sockets.internal_upstream
+          typed_config:
+            "@type": type.googleapis.com/envoy.extensions.transport_sockets.internal_upstream.v3.InternalUpstreamTransport
+            transport_socket:
+              name: envoy.transport_sockets.raw_buffer
+              typed_config:
+                "@type": type.googleapis.com/envoy.extensions.transport_sockets.raw_buffer.v3.RawBuffer
         load_assignment:
           cluster_name: mitm_internal
           endpoints:
@@ -468,6 +622,24 @@ data:
             "@type": type.googleapis.com/envoy.extensions.upstreams.http.v3.HttpProtocolOptions
             explicit_http_config:
               http_protocol_options: {}
+
+      # The passthrough chain's upstream. Not a dynamic forward proxy like the
+      # other two: there is no name to resolve. The destination arrives as the
+      # literal IP:port the actor's kernel recorded in SO_ORIGINAL_DST, relayed
+      # here as the envoy.network.transport_socket.original_dst_address filter
+      # state object, which ORIGINAL_DST consults ahead of metadata, headers,
+      # and the socket's own restored local address -- the last of which is what
+      # this cluster would otherwise use, and which on an internal listener
+      # points back into this process.
+      #
+      # No transport socket, so no upstream TLS: the actor is speaking some
+      # protocol this gateway does not parse, and wrapping that stream in a TLS
+      # session the actor did not ask for and cannot see the peer of would break
+      # every protocol it is meant to carry.
+      - name: egress_tcp_passthrough
+        type: ORIGINAL_DST
+        lb_policy: CLUSTER_PROVIDED
+        connect_timeout: 5s
 ---
 apiVersion: apps/v1
 kind: Deployment
diff --git a/manifests/ate-install/atenet-egress.yaml b/manifests/ate-install/atenet-egress.yaml
index 591ef31fc..113994153 100644
--- a/manifests/ate-install/atenet-egress.yaml
+++ b/manifests/ate-install/atenet-egress.yaml
@@ -104,6 +104,20 @@ data:
                       upgrade_configs:
                       - upgrade_type: CONNECT
                         connect_config: {}
+                      # A CONNECT tunnel is a session, not a request. Unlike a
+                      # WebSocket upgrade, Envoy never disables the route timeout
+                      # once the tunnel is established, so the 15s default would
+                      # be a ceiling on how long an actor may hold any outbound
+                      # connection open -- fine for a fetch, wrong for SSH, a
+                      # stream, or anything else that idles.
+                      #
+                      # 0 removes the ceiling on the tunnel's lifetime. What
+                      # still bounds it is inactivity: this HCM leaves
+                      # stream_idle_timeout at its 5 minute default, and every
+                      # tunnelled byte in either direction resets that timer. So
+                      # a busy tunnel lives as long as its peers do, and one with
+                      # nothing on it for 5 minutes is reset.
+                      timeout: 0s
               http_filters:
               # Actor-identity authorization: on every egress CONNECT, ext_proc
               # reads the actor certificate out of x-forwarded-client-cert,