diff --git a/.github/workflows/praxis.yaml b/.github/workflows/praxis.yaml
index 8acb30a4d..d719089ce 100644
--- a/.github/workflows/praxis.yaml
+++ b/.github/workflows/praxis.yaml
@@ -14,6 +14,8 @@ on:
     paths:
       - "praxis/**"
       - ".github/workflows/praxis.yaml"
+      - "core/attribution/wire.go"
+      - "core/attribution/testdata/**"
       - "core/parse/**"
       - "core/internal/parsercommon/**"
       - "core/model/extensions.go"
@@ -37,6 +39,8 @@ on:
     paths:
       - "praxis/**"
       - ".github/workflows/praxis.yaml"
+      - "core/attribution/wire.go"
+      - "core/attribution/testdata/**"
       - "core/parse/**"
       - "core/internal/parsercommon/**"
       - "core/model/extensions.go"
diff --git a/core/attribution/testdata/wire.golden b/core/attribution/testdata/wire.golden
new file mode 100644
index 000000000..1971e13aa
--- /dev/null
+++ b/core/attribution/testdata/wire.golden
@@ -0,0 +1 @@
+eyJhZ2VudCI6ImNsYXVkZS1jb2RlIiwicHJvYyI6InAxIiwiY2xpZW50UHJvYyI6InAyIiwic2Vzc2lvbiI6InMxIiwic291cmNlIjoiaGVhZGVyIiwic2Vzc2lvbkhlYWRlciI6IlgtQ2xhdWRlLUNvZGUtU2Vzc2lvbi1JZCIsImNoYWluU2Vzc2lvbiI6InMwIiwicmVjb3JkU2Vzc2lvbiI6InMxIn0
diff --git a/core/attribution/wire.go b/core/attribution/wire.go
index 9f98512e1..e39bf610b 100644
--- a/core/attribution/wire.go
+++ b/core/attribution/wire.go
@@ -1,6 +1,10 @@
 package attribution
 
-import "net/http"
+import (
+	"encoding/base64"
+	"encoding/json"
+	"net/http"
+)
 
 // The headers an Attribution will travel in: what a hop that resolved a request — a local relay
 // that ran Resolve at accept time — stamps on it for the recorder behind it, which then trusts
@@ -9,7 +13,8 @@ import "net/http"
 //
 // STRIP, THEN INJECT. A client can send these headers too, and one that did would choose its
 // own session and process. So the forward proxy removes them from what the client sent (Strip)
-// today, before any hop is trusted to set them. The encoder and decoder land with that hop.
+// today, before any hop is trusted to set them. The praxis hop sends one HeaderAttribution
+// instead: Wire.Encode is its encoder, and the gateway (praxis/) has the decoder.
 const (
 	HeaderAgent = "X-Cortex-Agent"
 	HeaderProc  = "X-Cortex-Proc"
@@ -31,3 +36,25 @@ func Strip(h http.Header) {
 		h.Del(k)
 	}
 }
+
+// HeaderAttribution carries a Wire from the praxis hop to cortex-gateway.
+const HeaderAttribution = "X-Cortex-Attribution"
+
+// Wire is the attribution result that the hop sends in X-Cortex-Attribution.
+// The JSON keys must stay equal to the Rust Attribution struct in praxis/filters.
+type Wire struct {
+	Agent         string `json:"agent,omitempty"`
+	Proc          string `json:"proc,omitempty"`
+	ClientProc    string `json:"clientProc,omitempty"`
+	Session       string `json:"session,omitempty"`
+	Source        string `json:"source,omitempty"`
+	SessionHeader string `json:"sessionHeader,omitempty"`
+	ChainSession  string `json:"chainSession,omitempty"`
+	RecordSession string `json:"recordSession,omitempty"`
+}
+
+// Encode gives the header value: the JSON of w in base64url with no padding.
+func (w Wire) Encode() string {
+	b, _ := json.Marshal(w) // a struct of strings always marshals
+	return base64.RawURLEncoding.EncodeToString(b)
+}
diff --git a/core/attribution/wire_test.go b/core/attribution/wire_test.go
index d9c74af42..4f83bbbda 100644
--- a/core/attribution/wire_test.go
+++ b/core/attribution/wire_test.go
@@ -1,7 +1,14 @@
 package attribution
 
 import (
+	"encoding/json"
+	"flag"
 	"net/http"
+	"os"
+	"path/filepath"
+	"slices"
+	"sort"
+	"strings"
 	"testing"
 )
 
@@ -17,3 +24,58 @@ func TestWire_StripRemovesOnlyAttribution(t *testing.T) {
 		t.Errorf("after Strip: %v, want only User-Agent", h)
 	}
 }
+
+var update = flag.Bool("update", false, "rewrite testdata/wire.golden from Wire.Encode")
+
+// golden is the Wire that testdata/wire.golden holds. The Rust test decodes the file and
+// compares it with these values.
+var golden = Wire{Agent: "claude-code", Proc: "p1", ClientProc: "p2", Session: "s1", Source: "header",
+	SessionHeader: "X-Claude-Code-Session-Id", ChainSession: "s0", RecordSession: "s1"}
+
+// The encoding of golden must equal the file that the gateway's decoder test reads.
+func TestWire_EncodeGolden(t *testing.T) {
+	path := filepath.Join("testdata", "wire.golden")
+	got := golden.Encode()
+	if *update {
+		if err := os.WriteFile(path, []byte(got+"\n"), 0o644); err != nil {
+			t.Fatal(err)
+		}
+	}
+	want, err := os.ReadFile(path)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if got != strings.TrimSpace(string(want)) {
+		t.Errorf("Encode() = %q, want %q", got, want)
+	}
+}
+
+// The JSON keys must be the camelCase names of the Rust Attribution struct.
+func TestWire_KeysMatchRust(t *testing.T) {
+	b, err := json.Marshal(golden)
+	if err != nil {
+		t.Fatal(err)
+	}
+	var m map[string]any
+	if err := json.Unmarshal(b, &m); err != nil {
+		t.Fatal(err)
+	}
+	var keys []string
+	for k := range m {
+		keys = append(keys, k)
+	}
+	sort.Strings(keys)
+	want := []string{"agent", "chainSession", "clientProc", "proc", "recordSession", "session", "sessionHeader", "source"}
+	if !slices.Equal(keys, want) {
+		t.Errorf("keys = %v, want %v", keys, want)
+	}
+}
+
+// The Rust decoder reads base64url with no padding.
+func TestWire_EncodeIsUnpadded(t *testing.T) {
+	for _, w := range []Wire{golden, {Agent: "a"}, {Agent: "ab"}, {}} {
+		if s := w.Encode(); strings.Contains(s, "=") {
+			t.Errorf("Encode(%+v) = %q has padding", w, s)
+		}
+	}
+}
diff --git a/core/listener/forwardproxy/egress.go b/core/listener/forwardproxy/egress.go
new file mode 100644
index 000000000..8f219518c
--- /dev/null
+++ b/core/listener/forwardproxy/egress.go
@@ -0,0 +1,95 @@
+package forwardproxy
+
+import (
+	"errors"
+	"fmt"
+	"log/slog"
+	"net"
+	"net/http"
+	"net/http/httputil"
+	"net/url"
+
+	"github.com/rossoctl/cortex/core/attribution"
+	"github.com/rossoctl/cortex/core/listener/egressguard"
+	"github.com/rossoctl/cortex/core/listener/httpx"
+	"github.com/rossoctl/cortex/core/model"
+	"github.com/rossoctl/cortex/core/pipeline"
+)
+
+// NewEgressHandler is the praxis-mode egress listener (spec Section 4, throwaway at phase B).
+//
+// cortex-gateway forwards each recorded request here with X-Cortex-Dest and
+// X-Cortex-Token. The handler checks the token, dials the origin that Dest
+// names through upstream, and relays the response with no buffering.
+func NewEgressHandler(token string, upstream http.RoundTripper) http.Handler {
+	rp := &httputil.ReverseProxy{
+		Transport:     upstream,
+		FlushInterval: -1,
+		Rewrite: func(pr *httputil.ProxyRequest) {
+			dest, _ := parseEgressDest(pr.In.Header.Get(headerDest)) // ServeHTTP checked it
+			pr.Out.URL.Scheme = dest.Scheme
+			pr.Out.URL.Host = dest.Host
+			pr.Out.Host = pr.In.Host
+			// Every hop and attribution header, not only Dest and Token: an origin must not
+			// get the session id or process names when the gateway's strip list is wrong.
+			stripHopHeaders(pr.Out.Header)
+			attribution.Strip(pr.Out.Header)
+			pr.Out.Header.Del(headerEgressError)
+		},
+		// The kind header is the egress listener's own signal. An origin must not set it.
+		ModifyResponse: func(resp *http.Response) error {
+			resp.Header.Del(headerEgressError)
+			return nil
+		},
+		ErrorHandler: func(w http.ResponseWriter, r *http.Request, err error) {
+			dest := pipeline.SanitizeLogValue(r.Header.Get(headerDest))
+			var refusal *egressguard.Refusal
+			if errors.As(err, &refusal) {
+				slog.Warn("egress: destination refused", "dest", dest, "error", pipeline.SanitizeLogValue(err.Error()))
+				w.Header().Set(headerEgressError, "destination_private")
+				// The same body as direct mode (serveOutbound), so the client sees one refusal shape.
+				httpx.WriteRejection(w, pipeline.DenyStatus(http.StatusForbidden, "destination_private",
+					fmt.Sprintf("refusing to reach private destination %s (%s)", refusal.Host, refusal.Addr)))
+				return
+			}
+			if r.Context().Err() != nil {
+				return // the client went away; nobody reads a status
+			}
+			fail := model.TransportError(err)
+			slog.Warn("egress: upstream failed", "dest", dest, "kind", fail.Kind, "error", pipeline.SanitizeLogValue(fail.Message))
+			w.Header().Set(headerEgressError, fail.Kind)
+			w.WriteHeader(http.StatusBadGateway)
+		},
+	}
+	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if !tokenMatches(r.Header.Get(headerToken), token) {
+			w.WriteHeader(http.StatusForbidden)
+			return
+		}
+		raw := r.Header.Get(headerDest)
+		if raw == probeDest {
+			w.WriteHeader(http.StatusNoContent)
+			return
+		}
+		if _, ok := parseEgressDest(raw); !ok {
+			slog.Warn("egress: bad destination", "dest", pipeline.SanitizeLogValue(raw))
+			http.Error(w, "bad "+headerDest, http.StatusBadRequest)
+			return
+		}
+		rp.ServeHTTP(w, r)
+	})
+}
+
+// parseEgressDest accepts only "http://host:port" and "https://host:port" (Decision 1).
+func parseEgressDest(raw string) (*url.URL, bool) {
+	u, err := url.Parse(raw)
+	if err != nil || (u.Scheme != "http" && u.Scheme != "https") || u.User != nil ||
+		u.Path != "" || u.RawQuery != "" || u.Fragment != "" || u.Opaque != "" {
+		return nil, false
+	}
+	host, port, err := net.SplitHostPort(u.Host)
+	if err != nil || host == "" || port == "" {
+		return nil, false
+	}
+	return u, true
+}
diff --git a/core/listener/forwardproxy/egress_test.go b/core/listener/forwardproxy/egress_test.go
new file mode 100644
index 000000000..eaa2d49b3
--- /dev/null
+++ b/core/listener/forwardproxy/egress_test.go
@@ -0,0 +1,266 @@
+package forwardproxy
+
+import (
+	"bufio"
+	"bytes"
+	"compress/gzip"
+	"context"
+	"io"
+	"net"
+	"net/http"
+	"net/http/httptest"
+	"net/netip"
+	"strings"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/rossoctl/cortex/core/attribution"
+	"github.com/rossoctl/cortex/core/listener/egressguard"
+)
+
+const egressTestToken = "tok-123"
+
+// egressRig is an origin and an egress listener in front of it.
+type egressRig struct {
+	origin *httptest.Server
+	egress *httptest.Server
+	hits   atomic.Int32
+}
+
+func newEgressRig(t *testing.T, token string, origin http.HandlerFunc) *egressRig {
+	t.Helper()
+	rig := &egressRig{}
+	rig.origin = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		rig.hits.Add(1)
+		origin(w, r)
+	}))
+	t.Cleanup(rig.origin.Close)
+	rig.egress = httptest.NewServer(NewEgressHandler(token, http.DefaultTransport.(*http.Transport).Clone()))
+	t.Cleanup(rig.egress.Close)
+	return rig
+}
+
+func (rig *egressRig) dest() string { return rig.origin.URL } // http://127.0.0.1:port
+
+func egressDo(t *testing.T, rig *egressRig, path, token, dest string, hdr http.Header) *http.Response {
+	t.Helper()
+	req, err := http.NewRequest(http.MethodGet, rig.egress.URL+path, nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+	for k, vs := range hdr {
+		req.Header[k] = vs
+	}
+	if token != "" {
+		req.Header.Set(headerToken, token)
+	}
+	if dest != "" {
+		req.Header.Set(headerDest, dest)
+	}
+	resp, err := (&http.Transport{Proxy: nil, DisableCompression: true}).RoundTrip(req)
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() { _ = resp.Body.Close() })
+	return resp
+}
+
+func okOrigin(w http.ResponseWriter, _ *http.Request) { w.WriteHeader(http.StatusOK) }
+
+func TestEgress_TokenRequired(t *testing.T) {
+	for _, tc := range []struct{ name, configured, given string }{
+		{"no token", egressTestToken, ""},
+		{"wrong token", egressTestToken, "nope"},
+		{"empty configured token", "", ""},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			rig := newEgressRig(t, tc.configured, okOrigin)
+			resp := egressDo(t, rig, "/", tc.given, rig.dest(), nil)
+			if resp.StatusCode != http.StatusForbidden {
+				t.Fatalf("status = %d, want 403", resp.StatusCode)
+			}
+			if n := rig.hits.Load(); n != 0 {
+				t.Fatalf("origin saw %d requests, want 0", n)
+			}
+		})
+	}
+}
+
+func TestEgress_ProbeAnswers204(t *testing.T) {
+	rig := newEgressRig(t, egressTestToken, okOrigin)
+	resp := egressDo(t, rig, "/", egressTestToken, probeDest, http.Header{headerProbe: {"1"}})
+	if resp.StatusCode != http.StatusNoContent {
+		t.Fatalf("status = %d, want 204", resp.StatusCode)
+	}
+	if n := rig.hits.Load(); n != 0 {
+		t.Fatalf("origin saw %d requests, want 0", n)
+	}
+}
+
+func TestEgress_BadDest400(t *testing.T) {
+	rig := newEgressRig(t, egressTestToken, okOrigin)
+	for _, dest := range []string{"", "ftp://h:1", "h:1", "http://", "http://h", "http://h:1/path"} {
+		req, _ := http.NewRequest(http.MethodGet, rig.egress.URL+"/", nil)
+		req.Header.Set(headerToken, egressTestToken)
+		req.Header.Set(headerDest, dest)
+		resp, err := http.DefaultTransport.RoundTrip(req)
+		if err != nil {
+			t.Fatal(err)
+		}
+		_ = resp.Body.Close()
+		if resp.StatusCode != http.StatusBadRequest {
+			t.Errorf("dest %q: status = %d, want 400", dest, resp.StatusCode)
+		}
+	}
+	if n := rig.hits.Load(); n != 0 {
+		t.Fatalf("origin saw %d requests, want 0", n)
+	}
+}
+
+func TestEgress_RelaysAndStripsHeaders(t *testing.T) {
+	var got *http.Request
+	rig := newEgressRig(t, egressTestToken, func(w http.ResponseWriter, r *http.Request) {
+		got = r.Clone(context.Background())
+		_, _ = io.WriteString(w, "hello")
+	})
+	req, _ := http.NewRequest(http.MethodGet, rig.egress.URL+"/v1/x?a=b", nil)
+	req.Host = "api.example.com"
+	req.Header.Set(headerToken, egressTestToken)
+	req.Header.Set(headerDest, rig.dest())
+	req.Header.Set(headerEgressError, "upstream_dns")
+	req.Header.Set("X-Cortex-Custom", "keep")
+	// Headers a broken gateway strip list would let through.
+	req.Header.Set(attribution.HeaderAttribution, "e30")
+	req.Header.Set(attribution.HeaderSession, "s1")
+	req.Header.Set(headerRequestID, "r1")
+	resp, err := http.DefaultTransport.RoundTrip(req)
+	if err != nil {
+		t.Fatal(err)
+	}
+	body, _ := io.ReadAll(resp.Body)
+	_ = resp.Body.Close()
+	if resp.StatusCode != http.StatusOK || string(body) != "hello" {
+		t.Fatalf("got %d %q, want 200 hello", resp.StatusCode, body)
+	}
+	if got.URL.Path != "/v1/x" || got.URL.RawQuery != "a=b" {
+		t.Errorf("origin URL = %s, want /v1/x?a=b", got.URL)
+	}
+	if got.Host != "api.example.com" {
+		t.Errorf("origin Host = %q, want api.example.com", got.Host)
+	}
+	if got.Header.Get("X-Cortex-Custom") != "keep" {
+		t.Errorf("X-Cortex-Custom did not reach the origin")
+	}
+	for _, h := range []string{headerDest, headerToken, headerEgressError, attribution.HeaderAttribution, attribution.HeaderSession, headerRequestID} {
+		if v := got.Header.Get(h); v != "" {
+			t.Errorf("origin got %s = %q, want none", h, v)
+		}
+	}
+}
+
+func TestEgress_GzipPassesUnchanged(t *testing.T) {
+	var zipped bytes.Buffer
+	zw := gzip.NewWriter(&zipped)
+	_, _ = zw.Write([]byte("compressed body"))
+	_ = zw.Close()
+	rig := newEgressRig(t, egressTestToken, func(w http.ResponseWriter, r *http.Request) {
+		if r.Header.Get("Accept-Encoding") != "gzip" {
+			t.Errorf("origin Accept-Encoding = %q, want gzip", r.Header.Get("Accept-Encoding"))
+		}
+		w.Header().Set("Content-Encoding", "gzip")
+		_, _ = w.Write(zipped.Bytes())
+	})
+	resp := egressDo(t, rig, "/", egressTestToken, rig.dest(), http.Header{"Accept-Encoding": {"gzip"}})
+	body, _ := io.ReadAll(resp.Body)
+	if resp.Header.Get("Content-Encoding") != "gzip" {
+		t.Errorf("Content-Encoding = %q, want gzip", resp.Header.Get("Content-Encoding"))
+	}
+	if !bytes.Equal(body, zipped.Bytes()) {
+		t.Errorf("body changed in transit: got %d bytes, want %d", len(body), zipped.Len())
+	}
+}
+
+func TestEgress_OriginEgressErrorRemoved(t *testing.T) {
+	rig := newEgressRig(t, egressTestToken, func(w http.ResponseWriter, _ *http.Request) {
+		w.Header().Set(headerEgressError, "upstream_dns")
+		w.WriteHeader(http.StatusOK)
+	})
+	resp := egressDo(t, rig, "/", egressTestToken, rig.dest(), nil)
+	if resp.StatusCode != http.StatusOK {
+		t.Fatalf("status = %d, want 200", resp.StatusCode)
+	}
+	if v := resp.Header.Get(headerEgressError); v != "" {
+		t.Fatalf("client got %s = %q, want none", headerEgressError, v)
+	}
+}
+
+func TestEgress_DialFailure502WithKind(t *testing.T) {
+	ln, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatal(err)
+	}
+	closed := "http://" + ln.Addr().String()
+	_ = ln.Close()
+	rig := newEgressRig(t, egressTestToken, okOrigin)
+	resp := egressDo(t, rig, "/", egressTestToken, closed, nil)
+	if resp.StatusCode != http.StatusBadGateway {
+		t.Fatalf("status = %d, want 502", resp.StatusCode)
+	}
+	if v := resp.Header.Get(headerEgressError); v != "upstream_refused" {
+		t.Fatalf("%s = %q, want upstream_refused", headerEgressError, v)
+	}
+}
+
+func TestEgress_GuardRefusal403(t *testing.T) {
+	refusing := &http.Transport{DialContext: func(context.Context, string, string) (net.Conn, error) {
+		return nil, &egressguard.Refusal{Host: "10.0.0.1", Addr: netip.MustParseAddr("10.0.0.1")}
+	}}
+	egress := httptest.NewServer(NewEgressHandler(egressTestToken, refusing))
+	t.Cleanup(egress.Close)
+	req, _ := http.NewRequest(http.MethodGet, egress.URL+"/", nil)
+	req.Header.Set(headerToken, egressTestToken)
+	req.Header.Set(headerDest, "http://10.0.0.1:80")
+	resp, err := http.DefaultTransport.RoundTrip(req)
+	if err != nil {
+		t.Fatal(err)
+	}
+	body, _ := io.ReadAll(resp.Body)
+	_ = resp.Body.Close()
+	// Direct mode's body (serveOutbound): the gateway relays it to the client.
+	if !strings.Contains(string(body), `"error":"destination_private"`) {
+		t.Errorf("body = %q, want the destination_private rejection", body)
+	}
+	if resp.StatusCode != http.StatusForbidden {
+		t.Fatalf("status = %d, want 403", resp.StatusCode)
+	}
+	if v := resp.Header.Get(headerEgressError); v != "destination_private" {
+		t.Fatalf("%s = %q, want destination_private", headerEgressError, v)
+	}
+}
+
+func TestEgress_StreamsFrames(t *testing.T) {
+	release := make(chan struct{})
+	rig := newEgressRig(t, egressTestToken, func(w http.ResponseWriter, _ *http.Request) {
+		w.Header().Set("Content-Type", "text/event-stream")
+		_, _ = io.WriteString(w, "data: one\n\n")
+		w.(http.Flusher).Flush()
+		<-release
+		_, _ = io.WriteString(w, "data: two\n\n")
+	})
+	defer close(release)
+	resp := egressDo(t, rig, "/", egressTestToken, rig.dest(), nil)
+	line := make(chan string, 1)
+	go func() {
+		s, _ := bufio.NewReader(resp.Body).ReadString('\n')
+		line <- s
+	}()
+	select {
+	case s := <-line:
+		if !strings.HasPrefix(s, "data: one") {
+			t.Fatalf("first frame = %q, want data: one", s)
+		}
+	case <-time.After(5 * time.Second):
+		t.Fatal("first frame did not arrive before the origin finished")
+	}
+}
diff --git a/core/listener/forwardproxy/hop.go b/core/listener/forwardproxy/hop.go
new file mode 100644
index 000000000..8531b4004
--- /dev/null
+++ b/core/listener/forwardproxy/hop.go
@@ -0,0 +1,172 @@
+package forwardproxy
+
+import (
+	"context"
+	"errors"
+	"log/slog"
+	"net"
+	"net/http"
+	"sync"
+	"sync/atomic"
+	"time"
+
+	"github.com/rossoctl/cortex/core/attribution"
+	"github.com/rossoctl/cortex/core/model"
+	"github.com/rossoctl/cortex/core/pipeline"
+)
+
+// Hop sends each recorded praxis-mode request to cortex-gateway at the dial (#1721).
+// Only the dial target moves: r.Host and the path stay the client's, and the real
+// destination goes in X-Cortex-Dest. The gateway records the rows of a request the
+// hop owns, so the forward proxy records none for it.
+type Hop struct {
+	gateway string
+	token   string
+	client  *http.Client
+	ready   atomic.Bool
+	owned   sync.Map // *pipeline.Context -> attribution.Wire
+}
+
+// NewHop returns a Hop to the gateway at gatewayAddr (host:port). It is not ready
+// until Probe gets a 204.
+func NewHop(gatewayAddr, token string) *Hop {
+	return &Hop{
+		gateway: gatewayAddr,
+		token:   token,
+		client: &http.Client{
+			// HTTP/1, no proxy env, and Go's default compression.
+			Transport: &http.Transport{
+				Proxy:               nil,
+				ForceAttemptHTTP2:   false,
+				MaxIdleConnsPerHost: 100,
+				IdleConnTimeout:     90 * time.Second,
+				DialContext:         (&net.Dialer{Timeout: 2 * time.Second}).DialContext,
+			},
+			CheckRedirect: func(*http.Request, []*http.Request) error { return http.ErrUseLastResponse },
+		},
+	}
+}
+
+// Ready reports whether a probe has passed.
+func (h *Hop) Ready() bool { return h.ready.Load() }
+
+// Probe sends a probe every `every` until one gets 204 or ctx ends. Logs once when ready.
+func (h *Hop) Probe(ctx context.Context, every time.Duration) {
+	var lastWarn time.Time
+	for {
+		status, err := h.probeOnce(ctx)
+		if status == http.StatusNoContent {
+			h.ready.Store(true)
+			slog.Info("praxis hop: cortex-gateway is ready", "gateway", h.gateway)
+			return
+		}
+		if time.Since(lastWarn) >= 30*time.Second {
+			lastWarn = time.Now()
+			slog.Warn("praxis hop: cortex-gateway is not ready", "gateway", h.gateway, "status", status, "error", err)
+		}
+		select {
+		case <-ctx.Done():
+			return
+		case <-time.After(every):
+		}
+	}
+}
+
+func (h *Hop) probeOnce(ctx context.Context) (int, error) {
+	ctx, cancel := context.WithTimeout(ctx, 2*time.Second)
+	defer cancel()
+	req, err := http.NewRequestWithContext(ctx, http.MethodGet, "http://"+h.gateway+"/", nil)
+	if err != nil {
+		return 0, err
+	}
+	req.Header.Set(headerToken, h.token)
+	req.Header.Set(headerProbe, "1")
+	req.Header.Set(headerDest, probeDest)
+	resp, err := h.client.Do(req)
+	if err != nil {
+		return 0, err
+	}
+	_ = resp.Body.Close()
+	return resp.StatusCode, nil
+}
+
+// own marks pctx as a request the hop sends, with its attribution w (branch point 2).
+func (h *Hop) own(pctx *pipeline.Context, w attribution.Wire) { h.owned.Store(pctx, w) }
+
+// release removes the mark own set.
+func (h *Hop) release(pctx *pipeline.Context) { h.owned.Delete(pctx) }
+
+// owns gives the attribution of a request the hop sends.
+func (h *Hop) owns(pctx *pipeline.Context) (attribution.Wire, bool) {
+	v, ok := h.owned.Load(pctx)
+	if !ok {
+		return attribution.Wire{}, false
+	}
+	return v.(attribution.Wire), true
+}
+
+// hopWire is the attribution the hop sends for pctx, read after hydration (Decision 8).
+func hopWire(pctx *pipeline.Context, clientHeaders http.Header, recordSession string) attribution.Wire {
+	return attribution.Wire{
+		Agent:         attribution.AgentOf(clientHeaders),
+		Proc:          pctx.AttributedProc(),
+		ClientProc:    pctx.AttributedClientProc(),
+		Session:       pctx.SessionID,
+		Source:        string(pctx.SessionSource),
+		SessionHeader: pctx.NamingHeader(), // the pinned value direct mode records; a plugin can write pctx.SessionHeader
+		ChainSession:  pctx.ChainSession,
+		RecordSession: recordSession,
+	}
+}
+
+// hopDest is X-Cortex-Dest for u: <scheme>://<host>:<port>, with the port always present.
+func hopDest(u *http.Request) string {
+	port := u.URL.Port()
+	if port == "" {
+		port = "80"
+		if u.URL.Scheme == "https" {
+			port = "443"
+		}
+	}
+	return u.URL.Scheme + "://" + net.JoinHostPort(u.URL.Hostname(), port)
+}
+
+// hopDo sends r to the gateway in place of client.Do (branch point 4). When the
+// gateway is not ready or refuses the dial, it answers 503 gateway_unavailable and
+// records the request row and the 503 row itself, with answered true.
+func (s *Server) hopDo(w http.ResponseWriter, r *http.Request, pctx *pipeline.Context, tl *tunnelLog, sid string) (resp *http.Response, err error, answered bool) {
+	h := s.Hop
+	if !h.Ready() {
+		s.gatewayUnavailable(w, r, pctx, tl, sid, "not ready")
+		return nil, nil, true
+	}
+	wire, _ := h.owns(pctx) // zero when there is no store: nothing records then
+	r.Header.Set(headerDest, hopDest(r))
+	r.Header.Set(headerToken, h.token)
+	r.Header.Set(headerRequestID, pctx.RequestID())
+	r.Header.Set(attribution.HeaderAttribution, wire.Encode())
+	r.URL.Scheme, r.URL.Host = "http", h.gateway
+	resp, err = h.client.Do(r)
+	var op *net.OpError
+	if errors.As(err, &op) && op.Op == "dial" {
+		s.gatewayUnavailable(w, r, pctx, tl, sid, model.TransportError(err).Message)
+		return nil, nil, true
+	}
+	return resp, err, false
+}
+
+// gatewayUnavailable answers 503 and records both rows, because the gateway never saw
+// the request (Decision 4).
+func (s *Server) gatewayUnavailable(w http.ResponseWriter, r *http.Request, pctx *pipeline.Context, tl *tunnelLog, sid, cause string) {
+	// Release first, so branch point 5 records the 503 row.
+	s.Hop.release(pctx)
+	slog.Debug("praxis hop: cortex-gateway is unavailable", "host", pipeline.SanitizeLogValue(r.Host), "cause", cause)
+	if s.Sessions != nil {
+		s.recordCapturedRequest(tl, pctx, sid, nil)
+	}
+	http.Error(w, `{"error":"gateway unavailable"}`, http.StatusServiceUnavailable)
+	s.recordOutboundResponseEvent(pctx, http.StatusServiceUnavailable, &pipeline.EventError{
+		Kind:    pipeline.ErrorKindGatewayUnavailable,
+		Message: "cortex-gateway is unavailable: " + cause,
+	})
+}
diff --git a/core/listener/forwardproxy/hop_direct_test.go b/core/listener/forwardproxy/hop_direct_test.go
new file mode 100644
index 000000000..30b9289db
--- /dev/null
+++ b/core/listener/forwardproxy/hop_direct_test.go
@@ -0,0 +1,75 @@
+package forwardproxy
+
+import (
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"sync"
+	"testing"
+
+	"github.com/rossoctl/cortex/core/attribution"
+	"github.com/rossoctl/cortex/core/plugins/inferenceparser"
+)
+
+// With Hop nil, none of the five hop branch points in serveOutbound changes a thing:
+// rows and captures as before, the client's hop headers reach the origin, Go's own
+// Accept-Encoding, and the origin gets the dial.
+func TestServeOutbound_DirectModeBranchPoints(t *testing.T) {
+	const jsonAnswer = `{"id":"m1","type":"message","role":"assistant","model":"claude-x","content":[{"type":"text","text":"hi"}],"usage":{"input_tokens":3,"output_tokens":1}}`
+	const sseAnswer = "event: message_start\ndata: {\"type\":\"message_start\",\"message\":{\"id\":\"m2\",\"model\":\"claude-x\",\"usage\":{\"input_tokens\":3,\"output_tokens\":0}}}\n\n" +
+		"event: message_stop\ndata: {\"type\":\"message_stop\"}\n\n"
+	var mu sync.Mutex
+	var seen []http.Header
+	origin := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		_, _ = io.ReadAll(r.Body)
+		mu.Lock()
+		seen = append(seen, r.Header.Clone())
+		mu.Unlock()
+		if r.URL.Query().Get("stream") == "1" {
+			w.Header().Set("Content-Type", "text/event-stream")
+			_, _ = io.WriteString(w, sseAnswer)
+			return
+		}
+		w.Header().Set("Content-Type", "application/json")
+		_, _ = io.WriteString(w, jsonAnswer)
+	}))
+	defer origin.Close()
+	proxy, rec := proxyWithCaptures(t, inferenceparser.NewInferenceParser())
+
+	hopHeaders := http.Header{
+		attribution.HeaderAttribution: {"client-attr"},
+		headerDest:                    {"http://client.example:1"},
+		headerToken:                   {"client-token"},
+		headerRequestID:               {"client-id"},
+		headerProbe:                   {"1"},
+	}
+	for _, q := range []string{"", "?stream=1"} {
+		postThrough(t, proxy, origin.URL+"/v1/messages"+q, `{"model":"claude-x","max_tokens":5,"messages":[{"role":"user","content":"hi"}]}`, hopHeaders)
+	}
+
+	mu.Lock()
+	defer mu.Unlock()
+	if len(seen) != 2 {
+		t.Fatalf("origin got %d requests, want 2", len(seen))
+	}
+	for i, h := range seen {
+		for k, v := range hopHeaders {
+			if got := h.Get(k); got != v[0] {
+				t.Errorf("request %d: origin %s = %q, want the client's %q", i, k, got, v[0])
+			}
+		}
+		if ae := h.Get("Accept-Encoding"); ae != "gzip" {
+			t.Errorf("request %d: origin Accept-Encoding = %q, want Go's gzip", i, ae)
+		}
+	}
+	rec.mu.Lock()
+	defer rec.mu.Unlock()
+	if len(rec.got) != 2 || len(rec.res) != 2 {
+		t.Fatalf("captures: %d request, %d response; want 2 and 2", len(rec.got), len(rec.res))
+	}
+	for id := range rec.got {
+		if rec.res[id] == nil {
+			t.Errorf("request %s has no response capture", id)
+		}
+	}
+}
diff --git a/core/listener/forwardproxy/hop_test.go b/core/listener/forwardproxy/hop_test.go
new file mode 100644
index 000000000..cec3daa70
--- /dev/null
+++ b/core/listener/forwardproxy/hop_test.go
@@ -0,0 +1,470 @@
+package forwardproxy
+
+import (
+	"bufio"
+	"bytes"
+	"compress/gzip"
+	"context"
+	"encoding/base64"
+	"encoding/json"
+	"io"
+	"net"
+	"net/http"
+	"net/http/httptest"
+	"net/url"
+	"strings"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/rossoctl/cortex/core/attribution"
+	"github.com/rossoctl/cortex/core/listener/skiphost"
+	"github.com/rossoctl/cortex/core/pipeline"
+	"github.com/rossoctl/cortex/core/plugins/plugintesting"
+	"github.com/rossoctl/cortex/core/session"
+)
+
+const hopTestToken = "hop-test-token"
+
+// seenRequest is what the fake gateway got.
+type seenRequest struct {
+	host, uri string
+	header    http.Header
+}
+
+// fakeGateway is an httptest.Server that keeps each request and answers with answer.
+type fakeGateway struct {
+	*httptest.Server
+	mu   sync.Mutex
+	seen []seenRequest
+}
+
+func newFakeGateway(t *testing.T, answer http.HandlerFunc) *fakeGateway {
+	t.Helper()
+	g := &fakeGateway{}
+	g.Server = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		g.mu.Lock()
+		g.seen = append(g.seen, seenRequest{host: r.Host, uri: r.RequestURI, header: r.Header.Clone()})
+		g.mu.Unlock()
+		if answer != nil {
+			answer(w, r)
+		}
+	}))
+	t.Cleanup(g.Close)
+	return g
+}
+
+func (g *fakeGateway) requests() []seenRequest {
+	g.mu.Lock()
+	defer g.mu.Unlock()
+	return append([]seenRequest(nil), g.seen...)
+}
+
+func (g *fakeGateway) only(t *testing.T) seenRequest {
+	t.Helper()
+	got := g.requests()
+	if len(got) != 1 {
+		t.Fatalf("gateway got %d requests, want 1", len(got))
+	}
+	return got[0]
+}
+
+// hopProxy is a forward proxy whose Hop dials gatewayAddr. The hop is ready unless
+// notReady. configure, when non-nil, runs before the proxy serves.
+func hopProxy(t *testing.T, gatewayAddr string, notReady bool, configure func(*Server), plugins ...pipeline.Plugin) (*httptest.Server, *session.Store) {
+	t.Helper()
+	p, err := plugintesting.BuildPipeline(plugins)
+	if err != nil {
+		t.Fatal(err)
+	}
+	store := session.New(5*time.Minute, 100, 0)
+	t.Cleanup(store.Close)
+	hop := NewHop(gatewayAddr, hopTestToken)
+	hop.ready.Store(!notReady)
+	srv := &Server{
+		OutboundPipeline: pipeline.NewHolder(p),
+		Sessions:         store,
+		Client:           &http.Client{Transport: &http.Transport{Proxy: nil}},
+		SessionIDHeaders: []string{session.ClaudeCodeSessionHeader},
+		Hop:              hop,
+	}
+	if configure != nil {
+		configure(srv)
+	}
+	ts := httptest.NewUnstartedServer(srv.Handler())
+	ts.Config.ConnContext = srv.ConnContext
+	ts.Start()
+	t.Cleanup(ts.Close)
+	return ts, store
+}
+
+func gatewayAddr(g *fakeGateway) string { return strings.TrimPrefix(g.URL, "http://") }
+
+// hopClient sends through the proxy. It keeps its own Accept-Encoding.
+func hopClient(proxy *httptest.Server) *http.Client {
+	return &http.Client{Transport: &http.Transport{Proxy: http.ProxyURL(mustParseURL(proxy.URL)), DisableCompression: true}}
+}
+
+func doVia(t *testing.T, c *http.Client, req *http.Request) (int, string, http.Header) {
+	t.Helper()
+	resp, err := c.Do(req)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer resp.Body.Close()
+	b, _ := io.ReadAll(resp.Body)
+	return resp.StatusCode, string(b), resp.Header
+}
+
+func decodeWire(t *testing.T, v string) attribution.Wire {
+	t.Helper()
+	b, err := base64.RawURLEncoding.DecodeString(v)
+	if err != nil {
+		t.Fatalf("decode %s %q: %v", attribution.HeaderAttribution, v, err)
+	}
+	var w attribution.Wire
+	if err := json.Unmarshal(b, &w); err != nil {
+		t.Fatal(err)
+	}
+	return w
+}
+
+func TestHop_DialsGatewayKeepsHostAndPath(t *testing.T) {
+	gw := newFakeGateway(t, nil)
+	proxy, _ := hopProxy(t, gatewayAddr(gw), false, nil)
+
+	req, _ := http.NewRequest(http.MethodGet, "http://api.example.com/v1/x?q=1", nil)
+	req.Header.Set(session.ClaudeCodeSessionHeader, "s1")
+	if st, _, _ := doVia(t, hopClient(proxy), req); st != http.StatusOK {
+		t.Fatalf("status = %d, want 200", st)
+	}
+	got := gw.only(t)
+	if got.uri != "/v1/x?q=1" || got.host != "api.example.com" {
+		t.Errorf("gateway saw %q Host %q, want /v1/x?q=1 Host api.example.com", got.uri, got.host)
+	}
+	if d := got.header.Get(headerDest); d != "http://api.example.com:80" {
+		t.Errorf("%s = %q, want http://api.example.com:80", headerDest, d)
+	}
+	if v := got.header.Get(headerToken); v != hopTestToken {
+		t.Errorf("%s = %q, want the hop's token", headerToken, v)
+	}
+	if got.header.Get(headerRequestID) == "" {
+		t.Errorf("%s is empty", headerRequestID)
+	}
+	w := decodeWire(t, got.header.Get(attribution.HeaderAttribution))
+	if w.Session != "s1" || w.RecordSession != "s1" || w.Source != string(pipeline.SessionSourceHeader) {
+		t.Errorf("wire = %+v, want session and recordSession s1 from the header", w)
+	}
+}
+
+// destWriterPlugin writes pctx.Scheme, pctx.Host and a Dest header, but does not redirect.
+type destWriterPlugin struct{}
+
+func (destWriterPlugin) Name() string { return "dest-writer" }
+func (destWriterPlugin) Capabilities() pipeline.PluginCapabilities {
+	return pipeline.PluginCapabilities{}
+}
+func (destWriterPlugin) OnRequest(_ context.Context, pctx *pipeline.Context) pipeline.Action {
+	pctx.Scheme, pctx.Host = "https", "elsewhere.example"
+	pctx.Headers.Set(headerDest, "http://elsewhere.example:1")
+	return pipeline.Action{Type: pipeline.Continue}
+}
+func (destWriterPlugin) OnResponse(context.Context, *pipeline.Context) pipeline.Action {
+	return pipeline.Action{Type: pipeline.Continue}
+}
+
+func TestHop_OnlyTheDialTargetMoves(t *testing.T) {
+	gw := newFakeGateway(t, nil)
+	proxy, _ := hopProxy(t, gatewayAddr(gw), false, nil, destWriterPlugin{})
+
+	req, _ := http.NewRequest(http.MethodGet, "http://api.example.com/p", nil)
+	doVia(t, hopClient(proxy), req)
+	got := gw.only(t)
+	if d := got.header.Get(headerDest); d != "http://api.example.com:80" {
+		t.Errorf("%s = %q, want the real destination http://api.example.com:80", headerDest, d)
+	}
+	if got.host != "api.example.com" || got.uri != "/p" {
+		t.Errorf("gateway saw %q Host %q, want /p Host api.example.com", got.uri, got.host)
+	}
+}
+
+func TestHop_OriginFormStaysDirect(t *testing.T) {
+	gw := newFakeGateway(t, nil)
+	proxy, _ := hopProxy(t, gatewayAddr(gw), false, nil)
+
+	// No proxy: the forward proxy gets "GET /p", with no scheme or host to dial.
+	resp, err := http.Get(proxy.URL + "/p")
+	if err != nil {
+		t.Fatal(err)
+	}
+	_ = resp.Body.Close()
+	if resp.StatusCode != http.StatusBadGateway {
+		t.Errorf("status = %d, want the direct-mode 502", resp.StatusCode)
+	}
+	if n := len(gw.requests()); n != 0 {
+		t.Errorf("gateway got %d requests, want 0", n)
+	}
+}
+
+// sessionHeaderWriterPlugin writes pctx.SessionHeader, which the Wire must not read.
+type sessionHeaderWriterPlugin struct{}
+
+func (sessionHeaderWriterPlugin) Name() string { return "session-header-writer" }
+func (sessionHeaderWriterPlugin) Capabilities() pipeline.PluginCapabilities {
+	return pipeline.PluginCapabilities{}
+}
+func (sessionHeaderWriterPlugin) OnRequest(_ context.Context, pctx *pipeline.Context) pipeline.Action {
+	pctx.SessionHeader = "X-Plugin-Wrote-This"
+	return pipeline.Action{Type: pipeline.Continue}
+}
+func (sessionHeaderWriterPlugin) OnResponse(context.Context, *pipeline.Context) pipeline.Action {
+	return pipeline.Action{Type: pipeline.Continue}
+}
+
+func TestHop_WireSessionHeaderIsPinned(t *testing.T) {
+	gw := newFakeGateway(t, nil)
+	proxy, _ := hopProxy(t, gatewayAddr(gw), false, nil, sessionHeaderWriterPlugin{})
+
+	req, _ := http.NewRequest(http.MethodGet, "http://api.example.com/p", nil)
+	req.Header.Set(session.ClaudeCodeSessionHeader, "s1")
+	doVia(t, hopClient(proxy), req)
+	w := decodeWire(t, gw.only(t).header.Get(attribution.HeaderAttribution))
+	if w.SessionHeader == "X-Plugin-Wrote-This" || w.SessionHeader == "" {
+		t.Errorf("wire SessionHeader = %q, want the header that named the session", w.SessionHeader)
+	}
+}
+
+func TestHop_ClientHopHeadersNeverReachGateway(t *testing.T) {
+	gw := newFakeGateway(t, nil)
+	proxy, _ := hopProxy(t, gatewayAddr(gw), false, nil)
+
+	req, _ := http.NewRequest(http.MethodGet, "http://api.example.com/p", nil)
+	forged := attribution.Wire{Session: "forged"}.Encode()
+	req.Header.Set(attribution.HeaderAttribution, forged)
+	req.Header.Set(headerDest, "http://evil.example:1")
+	req.Header.Set(headerToken, "forged-token")
+	req.Header.Set(headerRequestID, "forged-id")
+	req.Header.Set(headerProbe, "1")
+	req.Header.Set("X-Cortex-Custom", "v")
+	doVia(t, hopClient(proxy), req)
+
+	h := gw.only(t).header
+	if h.Get(headerDest) != "http://api.example.com:80" || h.Get(headerToken) != hopTestToken {
+		t.Errorf("Dest %q Token %q, want the hop's own", h.Get(headerDest), h.Get(headerToken))
+	}
+	if id := h.Get(headerRequestID); id == "" || id == "forged-id" {
+		t.Errorf("%s = %q, want the hop's own", headerRequestID, id)
+	}
+	if a := h.Get(attribution.HeaderAttribution); a == forged || decodeWire(t, a).Session == "forged" {
+		t.Errorf("%s is the client's", attribution.HeaderAttribution)
+	}
+	if _, ok := h[headerProbe]; ok {
+		t.Errorf("gateway got %s", headerProbe)
+	}
+	if v := h.Get("X-Cortex-Custom"); v != "v" {
+		t.Errorf("X-Cortex-Custom = %q, want v", v)
+	}
+	for _, k := range clientHopHeaders {
+		if n := len(h.Values(k)); n > 1 {
+			t.Errorf("%s has %d values, want 1", k, n)
+		}
+	}
+}
+
+func TestHop_OwnsEncoding(t *testing.T) {
+	gw := newFakeGateway(t, func(w http.ResponseWriter, _ *http.Request) {
+		var b bytes.Buffer
+		zw := gzip.NewWriter(&b)
+		_, _ = zw.Write([]byte("hello"))
+		_ = zw.Close()
+		w.Header().Set("Content-Encoding", "gzip")
+		_, _ = w.Write(b.Bytes())
+	})
+	proxy, _ := hopProxy(t, gatewayAddr(gw), false, nil)
+
+	req, _ := http.NewRequest(http.MethodGet, "http://api.example.com/p", nil)
+	req.Header.Set("Accept-Encoding", "gzip, deflate, br")
+	_, body, _ := doVia(t, hopClient(proxy), req)
+	if ae := gw.only(t).header.Get("Accept-Encoding"); ae != "gzip" {
+		t.Errorf("gateway Accept-Encoding = %q, want gzip only", ae)
+	}
+	if body != "hello" {
+		t.Errorf("client body = %q, want it decoded", body)
+	}
+}
+
+func TestHop_NoRowsForHoppedRequest(t *testing.T) {
+	gw := newFakeGateway(t, func(w http.ResponseWriter, r *http.Request) {
+		if r.URL.Path == "/sse" {
+			w.Header().Set("Content-Type", "text/event-stream")
+			_, _ = io.WriteString(w, "data: one\n\ndata: two\n\n")
+			return
+		}
+		_, _ = io.WriteString(w, "ok")
+	})
+	proxy, store := hopProxy(t, gatewayAddr(gw), false, nil)
+	for _, p := range []string{"/json", "/sse"} {
+		req, _ := http.NewRequest(http.MethodGet, "http://api.example.com"+p, nil)
+		doVia(t, hopClient(proxy), req)
+	}
+	if ev := allEvents(t, store); len(ev) != 0 {
+		t.Errorf("store holds %d rows, want none: the gateway records a hopped request", len(ev))
+	}
+}
+
+// The process-attribution twin of TestProcessAttribution_AServicesToolJoinsTheSessionThatAnsweredLast:
+// only touch makes the tool join a, the session that answered last.
+func TestHop_TouchStillRuns(t *testing.T) {
+	procs := newFakeProcs(fproc(500, 1, opencodeExe), fproc(600, 500, "/usr/bin/curl"))
+	release, arrived := make(chan struct{}), make(chan struct{})
+	var releaseOnce sync.Once
+	unblock := func() { releaseOnce.Do(func() { close(release) }) }
+	t.Cleanup(unblock)
+	gw := newFakeGateway(t, func(_ http.ResponseWriter, r *http.Request) {
+		if r.URL.Path == "/a" {
+			close(arrived)
+			<-release
+		}
+	})
+	proxy, _ := hopProxy(t, gatewayAddr(gw), false, func(s *Server) {
+		s.SessionIDHeaders = append(s.SessionIDHeaders, "X-Opencode-Session-Id")
+		s.ClientAffinity = true
+		s.Processes = procs
+	})
+	svc := procs.clientFor(proxy.URL, 500)
+
+	done := make(chan struct{})
+	go func() {
+		defer close(done)
+		sendAs(t, svc, "http://api.example.com/a", "opencode/latest/2.0.21/cli", "X-Opencode-Session-Id", "ses_a")
+	}()
+	<-arrived
+	sendAs(t, svc, "http://api.example.com/b", "opencode/latest/2.0.21/cli", "X-Opencode-Session-Id", "ses_b")
+	unblock()
+	<-done
+	sendAs(t, procs.clientFor(proxy.URL, 600), "http://api.example.com/tool", "curl/8.7.1", "", "")
+
+	for _, r := range gw.requests() {
+		if r.uri == "/tool" {
+			if w := decodeWire(t, r.header.Get(attribution.HeaderAttribution)); w.Session != "ses_a" {
+				t.Errorf("tool session = %q, want ses_a (the session that answered last)", w.Session)
+			}
+			return
+		}
+	}
+	t.Fatal("the gateway did not get /tool")
+}
+
+func assertGatewayUnavailableRows(t *testing.T, store *session.Store) {
+	t.Helper()
+	ev := allEvents(t, store)
+	if len(ev) != 2 {
+		t.Fatalf("store holds %d rows, want the request row and its 503 row: %+v", len(ev), ev)
+	}
+	req, resp := ev[0], ev[1]
+	if req.Phase != pipeline.SessionRequest || resp.Phase != pipeline.SessionResponse {
+		t.Fatalf("phases = %q, %q", req.Phase, resp.Phase)
+	}
+	if resp.StatusCode != http.StatusServiceUnavailable || resp.Error == nil || resp.Error.Kind != pipeline.ErrorKindGatewayUnavailable {
+		t.Errorf("response row = %d %+v, want 503 %s", resp.StatusCode, resp.Error, pipeline.ErrorKindGatewayUnavailable)
+	}
+	if resp.RequestID == "" || resp.RequestID != req.RequestID {
+		t.Errorf("response RequestID %q, want the request's %q", resp.RequestID, req.RequestID)
+	}
+}
+
+func TestHop_NotReadyIsGatewayUnavailable(t *testing.T) {
+	gw := newFakeGateway(t, nil)
+	proxy, store := hopProxy(t, gatewayAddr(gw), true, nil)
+	req, _ := http.NewRequest(http.MethodGet, "http://api.example.com/p", nil)
+	if st, _, _ := doVia(t, hopClient(proxy), req); st != http.StatusServiceUnavailable {
+		t.Errorf("status = %d, want 503", st)
+	}
+	if n := len(gw.requests()); n != 0 {
+		t.Errorf("gateway got %d requests, want none", n)
+	}
+	assertGatewayUnavailableRows(t, store)
+}
+
+func TestHop_DialFailureIsGatewayUnavailable(t *testing.T) {
+	proxy, store := hopProxy(t, closedAddr(t), false, nil)
+	req, _ := http.NewRequest(http.MethodGet, "http://api.example.com/p", nil)
+	if st, _, _ := doVia(t, hopClient(proxy), req); st != http.StatusServiceUnavailable {
+		t.Errorf("status = %d, want 503 (not 502 upstream_refused)", st)
+	}
+	assertGatewayUnavailableRows(t, store)
+}
+
+func TestHop_ProbeSetsReady(t *testing.T) {
+	var n atomic.Int32
+	gw := newFakeGateway(t, func(w http.ResponseWriter, _ *http.Request) {
+		if n.Add(1) < 3 {
+			w.WriteHeader(http.StatusServiceUnavailable)
+			return
+		}
+		w.WriteHeader(http.StatusNoContent)
+	})
+	hop := NewHop(gatewayAddr(gw), hopTestToken)
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	defer cancel()
+	hop.Probe(ctx, 10*time.Millisecond)
+	if !hop.Ready() {
+		t.Fatal("Ready() = false after a 204")
+	}
+	got := gw.requests()
+	if len(got) != 3 {
+		t.Errorf("gateway got %d probes, want 3", len(got))
+	}
+	for _, r := range got {
+		if r.header.Get(headerToken) != hopTestToken || r.header.Get(headerProbe) != "1" || r.header.Get(headerDest) != probeDest {
+			t.Errorf("probe headers = %v", r.header)
+		}
+	}
+}
+
+func TestHop_SkipHostsAndUpgradesDoNotHop(t *testing.T) {
+	gw := newFakeGateway(t, nil)
+	var originHits atomic.Int32
+	origin := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		originHits.Add(1)
+		if r.Header.Get("Upgrade") != "" {
+			wsEchoHandler(w, r)
+		}
+	}))
+	t.Cleanup(origin.Close)
+
+	skip, err := skiphost.New([]string{"127.0.0.1"})
+	if err != nil {
+		t.Fatal(err)
+	}
+	proxy, _ := hopProxy(t, gatewayAddr(gw), false, func(s *Server) { s.SkipHosts = skip })
+	req, _ := http.NewRequest(http.MethodGet, origin.URL+"/skipped", nil)
+	if st, _, _ := doVia(t, hopClient(proxy), req); st != http.StatusOK {
+		t.Errorf("skipped status = %d, want 200", st)
+	}
+
+	proxy2, _ := hopProxy(t, gatewayAddr(gw), false, nil)
+	conn, err := net.Dial("tcp", strings.TrimPrefix(proxy2.URL, "http://"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer conn.Close()
+	_ = conn.SetDeadline(time.Now().Add(10 * time.Second))
+	u, _ := url.Parse(origin.URL)
+	writeUpgradeRequest(conn, origin.URL+"/ws", u.Host)
+	resp, err := http.ReadResponse(bufio.NewReader(conn), nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if resp.StatusCode != http.StatusSwitchingProtocols {
+		t.Errorf("upgrade status = %d, want 101 from the origin", resp.StatusCode)
+	}
+
+	if n := len(gw.requests()); n != 0 {
+		t.Errorf("gateway got %d requests, want none", n)
+	}
+	if n := originHits.Load(); n != 2 {
+		t.Errorf("origin got %d requests, want 2", n)
+	}
+}
diff --git a/core/listener/forwardproxy/hopheaders.go b/core/listener/forwardproxy/hopheaders.go
new file mode 100644
index 000000000..ddfc3bb3c
--- /dev/null
+++ b/core/listener/forwardproxy/hopheaders.go
@@ -0,0 +1,36 @@
+package forwardproxy
+
+import (
+	"crypto/subtle"
+	"net/http"
+
+	"github.com/rossoctl/cortex/core/attribution"
+)
+
+// The hop headers between Cortex, cortex-gateway and the egress listener
+// (praxis mode, #1721).
+const (
+	headerAttribution = attribution.HeaderAttribution
+	headerDest        = "X-Cortex-Dest"
+	headerToken       = "X-Cortex-Token"
+	headerRequestID   = "X-Cortex-Request-Id"
+	headerProbe       = "X-Cortex-Probe"
+	headerEgressError = "X-Cortex-Egress-Error"
+	// probeDest has no "://", so it can never name an origin (Decision 2).
+	probeDest = "cortex-probe"
+)
+
+// clientHopHeaders are the five hop headers a client must not send in praxis mode.
+var clientHopHeaders = []string{headerAttribution, headerDest, headerToken, headerRequestID, headerProbe}
+
+// stripHopHeaders removes clientHopHeaders from h.
+func stripHopHeaders(h http.Header) {
+	for _, k := range clientHopHeaders {
+		h.Del(k)
+	}
+}
+
+// tokenMatches compares in constant time. An empty token matches nothing.
+func tokenMatches(given, token string) bool {
+	return token != "" && subtle.ConstantTimeCompare([]byte(given), []byte(token)) == 1
+}
diff --git a/core/listener/forwardproxy/record.go b/core/listener/forwardproxy/record.go
index b5c1982c4..b929b1539 100644
--- a/core/listener/forwardproxy/record.go
+++ b/core/listener/forwardproxy/record.go
@@ -99,6 +99,12 @@ func (s *Server) recordOutboundResponseEvent(pctx *pipeline.Context, statusCode
 	if s.Sessions == nil {
 		return
 	}
+	// Hop branch point 5: the gateway records the response row of a request the hop sends.
+	if s.Hop != nil {
+		if _, ok := s.Hop.owns(pctx); ok {
+			return
+		}
+	}
 	// Prefer the session pinned when the request event was recorded so the
 	// response lands in the same bucket. Fall back to ActiveSession() only
 	// when nothing was pinned (defensive — a response-only path), then to
diff --git a/core/listener/forwardproxy/server.go b/core/listener/forwardproxy/server.go
index 8c900ebc0..20b54dde2 100644
--- a/core/listener/forwardproxy/server.go
+++ b/core/listener/forwardproxy/server.go
@@ -161,6 +161,10 @@ type Server struct {
 	// config.ListenerConfig.EgressGuard.
 	EgressGuard *egressguard.Guard
 
+	// Hop, when non-nil, sends each recorded request to cortex-gateway at the dial
+	// (praxis mode, #1721). nil in direct mode, and then serveOutbound is as before.
+	Hop *Hop
+
 	// Processes, when non-nil, names the process behind each client connection, and a
 	// request with no session header is filed under that process's session — or its
 	// nearest ancestor's — before client affinity is consulted; see core/attribution. Set only where every client is a process on this
@@ -445,6 +449,10 @@ func (s *Server) serveOutbound(w http.ResponseWriter, r *http.Request, tl *tunne
 	// Before anything reads the headers: a client must not assert its own attribution, and
 	// no hop in front of this listener is trusted to (attribution.Strip).
 	attribution.Strip(r.Header)
+	// Hop branch point 1: a client must not send its own hop headers.
+	if s.Hop != nil {
+		stripHopHeaders(r.Header)
+	}
 	pctx := &pipeline.Context{
 		Direction: pipeline.Outbound,
 		Method:    r.Method,
@@ -698,6 +706,9 @@ func (s *Server) serveOutbound(w http.ResponseWriter, r *http.Request, tl *tunne
 		}
 	}
 
+	// Hop branch point 2: the gateway records the rows of a request the hop sends.
+	// An origin-form request has no dial target; it stays on the direct path, where client.Do refuses it with a 502.
+	hopped := s.Hop != nil && !skipped && !forwardUpgrade && r.URL.Scheme != "" && r.URL.Host != ""
 	if !skipped && s.Sessions != nil {
 		sid := s.recordingSessionID(sessionID, r.Header, chain)
 		// Pin this session so the paired response event records into the
@@ -716,7 +727,12 @@ func (s *Server) serveOutbound(w http.ResponseWriter, r *http.Request, tl *tunne
 		// operator wants to see (it carries Host, and the paired response
 		// carries StatusCode). skip_hosts traffic never reaches here (the
 		// !skipped guard above), so it stays suppressed by design.
-		s.recordCapturedRequest(tl, pctx, sid, capture)
+		if hopped {
+			s.Hop.own(pctx, hopWire(pctx, r.Header, sid))
+			defer s.Hop.release(pctx)
+		} else {
+			s.recordCapturedRequest(tl, pctx, sid, capture)
+		}
 	}
 
 	// Propagate every header mutation the outbound pipeline made to the
@@ -744,7 +760,8 @@ func (s *Server) serveOutbound(w http.ResponseWriter, r *http.Request, tl *tunne
 	// falls back to the current pipeline (Holder.ForPctx).
 	pinned := s.OutboundPipeline.ForPctx(pctx)
 
-	ownEncoding := !skipped && (pinned.NeedsResponseBody() || pinned.HasStreamingResponders())
+	// Hop branch point 3: the hop always owns the encoding.
+	ownEncoding := hopped || (!skipped && (pinned.NeedsResponseBody() || pinned.HasStreamingResponders()))
 	stripForUpstream(r, ownEncoding, forwardUpgrade)
 
 	// Clear RequestURI — set by the server but must be empty for client requests
@@ -784,7 +801,17 @@ func (s *Server) serveOutbound(w http.ResponseWriter, r *http.Request, tl *tunne
 		// to), so the exchange rides this client as HTTP/1.1.
 		client = s.TLSBridge.Upstream
 	}
-	resp, err := client.Do(r)
+	// Hop branch point 4: after the redirect re-check, dial the gateway in place of the origin.
+	var resp *http.Response
+	var err error
+	if hopped {
+		var answered bool
+		if resp, err, answered = s.hopDo(w, r, pctx, tl, pctx.OutboundSessionID); answered {
+			return
+		}
+	} else {
+		resp, err = client.Do(r)
+	}
 	// A Resender reads refused answers, and a 101 is not one — it is the upgrade
 	// succeeding. Peeking at its (empty) body and sending the GET again would tear
 	// down the connection the origin just switched, so a forwarded upgrade never
diff --git a/core/model/session.go b/core/model/session.go
index 0194d297c..a5d965556 100644
--- a/core/model/session.go
+++ b/core/model/session.go
@@ -749,6 +749,11 @@ const ErrorKindTurnInterrupted = "turn_interrupted"
 // shared conformance suite matches without a translation table.
 const ErrorKindDestinationPrivate = "destination_private"
 
+// ErrorKindGatewayUnavailable is the praxis hop answering 503 itself because
+// cortex-gateway is not ready or does not accept the dial (#1721). Cortex records
+// the request row and this row, because the gateway never saw the request.
+const ErrorKindGatewayUnavailable = "gateway_unavailable"
+
 // SessionSource is how a listener resolved Context.Session.
 type SessionSource string
 
diff --git a/core/pipeline/model_alias.go b/core/pipeline/model_alias.go
index eadf47d10..cfdb74a19 100644
--- a/core/pipeline/model_alias.go
+++ b/core/pipeline/model_alias.go
@@ -148,6 +148,7 @@ const (
 	ErrorKindUpgradeDeclined    = model.ErrorKindUpgradeDeclined
 	ErrorKindTurnInterrupted    = model.ErrorKindTurnInterrupted
 	ErrorKindDestinationPrivate = model.ErrorKindDestinationPrivate
+	ErrorKindGatewayUnavailable = model.ErrorKindGatewayUnavailable
 	SessionSourceHeader         = model.SessionSourceHeader
 	SessionSourceProcess        = model.SessionSourceProcess
 	SessionSourcePending        = model.SessionSourcePending
diff --git a/docs/superpowers/plans/2026-11-09-praxis-mode-4-hop-and-egress.md b/docs/superpowers/plans/2026-11-09-praxis-mode-4-hop-and-egress.md
new file mode 100644
index 000000000..773a5cc1a
--- /dev/null
+++ b/docs/superpowers/plans/2026-11-09-praxis-mode-4-hop-and-egress.md
@@ -0,0 +1,434 @@
+# Praxis mode 4/8: the hop and the egress listener — Implementation Plan
+
+> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
+
+**Goal:** In `praxis.enabled` mode, send each recorded request from the forward proxy through `cortex-gateway` and back out through a Cortex egress listener, with no change to direct mode (#1721).
+
+**Architecture:** A dial-only hop (`core/listener/forwardproxy/hop.go`) moves the dial of a non-skipped request to `praxis.listen` over HTTP/1, with five branch points in `serveOutbound`. The gateway records the rows and forwards to the egress listener (`egress.go`) on `praxis.egress`, which dials the origin with the TLS bridge's upstream options. Ingestion (`core/sessionapi/ingest.go`) joins the gateway's rows to Cortex state (tunnel pairing, Cortex's own cuts) through a `HopJoins` interface that the hop implements.
+
+**Tech Stack:** Go 1.26.10 (`core`, `cmd/cortex`), Rust (`praxis/`, Praxis 0.8.0 API).
+
+**Spec:** `docs/superpowers/specs/2026-11-07-praxis-mode-design.md` (Sections 1 "Joins", 2, 4 "The hop", "The egress listener", "Fail closed", "Config", 5). Issue: #1721. Earlier plans: `2026-11-09-praxis-mode-2-gateway-and-ingest.md`, `2026-11-09-praxis-mode-3-supervise-gateway.md`. The W1 spike `5ee3950b` (`spike/hairpin-w1`) is a reference only.
+
+## Global Constraints
+
+- Direct mode does not change. Every existing Go test passes with no change to its expectations.
+- `serveOutbound` gets exactly five branch points for the hop, and no other. Each one tests `s.Hop != nil` or the local `hopped`.
+- The hop speaks HTTP/1, reads no proxy env (`Proxy: nil`) and keeps Go's default compression (no `DisableCompression`).
+- The hop sets only `r.URL.Scheme`, `r.URL.Host` and the transport. `r.Host` and the path do not change. It is not `pctx.Redirect`.
+- Header names, verbatim: `X-Cortex-Attribution`, `X-Cortex-Dest`, `X-Cortex-Token`, `X-Cortex-Request-Id`, `X-Cortex-Probe`, `X-Cortex-Egress-Error`.
+- Join entries live at most 60 s (`hopJoinTTL = 60 * time.Second`).
+- New error kind: `gateway_unavailable`, status 503.
+- Run per-module Go commands with `GOWORK=off`. Log with `log/slog`.
+- Comments and commit messages in ASD-STE100 style. Commits use `git commit -s` and end with `Assisted-By: Claude (Anthropic AI) <noreply@anthropic.com>`. No `Co-Authored-By`.
+- Prose says Cortex for the product (CLAUDE.md naming rule).
+
+## Decisions (made for this plan, with reasons)
+
+1. **`X-Cortex-Dest` is `<scheme>://<host>:<port>`**, port always present (80 for http, 443 for https when the URL has none). Reason: the egress listener must know the scheme, and the spike used this form.
+2. **The probe is `GET /` to the gateway with `X-Cortex-Token`, `X-Cortex-Probe: 1` and `X-Cortex-Dest: cortex-probe`.** The egress listener answers `204` to a valid token with that Dest and dials nothing. Reason: the probe then passes through the whole chain (Cortex → gateway → egress), and `cortex-probe` has no `://`, so it can never name an origin.
+3. **The probe runs every 1 s until one passes, then stops.** After that a failed dial to the gateway gives 503 `gateway_unavailable` per request. Reason: the spec asks for fail closed at startup and while the gateway is down. A per-request dial check covers "down" with no second state machine.
+4. **The hop records `gateway_unavailable` itself: the request row and a 503 response row**, through the normal tunnel pairing path. Reason: Praxis never saw the request, so no other row exists (spec, "Who records which rows").
+5. **Branch point 5 lives in `recordOutboundResponseEvent`**, the one function every response-row site calls. When the hop owns `pctx`, it records no row. When `fail.Kind == "proxy_error"` (Cortex's own idle cut), it keeps that error as a cut join for the request id. Reason: one branch for all response sites, and the cut is known exactly there.
+6. **The hop marks the requests it owns in a `sync.Map` keyed by `*pipeline.Context`**, set at branch point 2 and removed when `serveOutbound` returns. Reason: no change to `pipeline.Context`, which other repositories consume.
+7. **The request id on the wire is `pctx.RequestID()`.** Reason: the gateway uses `X-Cortex-Request-Id` as the line's `requestId`, so both sides use one id.
+8. **`attribution.Wire` is filled from `pctx`** after hydration: `Proc`, `ClientProc` (`AttributedProc`, `AttributedClientProc`), `Session` (`pctx.SessionID`), `Source`, `SessionHeader`, `ChainSession`, `RecordSession` (the recording sid), and `Agent` from `attribution.AgentOf(r.Header)` read at branch point 2. Reason: the hop must not resolve attribution a second time.
+9. **The gateway keeps `X-Cortex-Dest` and `X-Cortex-Token` for its upstream**, and its request capture still removes all twelve hop headers. Reason: the interim upstream is always the egress listener, which removes both. The token must never be in a capture.
+10. **The egress listener answers an egress-guard refusal with 403 and `X-Cortex-Egress-Error: destination_private`**, and the Rust `egress_kind` learns `destination_private`. Reason: direct mode records that refusal as `destination_private`, so both modes give the same row.
+11. **The egress listener uses `bridge.Upstream.Transport` when the TLS bridge is on, else a `tlsbridge.NewUpstreamClient(nil, false, upOpts...)` transport.** Reason: the spec's "same upstream options as the TLS bridge" (`egress_guard`, CA bundle, `upstream_insecure`).
+12. **`HopJoins` is an interface in `core/sessionapi`.** `forwardproxy` does not import `sessionapi`. `cmd/cortex` has the compile-time check. Reason: no new package and no import cycle.
+13. **Release of an erased line's join marks the tunnel answered**, so the tunnel writes neither its open row nor a close row. Reason: spec Section 1, "A dropped line also releases its tunnel join with no write".
+14. **A join that expires** removes one "awaiting" count and settles the tunnel as today (open row plus a "carried nothing" close row). Reason: spec, known edge case in #1721.
+15. **The end-to-end tests run in process in `cmd/cortex`**, behind the `praxis_e2e` build tag and `CORTEX_GATEWAY_BIN`, with the real `cortex-gateway` binary. The client sends plain HTTP through the forward proxy, and a test dialer maps every origin host to the mock origin. Reason: real names such as `chatgpt.com` must not reach DNS, and the TLS bridge needs no new test setup here. Tunnel pairing is a forwardproxy contract test (Task 6). The supervisor is covered by #1720's tests.
+16. **The e2e test uses the same wiring function as `main`** (`startPraxis`, Task 7). Reason: the test must not drift from the real wiring.
+17. **CI:** the `Praxis` workflow builds `cortex-gateway` and runs the e2e tests after `cargo test`. Reason: spec, "only on `praxis/**` changes and before a release". The workflow's blocking rule stays as it is.
+
+## Review Focus
+
+1. A client that sends its own `X-Cortex-Token` or `X-Cortex-Dest` in praxis mode: the hop must remove them at entry and set its own. Pinned in Task 4 (`TestHop_ClientHopHeadersNeverReachGateway`).
+2. The gateway goes down after the probe passed: each request must get 503 `gateway_unavailable` with both rows, not a 502 `upstream_refused`. Pinned in Task 4 (`TestHop_DialFailureIsGatewayUnavailable`).
+3. A bridged tunnel whose request row arrives after the client closed the tunnel: one pair, and no "carried nothing" close row. Pinned in Task 6 (`TestHopJoin_TunnelClosedBeforeRowGetsOnePair`).
+4. An origin that sends its own `X-Cortex-Egress-Error`: the header must not reach the gateway. Pinned in Task 3 (`TestEgress_OriginEgressErrorRemoved`).
+5. A reload that adds an outbound plugin in praxis mode: no pipeline is built, and the skip is logged. Pinned in Task 7 (`TestOutboundEntries_PraxisSkipsPlugins`).
+
+---
+
+## File map
+
+| File | Task | Change |
+|---|---|---|
+| `praxis/filters/src/record/attribution.rs`, `praxis/filters/src/filters/attribution.rs`, `praxis/filters/src/record/row.rs`, `praxis/server/tests/gateway.rs`, `praxis/README.md` | 1 | keep Dest and Token for the upstream, `destination_private` kind |
+| `core/attribution/wire.go`, `core/attribution/wire_test.go`, `core/attribution/testdata/wire.golden`, `praxis/filters/tests/attribution_wire.rs` | 2 | `attribution.Wire` encoder and its cross-language golden |
+| `core/listener/forwardproxy/hopheaders.go`, `egress.go`, `egress_test.go` | 3 | header constants, the egress listener |
+| `core/listener/forwardproxy/hop.go`, `hop_test.go`, `hop_direct_test.go`, `server.go`, `record.go`, `core/model/session.go`, `core/pipeline/model_alias.go` | 4 | the hop, five branch points, `gateway_unavailable`, the probe |
+| `core/sessionapi/ingest.go`, `core/sessionapi/ingest_joins_test.go` | 5 | `HopJoins` and its use in `fileIngest` |
+| `core/listener/forwardproxy/hopjoins.go`, `hopjoins_test.go`, `tunnel.go`, `hop.go` | 6 | the join registry and the tunnel hold |
+| `cmd/cortex/praxis.go`, `praxis_test.go`, `ingest.go`, `ingest_test.go`, `main.go`, `gateway.go`, `core/config/listenerplan.go`, `core/config/listenerplan_test.go`, `core/config/config.go`, `docs/session-api.md`, `praxis/README.md` | 7 | wiring, config rules, docs |
+| `cmd/cortex/praxis_e2e_test.go`, `.github/workflows/praxis.yaml` | 8 | end-to-end tests and CI |
+
+## PRs
+
+| PR | Title | Parent | Tasks | estLines |
+|---|---|---|---|---|
+| P1 | `Feat: Add the praxis hop and the egress listener to the forward proxy` | "" | 1, 2, 3, 4 | 1250 |
+| P2 | `Feat: Join gateway rows to tunnels and turn on the hop in praxis mode` | P1 | 5, 6, 7 | 1050 |
+| P3 | `Test: Run the praxis-mode end-to-end checks through cortex-gateway` | P2 | 8 | 850 |
+
+P1 is inert alone: nothing sets `Server.Hop` until P2. P3 needs the wiring of P2.
+
+---
+
+### Task 1: The gateway keeps `X-Cortex-Dest` and `X-Cortex-Token` for the egress listener
+
+Depends on: none. Estimated lines: 80.
+
+**Files:**
+- Modify: `praxis/filters/src/record/attribution.rs` (`HOP_HEADERS` stays 12 entries for the capture; add `UPSTREAM_STRIP_HEADERS: [&str; 10]`, the ten without `X-Cortex-Dest` and `TOKEN_HEADER`)
+- Modify: `praxis/filters/src/filters/attribution.rs` (`request_headers_to_remove` uses `UPSTREAM_STRIP_HEADERS`)
+- Modify: `praxis/filters/src/record/row.rs` (`egress_kind`: `"destination_private" => "destination_private"`)
+- Modify: `praxis/server/tests/gateway.rs` (`hop_headers_removed_custom_kept`)
+- Modify: `praxis/README.md` (gateway decision 20)
+
+- [ ] **Step 1: Change the tests first.**
+  - In `hop_headers_removed_custom_kept`: assert that `x-cortex-dest` and `x-cortex-token` now reach the upstream with the values sent, and that the other ten hop headers do not. `X-Cortex-Custom` still reaches it. The request capture in the ingested line still has none of the twelve.
+  - In `row.rs` tests: `egress_kind("destination_private") == "destination_private"`.
+- [ ] **Step 2:** `cd praxis && cargo test --locked` fails on both.
+- [ ] **Step 3:** Implement. Doc comment on `UPSTREAM_STRIP_HEADERS`: the interim upstream is always the egress listener, which removes Dest and Token. At phase B both go on this list.
+- [ ] **Step 4:** Update README decision 20: the gateway keeps Dest and Token for its upstream, the egress listener (#1721), and removes them before any other upstream.
+- [ ] **Step 5:** `cd praxis && cargo fmt --all --check && cargo clippy --all-targets --locked -- -D warnings && cargo test --locked` passes.
+- [ ] **Step 6:** Commit `feat: Keep X-Cortex-Dest and X-Cortex-Token for the egress listener`.
+
+### Task 2: `attribution.Wire`, the Go encoder of `X-Cortex-Attribution`
+
+Depends on: none. Estimated lines: 150.
+
+**Files:**
+- Modify: `core/attribution/wire.go`
+- Create: `core/attribution/wire_test.go`, `core/attribution/testdata/wire.golden`
+- Create: `praxis/filters/tests/attribution_wire.rs`
+
+**Interfaces:**
+- Produces:
+  ```go
+  // Wire is the attribution result that the hop sends in X-Cortex-Attribution.
+  type Wire struct {
+      Agent         string `json:"agent,omitempty"`
+      Proc          string `json:"proc,omitempty"`
+      ClientProc    string `json:"clientProc,omitempty"`
+      Session       string `json:"session,omitempty"`
+      Source        string `json:"source,omitempty"`
+      SessionHeader string `json:"sessionHeader,omitempty"`
+      ChainSession  string `json:"chainSession,omitempty"`
+      RecordSession string `json:"recordSession,omitempty"`
+  }
+  const HeaderAttribution = "X-Cortex-Attribution"
+  func (w Wire) Encode() string // base64.RawURLEncoding of the JSON
+  ```
+
+- [ ] **Step 1: Write the failing tests.**
+  - `TestWire_EncodeGolden`: encode `Wire{Agent:"claude-code", Proc:"p1", ClientProc:"p2", Session:"s1", Source:"header", SessionHeader:"X-Claude-Code-Session-Id", ChainSession:"s0", RecordSession:"s1"}`. It must equal `testdata/wire.golden` (trimmed). `-update` rewrites the file.
+  - `TestWire_KeysMatchRust`: the JSON keys of a full Wire are exactly `agent, proc, clientProc, session, source, sessionHeader, chainSession, recordSession`.
+  - `TestWire_EncodeIsUnpadded`: no `=` in the output.
+  - Rust `decodes_go_golden` (`praxis/filters/tests/attribution_wire.rs`): read `../../core/attribution/testdata/wire.golden` from `CARGO_MANIFEST_DIR`, `decode(Some(..))`, assert every field equals the Go values.
+- [ ] **Step 2:** Run `GOWORK=off go -C core test ./attribution/ -run Wire`. Expect FAIL (undefined).
+- [ ] **Step 3:** Implement. Update the package comment "The encoder and decoder land with that hop": the encoder is `Wire.Encode`, the decoder is the gateway's.
+- [ ] **Step 4:** Run with `-update` once, then without. Run the Rust test. Both PASS.
+- [ ] **Step 5:** Commit `feat: Add attribution.Wire, the encoder of X-Cortex-Attribution`.
+
+### Task 3: The egress listener
+
+Depends on: none. Estimated lines: 300.
+
+**Files:**
+- Create: `core/listener/forwardproxy/hopheaders.go`, `core/listener/forwardproxy/egress.go`, `core/listener/forwardproxy/egress_test.go`
+
+**Interfaces:**
+- Produces (in `hopheaders.go`):
+  ```go
+  const (
+      headerDest        = "X-Cortex-Dest"
+      headerToken       = "X-Cortex-Token"
+      headerRequestID   = "X-Cortex-Request-Id"
+      headerProbe       = "X-Cortex-Probe"
+      headerEgressError = "X-Cortex-Egress-Error"
+      probeDest         = "cortex-probe" // Decision 2
+  )
+  // clientHopHeaders are the five hop headers a client must not send in praxis mode.
+  var clientHopHeaders = []string{attribution.HeaderAttribution, headerDest, headerToken, headerRequestID, headerProbe}
+  func stripHopHeaders(h http.Header)
+  func tokenMatches(given, token string) bool // constant time; empty token matches nothing
+  ```
+- Produces (in `egress.go`):
+  ```go
+  // NewEgressHandler is the praxis-mode egress listener (spec Section 4, throwaway at phase B).
+  func NewEgressHandler(token string, upstream http.RoundTripper) http.Handler
+  ```
+
+Behavior, in order:
+1. A token that does not match: 403, no dial, nothing recorded.
+2. `X-Cortex-Dest: cortex-probe`: 204, no dial.
+3. A Dest that is not `http://host:port` or `https://host:port`: 400.
+4. Else an `httputil.ReverseProxy` with `Rewrite` (set `Out.URL.Scheme`, `Out.URL.Host` from Dest, `Out.Host = In.Host`, delete Dest, Token and Egress-Error from `Out.Header`), `FlushInterval: -1`, `ModifyResponse` that deletes `X-Cortex-Egress-Error`, and an `ErrorHandler`:
+   - `*egressguard.Refusal` (errors.As): 403, `X-Cortex-Egress-Error: destination_private` (Decision 10).
+   - Else when the client is still there: 502, `X-Cortex-Egress-Error: <model.TransportError(err).Kind>`.
+   - Log with `pipeline.SanitizeLogValue` on every client-controlled value.
+
+- [ ] **Step 1: Write the failing tests** (`egress_test.go`, with an `httptest` origin):
+  - `TestEgress_TokenRequired`: no token, a wrong token, and an empty configured token each give 403. The origin sees no request.
+  - `TestEgress_ProbeAnswers204`: valid token, Dest `cortex-probe` gives 204. The origin sees nothing.
+  - `TestEgress_BadDest400`: `""`, `ftp://h:1`, `h:1`, `http://`.
+  - `TestEgress_RelaysAndStripsHeaders`: the origin gets the path, the query, the client's `Host`, `X-Cortex-Custom`, and no Dest, Token or Egress-Error.
+  - `TestEgress_GzipPassesUnchanged`: the client sends `Accept-Encoding: gzip`. The origin answers gzip. The bytes arrive still gzip, with `Content-Encoding: gzip`.
+  - `TestEgress_OriginEgressErrorRemoved`: the origin sets `X-Cortex-Egress-Error: upstream_dns`. The client does not get it.
+  - `TestEgress_DialFailure502WithKind`: Dest on a closed port gives 502 and `X-Cortex-Egress-Error: upstream_refused`.
+  - `TestEgress_GuardRefusal403`: a transport whose `DialContext` returns `&egressguard.Refusal{...}` gives 403 and `destination_private`.
+  - `TestEgress_StreamsFrames`: an SSE origin that flushes one frame, then waits on a channel. The client reads the first frame before the test releases the channel.
+- [ ] **Step 2:** `GOWORK=off go -C core test ./listener/forwardproxy/ -run Egress` FAILS.
+- [ ] **Step 3:** Implement.
+- [ ] **Step 4:** Same command PASSES. Also `-race`.
+- [ ] **Step 5:** Commit `feat: Add the praxis-mode egress listener`.
+
+### Task 4: The hop, its five branch points, `gateway_unavailable` and the probe
+
+Depends on: 2, 3. Estimated lines: 720.
+
+**Files:**
+- Create: `core/listener/forwardproxy/hop.go`, `hop_test.go`, `hop_direct_test.go`
+- Modify: `core/listener/forwardproxy/server.go` (`Server.Hop *Hop`, five branch points in `serveOutbound`)
+- Modify: `core/listener/forwardproxy/record.go` (branch point 5 in `recordOutboundResponseEvent`)
+- Modify: `core/model/session.go` (`ErrorKindGatewayUnavailable = "gateway_unavailable"`, doc comment), `core/pipeline/model_alias.go` (alias)
+
+**Interfaces:**
+- Consumes: Task 2 `attribution.Wire`, `HeaderAttribution`. Task 3 header constants, `stripHopHeaders`, `probeDest`.
+- Produces:
+  ```go
+  type Hop struct { /* gateway addr, token, client *http.Client, ready atomic.Bool, owned sync.Map */ }
+  func NewHop(gatewayAddr, token string) *Hop
+  func (h *Hop) Ready() bool
+  // Probe sends a probe every `every` until one gets 204 or ctx ends. Logs once when ready.
+  func (h *Hop) Probe(ctx context.Context, every time.Duration)
+  // Server field:
+  Hop *Hop // nil in direct mode
+  // Internal, used by Task 6:
+  func (h *Hop) own(pctx *pipeline.Context, w attribution.Wire)   // branch point 2
+  func (h *Hop) release(pctx *pipeline.Context)                    // deferred at branch point 2
+  func (h *Hop) owns(pctx *pipeline.Context) (attribution.Wire, bool)
+  func (s *Server) hopDo(w http.ResponseWriter, r *http.Request, pctx *pipeline.Context, tl *tunnelLog, sid string) (resp *http.Response, err error, answered bool)
+  ```
+
+The hop client: `http.Transport{Proxy: nil, ForceAttemptHTTP2: false, MaxIdleConnsPerHost: 100, IdleConnTimeout: 90 * time.Second, DialContext: (&net.Dialer{Timeout: 2 * time.Second}).DialContext}`, `CheckRedirect` returns `http.ErrUseLastResponse`.
+
+The five branch points in `serveOutbound`, in this order:
+1. Right after `attribution.Strip(r.Header)` in `serveOutbound` (`server.go`, the call near the top of the function; not the one in `handleConnect`): `if s.Hop != nil { stripHopHeaders(r.Header) }`.
+2. At the request row (`if !skipped && s.Sessions != nil` block): `hopped := s.Hop != nil && !skipped && !forwardUpgrade`. When `hopped`, call `s.Hop.own(pctx, wire)` and `defer s.Hop.release(pctx)` in place of `s.recordCapturedRequest`. `defer s.touch` stays (spec: keep `touch`). Build `wire` here (Decision 8).
+3. `ownEncoding := hopped || (!skipped && (...))`.
+4. At `client.Do(r)`: `if hopped { resp, err, answered = s.hopDo(w, r, pctx, tl, sid); if answered { return } } else { resp, err = client.Do(r) }`. `hopDo` runs after the redirect re-check, so the order is the spec's.
+5. In `recordOutboundResponseEvent`, first line after the `s.Sessions == nil` check: `if s.Hop != nil { if _, ok := s.Hop.owns(pctx); ok { return } }`. Task 6 adds the cut note here.
+
+`hopDo`:
+- When `!s.Hop.Ready()`: answer 503 `gateway_unavailable` (below) and return `answered = true`.
+- Set `X-Cortex-Dest` (Decision 1, from `r.URL` before the swap), `X-Cortex-Token`, `X-Cortex-Request-Id: pctx.RequestID()`, `X-Cortex-Attribution: wire.Encode()`. Then `r.URL.Scheme, r.URL.Host = "http", gatewayAddr`.
+- `resp, err := h.client.Do(r)`. A dial failure (`*net.OpError` with `Op == "dial"`, via errors.As) answers 503 `gateway_unavailable`, `answered = true`. Any other result returns to the caller.
+- The 503 path: `s.Hop.release(pctx)` first, so branch point 5 records. Then `s.recordCapturedRequest(tl, pctx, sid, nil)`, `http.Error(w, `{"error":"gateway unavailable"}`, 503)`, and `s.recordOutboundResponseEvent(pctx, 503, &pipeline.EventError{Kind: pipeline.ErrorKindGatewayUnavailable, Message: "cortex-gateway is unavailable: " + cause})`. `cause` is `"not ready"` or `model.TransportError(err).Message`.
+
+`Probe`: `GET http://<gateway>/` with `X-Cortex-Token`, `X-Cortex-Probe: 1`, `X-Cortex-Dest: cortex-probe`, 2 s timeout per try. Set ready on 204. Log at Info once when ready, and at Warn at most every 30 s while not ready.
+
+- [ ] **Step 1: Write the failing tests** (`hop_test.go`). A fake gateway is an `httptest.Server` that records each request and answers as the test says. Build the Server as the existing tests do (`support_test.go` helpers), with an empty pipeline and a store.
+  - `TestHop_DialsGatewayKeepsHostAndPath`: a plain proxied `GET http://api.example.com/v1/x?q=1`. The gateway sees path `/v1/x?q=1`, `Host: api.example.com`, `X-Cortex-Dest: http://api.example.com:80`, the token, a non-empty request id, and an attribution header that decodes to the request's session.
+  - `TestHop_OnlyTheDialTargetMoves` (rewrite of the spike test): a plugin that writes `pctx.Scheme`, `pctx.Host` and a `X-Cortex-Dest` header in `pctx.Headers`. The gateway still sees the hop's Dest, and the real destination stays in Dest.
+  - `TestHop_ClientHopHeadersNeverReachGateway`: the client sends all five with forged values, plus `X-Cortex-Custom: v`. The gateway sees only the hop's values for Dest, Token, Request-Id and Attribution, no `X-Cortex-Probe`, and `X-Cortex-Custom: v`.
+  - `TestHop_OwnsEncoding`: the client sends `Accept-Encoding: gzip, deflate, br`. The gateway sees `Accept-Encoding: gzip` only. A gzip answer reaches the client decoded.
+  - `TestHop_NoRowsForHoppedRequest`: after a 200 and after an SSE stream, the store holds no row for the request.
+  - `TestHop_TouchStillRuns`: with process attribution on (the existing test helper for a chain), the session is touched.
+  - `TestHop_NotReadyIsGatewayUnavailable`: before any probe: 503 to the client, the gateway sees nothing, the store has the request row and a response row with status 503 and kind `gateway_unavailable`.
+  - `TestHop_DialFailureIsGatewayUnavailable`: ready, gateway on a closed port: the same two rows.
+  - `TestHop_ProbeSetsReady`: the fake gateway answers 503 twice, then 204. `Probe` returns, `Ready()` is true, and each probe carried the token, `X-Cortex-Probe` and `X-Cortex-Dest: cortex-probe`.
+  - `TestHop_SkipHostsAndUpgradesDoNotHop`: a `skip_hosts` match and a forwarded WebSocket upgrade dial the origin, not the gateway.
+- [ ] **Step 2: Write the direct-mode test** (`hop_direct_test.go`): `TestServeOutbound_DirectModeBranchPoints`. `Hop` nil, inference-parser-like test plugin that reads bodies (use the existing test plugin helpers). One request each for JSON and SSE. Assert: the request row and its capture exist, the response row and its capture exist, a client `X-Cortex-Dest`, `-Token`, `-Attribution`, `-Request-Id`, `-Probe` reach the origin unchanged, `Accept-Encoding` at the origin is as before (Go's `gzip`), and the origin (not a gateway) got the dial.
+- [ ] **Step 3:** `GOWORK=off go -C core test ./listener/forwardproxy/ -run 'Hop|DirectModeBranchPoints'` FAILS.
+- [ ] **Step 4:** Implement. Add the error kind to `docs/session-api.md`'s error-kind list in Task 7, not here.
+- [ ] **Step 5:** `GOWORK=off go -C core test -race ./listener/forwardproxy/ ./model/ ./pipeline/` PASSES. The whole existing forwardproxy suite passes with no change.
+- [ ] **Step 6:** Commit `feat: Hop praxis-mode requests to cortex-gateway at the dial`.
+
+### Task 5: Ingestion uses the hop's joins
+
+Depends on: none. Estimated lines: 230.
+
+**Files:**
+- Modify: `core/sessionapi/ingest.go`
+- Create: `core/sessionapi/ingest_joins_test.go`
+
+**Interfaces:**
+- Produces:
+  ```go
+  // HopJoins is the forward proxy's join state in praxis mode (spec Section 1, "Joins").
+  type HopJoins interface {
+      // AppendRequest writes a request row through its bridged tunnel's pairing path and
+      // reports whether it did. False: the caller appends the row.
+      AppendRequest(requestID, sid string, ev model.SessionEvent, c *model.RequestCapture) bool
+      // Release drops the join of requestID with no write.
+      Release(requestID string)
+      // TakeCut returns the error of a cut that Cortex made on requestID, or nil.
+      TakeCut(requestID string) *model.EventError
+  }
+  // Ingest gets: Joins HopJoins // nil: no joins
+  ```
+
+`fileIngest` changes:
+- The `erased` branch: `if j := in.cfg.Joins; j != nil { j.Release(l.RequestID) }` before `refuse`. Remove the `#1721` comment.
+- A response row: `if cut := joins.TakeCut(id); cut != nil` then set `ev.Error.Kind, ev.Error.Message = cut.Kind, cut.Message` when `ev.Error != nil`, else `ev.Error = cut`. Before `settleIngested`.
+- A request row (`ev.Phase == model.SessionRequest`): keep the `holdRequest` call as it is. Then, in place of the last two lines (`s.store.AppendCapture(sid, ev, in.capture(l, &ev))` and `in.lines.Add(1)`): `c := in.capture(l, &ev)`; when `joins != nil && joins.AppendRequest(l.RequestID, sid, ev, c)`, only `in.lines.Add(1)` and return. Else `s.store.AppendCapture(sid, ev, c)` and `in.lines.Add(1)` as now. Response rows always go to `AppendCapture`.
+- `joins` is `in.cfg.Joins`. Add the field `Joins HopJoins` to the `Ingest` struct with the doc comment `// Joins is the forward proxy's join state in praxis mode. nil: no joins.`
+
+- [ ] **Step 1: Write the failing tests** with a fake `HopJoins` that records calls:
+  - `TestIngestJoins_ErasedLineReleasesJoin`: erase the session, then send a line with an earlier `startedAt`. `Release(id)` is called once, nothing is appended, `erased` is counted.
+  - `TestIngestJoins_RequestRowGoesThroughJoin`: the fake returns true. The store gets no row from ingestion, and `ingest_lines_total` is 1.
+  - `TestIngestJoins_RequestRowFallsBack`: the fake returns false. The row is appended as today.
+  - `TestIngestJoins_CutReplacesClientCanceled`: a response row with `client_canceled`. The fake cut is `proxy_error` with a message. The stored row has kind `proxy_error` and that message, and keeps its `code`.
+  - `TestIngestJoins_NilJoinsUnchanged`: the existing behavior with `Joins == nil`.
+- [ ] **Step 2:** `GOWORK=off go -C core test ./sessionapi/ -run IngestJoins` FAILS.
+- [ ] **Step 3:** Implement.
+- [ ] **Step 4:** PASS, and the whole `./sessionapi/` suite passes.
+- [ ] **Step 5:** Commit `feat: Join ingested rows to Cortex's tunnels and cuts`.
+
+### Task 6: The join registry and the tunnel hold
+
+Depends on: 4. Estimated lines: 450.
+
+**Files:**
+- Create: `core/listener/forwardproxy/hopjoins.go`, `hopjoins_test.go`
+- Modify: `core/listener/forwardproxy/tunnel.go` (`awaiting int` on `tunnelLog`, `settleLocked` waits while `awaiting > 0`, new methods below)
+- Modify: `core/listener/forwardproxy/hop.go` (register a join in `hopDo`, the cut note at branch point 5, `Hop` methods that implement `HopJoins`)
+
+**Interfaces:**
+- Consumes: Task 4 `Hop`, `hopDo`, `owns`.
+- Produces (method set of `*Hop`, matching Task 5's `HopJoins` exactly):
+  ```go
+  func (h *Hop) AppendRequest(requestID, sid string, ev model.SessionEvent, c *model.RequestCapture) bool
+  func (h *Hop) Release(requestID string)
+  func (h *Hop) TakeCut(requestID string) *model.EventError
+  const hopJoinTTL = 60 * time.Second
+  // on tunnelLog:
+  func (t *tunnelLog) await()                // awaiting++
+  func (t *tunnelLog) joined(sid string, ev pipeline.SessionEvent, c *pipeline.RequestCapture) bool // recordWith, awaiting--, settleLocked
+  func (t *tunnelLog) unawait(answered bool) // awaiting--; when answered, t.answered = true; settleLocked
+  ```
+
+Rules:
+- `hopDo` registers `requestID → {tl, timer}` for a bridged request (`tl != nil`) before it sends, and calls `tl.await()`. It does not mark the tunnel answered.
+- The 503 path of `hopDo` records its rows first, then drops the join with `unawait(false)`.
+- `AppendRequest`: remove the entry and stop its timer. No entry, or `tl == nil`: return false. Else return `tl.joined(...)`. When `joined` returns false (the open row was already written), the caller appends.
+- `Release`: remove the entry, `tl.unawait(true)` (Decision 13).
+- Expiry (`time.AfterFunc(h.ttl, ...)`): remove the entry, `tl.unawait(false)` (Decision 14).
+- Branch point 5: when the hop owns `pctx` and `fail != nil && fail.Kind == "proxy_error"`, keep `requestID → copy of fail` in `cuts` with its time. `TakeCut` removes and returns it. Entries older than `h.ttl` are swept on each insert.
+- `h.ttl` is a field set to `hopJoinTTL` by `NewHop`, so tests can set it short.
+
+- [ ] **Step 1: Write the failing tests** (`hopjoins_test.go`). Use the existing bridged-tunnel test setup (`tlsbridge_integration_test.go` / `bridgedopen_test.go` helpers) with a fake gateway that answers 200 and a test that calls `hop.AppendRequest` as ingestion would.
+  - `TestHopJoin_TunnelClosedBeforeRowGetsOnePair`: the client closes the tunnel after its answer. No row is in the store. Then `AppendRequest` with the request row. The store holds exactly the open row and the request row, in that order, and no close row.
+  - `TestHopJoin_ExpiredJoinSettlesAsToday`: ttl 50 ms, no `AppendRequest`. After the ttl the store holds the open row and a "carried nothing" close row.
+  - `TestHopJoin_ReleaseWritesNothing`: `Release` after the tunnel closed. The store holds no row for that tunnel, also after twice the ttl.
+  - `TestHopJoin_SecondRequestAppendsItself`: two requests on one tunnel. The first `AppendRequest` returns true, the second returns false.
+  - `TestHopJoin_PlainRequestHasNoPairJoin`: a plain proxied request: `AppendRequest` returns false.
+  - `TestHopJoin_IdleCutIsTaken`: a hopped SSE stream cut by the idle cutoff (set `streamReadIdleTimeout` short the way `maxidle_test.go` does). `TakeCut(id)` returns kind `proxy_error` once, then nil.
+  - `TestHopJoin_GatewayUnavailablePairsAndReleases`: not ready, bridged. The open row, the request row and the 503 row are written once, and no close row follows.
+- [ ] **Step 2:** `GOWORK=off go -C core test ./listener/forwardproxy/ -run HopJoin` FAILS.
+- [ ] **Step 3:** Implement.
+- [ ] **Step 4:** `GOWORK=off go -C core test -race ./listener/forwardproxy/` PASSES, all existing tunnel tests included.
+- [ ] **Step 5:** Commit `feat: Hold a bridged tunnel until its gateway row joins`.
+
+### Task 7: Turn on the hop and the egress listener in praxis mode
+
+Depends on: 3, 4, 5, 6. Estimated lines: 370.
+
+**Files:**
+- Create: `cmd/cortex/praxis.go`, `cmd/cortex/praxis_test.go`
+- Modify: `cmd/cortex/ingest.go` (`readGatewayToken`, `ingestOptions` takes `joins sessionapi.HopJoins`), `cmd/cortex/ingest_test.go`
+- Modify: `cmd/cortex/main.go` (`buildPipelines`, the forward-role block, the session API options, `upOpts` moves out of the TLS bridge block)
+- Modify: `cmd/cortex/gateway.go` (the comment "Until the egress listener lands (#1721)")
+- Modify: `core/config/listenerplan.go`, `core/config/listenerplan_test.go`, `core/config/config.go` (`PraxisConfig` doc comment)
+- Modify: `docs/session-api.md` (`gateway_unavailable` in the error kinds, the joins under "What Cortex does with a line"), `praxis/README.md` (a "Hop decisions (#1721)" list: Decisions 1-17 in one line each)
+
+**Interfaces:**
+- Consumes: `forwardproxy.NewHop`, `Hop.Probe`, `forwardproxy.NewEgressHandler`, `sessionapi.HopJoins`, `Ingest.Joins`.
+- Produces:
+  ```go
+  // outboundEntries is the outbound plugin list to build: none in praxis mode, and it logs the skipped names.
+  func outboundEntries(c *config.Config, log *slog.Logger) []config.PluginEntry
+  // readGatewayToken reads the token file that CORTEX_GATEWAY_TOKEN_FILE names.
+  func readGatewayToken(getenv func(string) string, readFile func(string) ([]byte, error)) (string, error)
+  // startPraxis sets fpSrv.Hop and decline mode, serves the egress listener on ln with upstream,
+  // starts the probe, and returns the hop for Ingest.Joins.
+  func startPraxis(ctx context.Context, cfg *config.Config, fpSrv *forwardproxy.Server, token string, upstream http.RoundTripper, ln net.Listener) *forwardproxy.Hop
+  var _ sessionapi.HopJoins = (*forwardproxy.Hop)(nil)
+  ```
+
+Rules:
+- `buildPipelines` is the closure in `cmd/cortex/main.go`. Change its call `plugins.BuildWithDeps(c.Pipeline.Outbound.Plugins, outDeps)` to `plugins.BuildWithDeps(outboundEntries(c, slog.Default()), outDeps)`. Leave `spiffeProviderNeeded` as it is. The log line: `slog.Info("praxis mode: outbound plugins not built; cortex-gateway processes the traffic", "plugins", names)`, only when the list is not empty. Startup and each reload go through it. The config keeps the plugins.
+- `ListenerPlan`: in the forward role, when `cfg.PraxisEnabled()`, add `PlannedListener{Name: "praxis-egress", Addr: cfg.Praxis.EgressAddr()}`.
+- `startPraxis` sets `fpSrv.ResponsesWebSockets = config.ResponsesWebSocketsDecline`. Call it before the forward proxy serves (the same rule as `startVMControl`). The egress server goes through `bootstrap.ServeHTTPServer("praxis-egress", ...)` into `httpServers`, so shutdown closes it.
+- `upstream` is Decision 11.
+- `mtls` and `bind_loopback_only` refusals exist (`validatePraxis`). Do not add them again.
+
+- [ ] **Step 1: Write the failing tests.**
+  - `TestOutboundEntries_PraxisSkipsPlugins`: a config with two outbound plugins and `praxis.enabled`. The result is empty, and the log (a `slog` handler into a buffer) names both. With `praxis.enabled: false` the same two come back, in order. The config's list does not change.
+  - `TestOutboundEntries_PraxisEmptyListLogsNothing`.
+  - `TestListenerPlan_PraxisEgress` (core/config): praxis on adds `praxis-egress` at `127.0.0.1:47606`. Praxis off, or no forward role, does not.
+  - `TestStartPraxis_DeclinesResponsesWebSocket`: after `startPraxis`, `fpSrv.ResponsesWebSockets == decline` and `fpSrv.Hop != nil`.
+  - `TestStartPraxis_EgressServesProbe`: a probe from the hop through a fake gateway (an `httputil.ReverseProxy` to `ln`) makes `Hop.Ready()` true within 3 s.
+  - `TestReadGatewayToken_*`: the three errors that `ingestOptions` has today move here, with the same messages.
+- [ ] **Step 2:** Run (gate step for `cmd/cortex` and `core/config`). FAIL.
+- [ ] **Step 3:** Implement. Update the comments that name #1721 as future work (`config.go`, `gateway.go`, `attribution.rs` was Task 1).
+- [ ] **Step 4:** `TAGS=$(GOWORK=off go -C scripts/profile-tags run . local) && GOWORK=off go -C cmd/cortex test -tags "$TAGS" ./... && GOWORK=off go -C core test ./config/ ./sessionapi/` PASSES.
+- [ ] **Step 5:** Commit `feat: Turn on the hop and the egress listener in praxis mode`.
+
+### Task 8: End-to-end tests through the real gateway
+
+Depends on: 1, 7. Estimated lines: 850.
+
+**Files:**
+- Create: `cmd/cortex/praxis_e2e_test.go` (`//go:build praxis_e2e`)
+- Modify: `.github/workflows/praxis.yaml`
+
+**Harness** (`newPraxisChain(t, opts)`):
+- Skip unless `CORTEX_GATEWAY_BIN` names a file.
+- A mock origin (`httptest.NewServer`) that records each request's headers and serves the handlers below.
+- A `session.Store`, a `sessionapi.Server` with `WithIngest` (token, rates, `Joins: hop`) on a loopback listener, and a forward proxy built as `main` builds it with an empty outbound chain.
+- `startPraxis` with an `upstream` transport whose `DialContext` sends every address to the mock origin (Decision 15).
+- `gatewayYAML(cfg, tokenPath)` written to a temp dir. Start `CORTEX_GATEWAY_BIN -c <yaml> --exit-on-stdin-eof` with an empty env and a stdin pipe. `t.Cleanup` closes the pipe and waits.
+- A direct-mode twin (`newDirectChain`): the same mock origin and dialer, praxis off, outbound `inference-parser` from the local profile, so the two runs can be compared.
+- The client: `http.Client` with `Proxy` = the forward proxy, plain `http://` URLs.
+
+- [ ] **Step 1: Write the tests.** Each one names the W1 item or the issue's "Done when" line.
+  - `TestPraxisE2E_HostAndCostMatchDirect` (W1 1): one chat-completions JSON request with `usage`. The response row's `host` and `plugins.cost` equal the direct run's.
+  - `TestPraxisE2E_GzipSSEArrivesAsFrames` (W1 3): the origin sends gzip SSE, one frame, then waits. The client reads frame 1 before the test releases frame 2.
+  - `TestPraxisE2E_HangUpIsClientCanceled` (W1 5): the client cancels mid-stream. The response row's kind is `client_canceled`, and the origin's request context ends within 5 s.
+  - `TestPraxisE2E_FailClosedAtStartup` (W1 7): start the chain without the gateway. A request gets 503 and a `gateway_unavailable` row. Start the gateway. Within 10 s a request gets 200.
+  - `TestPraxisE2E_SilenceMidStream` (W1 8): the origin sends one frame, is silent for 65 s, then sends the rest. The client gets every frame and the row has no error.
+  - `TestPraxisE2E_OverTenMiBEachWay` (W1 9): an 11 MiB request body (not inference) and an 11 MiB streamed response. `bytesUp` and `bytesDown` are the sizes.
+  - `TestPraxisE2E_CodexMislabeledSSE`: host `chatgpt.com`, path `/backend-api/codex/responses`, `Content-Type: application/json`, 2xx SSE body. The client reads frame 1 before frame 2 is released. A 405 with `{"detail":"Method Not Allowed"}`: the client gets that body, and the row carries the error detail.
+  - `TestPraxisE2E_AcceptEncodingDecoded`: the client sends `Accept-Encoding: gzip, deflate, br`. The origin sees only `gzip` and answers gzip JSON. The client gets plain JSON, and the row has the tokens.
+  - `TestPraxisE2E_NoHopHeaderReachesOrigin`: the client sends the five hop headers and `X-Cortex-Custom`. The origin sees none of the twelve hop headers and sees `X-Cortex-Custom`.
+  - `TestPraxisE2E_TokenlessRequestTo47605`: a request straight to `praxis.listen` with no token gets 403. `/v1/sessions` shows no new row.
+  - `TestPraxisE2E_SameEventsAsDirect`: the JSON and SSE requests through both chains. Compare `/v1/sessions/{id}` events field by field after removing `seq`, `at`, `sessionId`, `requestId`. `ttfbMs` and `durationMs` agree within 50 ms. Any other difference fails the test. Do not widen the list in the test: fix the cause, or add it to the README's known differences with its reason and a pinned assertion.
+- [ ] **Step 2:** Build the gateway and run:
+  `(cd praxis && cargo build --locked -p cortex-gateway) && TAGS=$(GOWORK=off go -C scripts/profile-tags run . local) && CORTEX_GATEWAY_BIN=$PWD/praxis/target/debug/cortex-gateway GOWORK=off go -C cmd/cortex test -count=1 -timeout 15m -tags "praxis_e2e,$TAGS" -run TestPraxisE2E .`
+  Fix what fails in the task that owns the code, in this PR.
+- [ ] **Step 3: CI.** In `praxis.yaml`, after `cargo test`, add `actions/setup-go@b7ad1dad31e06c5925ef5d2fc7ad053ef454303e  # v7.0.0` (`go-version-file: cmd/cortex/go.mod`) and a step `End-to-end (praxis mode)` with the Step 2 command and `continue-on-error: ${{ steps.changes.outputs.blocking != 'true' }}`. Give the step `id: e2e` and no `working-directory`, so it runs from the repo root as the command expects. Add its outcome to the report step as the other three are (`E2E: ${{ steps.e2e.outcome }}` in `env`, and a `::warning::` line). Add to both `paths` lists: `core/listener/forwardproxy/hop*.go`, `core/listener/forwardproxy/egress*.go`, `cmd/cortex/praxis*.go`, `cmd/cortex/gateway.go`.
+- [ ] **Step 4:** Commit `test: Run the praxis-mode end-to-end checks through cortex-gateway`.
+
+---
+
+## Gate
+
+From the worktree root:
+
+```sh
+test -z "$(gofmt -l core cmd/cortex)" && GOWORK=off go -C core vet ./... && GOWORK=off go -C core test -race ./attribution/... ./listener/... ./sessionapi/... ./config/... ./model/... ./pipeline/... && TAGS=$(GOWORK=off go -C scripts/profile-tags run . local) && GOWORK=off go -C cmd/cortex vet -tags "$TAGS" ./... && GOWORK=off go -C cmd/cortex test -tags "$TAGS" ./... && (cd praxis && cargo fmt --all --check && cargo clippy --all-targets --locked -- -D warnings && cargo test --locked && cargo build --locked -p cortex-gateway) && CORTEX_GATEWAY_BIN=$PWD/praxis/target/debug/cortex-gateway GOWORK=off go -C cmd/cortex test -count=1 -timeout 15m -tags "praxis_e2e,$TAGS" -run TestPraxisE2E .
+```
+
+Before P1 and P2 there is no `praxis_e2e` file, so the last step runs no test and passes.
+
+## Self-review notes
+
+- Spec coverage: the hop (Task 4), the entry strip (Task 4, point 1), `ownEncoding` (Task 4, point 3), the egress listener (Task 3), joins (Tasks 5, 6), the probe and `gateway_unavailable` (Task 4), config skip and decline (Task 7), the tunnel hold (Task 6), the end-to-end list (Task 8), the direct-mode branch test (Task 4), the gateway's kept headers (Task 1).
+- Out of scope: agentop and install (step 5), MCP and A2A (step 6), the release job (step 7), phase B absolute form.
diff --git a/praxis/README.md b/praxis/README.md
index eac8e854f..53523f298 100644
--- a/praxis/README.md
+++ b/praxis/README.md
@@ -43,7 +43,10 @@ The binary registers two filters. Put them first in the chain, in this order,
 and add `access_log` after them:
 
 - `cortex_attribution` (`token_file`): refuses a request without the token in
-  `X-Cortex-Token` with 403, and removes the hop headers before the upstream.
+  `X-Cortex-Token` with 403. Before the upstream, it removes every hop header
+  but `X-Cortex-Dest` and `X-Cortex-Token`, which the Cortex egress listener
+  reads and then removes. So the upstream must be the egress listener, never an
+  origin (decision 20).
 - `cortex_observe` (`token_file`, `ingest` as `host:port`): records the request
   and response rows and streams them to `POST /v1/ingest`.
 
@@ -78,7 +81,7 @@ gives the full text.
 17. The ingest stream ends on Cortex shutdown, so restarts stay quick.
 18. A stream cut after the response head is `client_canceled`, also when the upstream cut it. Praxis 0.8.0 gives no error source after the head.
 19. A response row leaves out the request-only inference fields (`messages`, `tools`, `toolResults`, `historyToolCalls`, `historyThinking`). Cortex puts them back from the request row (`docs/session-api.md`, line format v1).
-20. The gateway removes `X-Cortex-Dest` and `X-Cortex-Token` before the upstream, against the issue text. No egress listener exists yet, so they would send the token to the origin. #1721 keeps them again for its egress listener.
+20. The gateway keeps `X-Cortex-Dest` and `X-Cortex-Token` for its upstream, which is always the Cortex egress listener (#1721). The egress listener removes both before it dials the origin. The gateway must remove them before any other upstream. The request capture never has them.
 21. A buffered response body that goes over the 10 MiB cap after the response head gives a row with the upstream status, not 502. The client has that status and a cut body. Direct mode reads the body before it sends the head, so its client and its row get 502. A `Content-Length` over the cap gives 502 in both modes.
 
 ## Accepted edge cases (#1719)
diff --git a/praxis/filters/src/filters/attribution.rs b/praxis/filters/src/filters/attribution.rs
index 1ac6c4c0d..9bb61bc2e 100644
--- a/praxis/filters/src/filters/attribution.rs
+++ b/praxis/filters/src/filters/attribution.rs
@@ -10,7 +10,7 @@ use praxis_filter::{FilterAction, FilterError, HttpFilter, HttpFilterContext, Re
 use serde::Deserialize;
 
 use crate::record::attribution::{
-    ATTRIBUTION_HEADER, Attribution, HOP_HEADERS, TOKEN_HEADER, decode, token_matches,
+    ATTRIBUTION_HEADER, Attribution, TOKEN_HEADER, decode, token_matches, upstream_strip_headers,
 };
 
 /// Set on a request with a valid token, unless it is a probe. `cortex_observe`
@@ -77,9 +77,7 @@ impl HttpFilter for CortexAttribution {
             ctx.extensions.insert(TokenValid(attribution));
         }
         ctx.request_headers_to_remove.extend(
-            HOP_HEADERS
-                .iter()
-                .filter_map(|n| HeaderName::from_bytes(n.as_bytes()).ok()),
+            upstream_strip_headers().filter_map(|n| HeaderName::from_bytes(n.as_bytes()).ok()),
         );
         Ok(FilterAction::Continue)
     }
diff --git a/praxis/filters/src/record/attribution.rs b/praxis/filters/src/record/attribution.rs
index 1211cbc76..61fb36753 100644
--- a/praxis/filters/src/record/attribution.rs
+++ b/praxis/filters/src/record/attribution.rs
@@ -22,11 +22,11 @@ pub struct Attribution {
 
 pub const ATTRIBUTION_HEADER: &str = "X-Cortex-Attribution";
 pub const TOKEN_HEADER: &str = "X-Cortex-Token";
+pub const DEST_HEADER: &str = "X-Cortex-Dest";
 
-/// The hop headers. The upstream and the request capture get none of them.
-/// `X-Cortex-Token` is the gateway's secret, so it must not reach an origin.
-/// #1721 keeps `X-Cortex-Dest` and the token for its egress listener, which
-/// removes them itself.
+/// The hop headers. The request capture gets none of them: `X-Cortex-Token` is
+/// the gateway's secret. The upstream gets only those that
+/// [`upstream_strip_headers`] does not give.
 pub const HOP_HEADERS: [&str; 12] = [
     "X-Cortex-Agent",
     "X-Cortex-Proc",
@@ -38,10 +38,20 @@ pub const HOP_HEADERS: [&str; 12] = [
     ATTRIBUTION_HEADER,
     "X-Cortex-Request-Id",
     "X-Cortex-Probe",
-    "X-Cortex-Dest",
+    DEST_HEADER,
     TOKEN_HEADER,
 ];
 
+/// The hop headers that the gateway removes before its upstream: all but
+/// `X-Cortex-Dest` and `X-Cortex-Token`. The interim upstream is always the
+/// Cortex egress listener (#1721), which removes those two itself. At phase B,
+/// when the gateway dials origins, it removes all of [`HOP_HEADERS`].
+pub fn upstream_strip_headers() -> impl Iterator<Item = &'static str> {
+    HOP_HEADERS
+        .into_iter()
+        .filter(|n| *n != DEST_HEADER && *n != TOKEN_HEADER)
+}
+
 /// Decodes the header. A missing header, bad base64url or bad JSON gives the default.
 pub fn decode(header: Option<&str>) -> Attribution {
     header
diff --git a/praxis/filters/src/record/row.rs b/praxis/filters/src/record/row.rs
index d26035f28..30329ec10 100644
--- a/praxis/filters/src/record/row.rs
+++ b/praxis/filters/src/record/row.rs
@@ -69,6 +69,7 @@ fn egress_kind(v: &str) -> &'static str {
         "upstream_tls" => "upstream_tls",
         "upstream_timeout" => "upstream_timeout",
         "upstream_refused" => "upstream_refused",
+        "destination_private" => "destination_private",
         _ => "upstream_error",
     }
 }
@@ -854,6 +855,12 @@ mod tests {
         }
     }
 
+    #[test]
+    fn destination_private_egress_error_keeps_its_kind() {
+        let (r, _) = egress("destination_private");
+        assert_eq!(r["event"]["error"]["kind"], "destination_private");
+    }
+
     #[test]
     fn unknown_egress_error_is_upstream_error() {
         let (r, _) = egress("something_else");
diff --git a/praxis/filters/tests/attribution_wire.rs b/praxis/filters/tests/attribution_wire.rs
new file mode 100644
index 000000000..1d5753b31
--- /dev/null
+++ b/praxis/filters/tests/attribution_wire.rs
@@ -0,0 +1,24 @@
+//! Parity with the Go encoder: core/attribution/testdata/wire.golden is
+//! attribution.Wire.Encode of the values below (TestWire_EncodeGolden).
+
+use std::path::Path;
+
+use cortex_praxis_filters::record::attribution::{Attribution, decode};
+
+#[test]
+fn decodes_go_golden() {
+    let path =
+        Path::new(env!("CARGO_MANIFEST_DIR")).join("../../core/attribution/testdata/wire.golden");
+    let golden = std::fs::read_to_string(&path).expect("read wire.golden");
+    let want = Attribution {
+        agent: "claude-code".into(),
+        proc: "p1".into(),
+        client_proc: "p2".into(),
+        session: "s1".into(),
+        source: "header".into(),
+        session_header: "X-Claude-Code-Session-Id".into(),
+        chain_session: "s0".into(),
+        record_session: "s1".into(),
+    };
+    assert_eq!(decode(Some(&golden)), want);
+}
diff --git a/praxis/server/tests/gateway.rs b/praxis/server/tests/gateway.rs
index d483a780c..6a236c526 100644
--- a/praxis/server/tests/gateway.rs
+++ b/praxis/server/tests/gateway.rs
@@ -546,12 +546,22 @@ fn hop_headers_removed_custom_kept() {
         ("X-Cortex-Chain-Session", "cs"),
         ("X-Cortex-Attribution", "e30"),
         ("X-Cortex-Request-Id", "rid-1"),
-        ("X-Cortex-Dest", "api.example.com:443"),
+        ("X-Cortex-Dest", "https://api.example.com:443"),
         ("X-Cortex-Token", TOKEN),
         ("X-Cortex-Custom", "kept"),
     ];
-    let (h, _) = send(&g, "GET", "/v1/models", &hop, b"");
-    assert_eq!(status(&h), 200);
+    // An inference request, so that the request line has a capture.
+    let mut h = fixture_headers();
+    h.retain(|(n, _)| !n.eq_ignore_ascii_case("x-cortex-token"));
+    h.extend(hop.iter().map(|(n, v)| (n.to_string(), v.to_string())));
+    let (rh, _) = send(
+        &g,
+        "POST",
+        "/v1/messages",
+        &as_refs(&h),
+        &fixture("request.json"),
+    );
+    assert_eq!(status(&rh), 200);
     let seen = o
         .seen
         .recv_timeout(WAIT)
@@ -565,15 +575,39 @@ fn hop_headers_removed_custom_kept() {
         .filter(|n| n.starts_with("x-cortex-"))
         .collect();
     cortex.sort();
-    // The token is the gateway's secret: no hop header reaches the origin.
-    assert_eq!(cortex, ["x-cortex-custom"]);
+    // The interim upstream is the egress listener: it gets Dest and the token, and removes them.
+    assert_eq!(
+        cortex,
+        ["x-cortex-custom", "x-cortex-dest", "x-cortex-token"]
+    );
     assert_eq!(header(&seen.head, "x-cortex-custom"), Some("kept"));
-    for l in i.take(2) {
+    assert_eq!(
+        header(&seen.head, "x-cortex-dest"),
+        Some("https://api.example.com:443")
+    );
+    assert_eq!(header(&seen.head, "x-cortex-token"), Some(TOKEN));
+    let lines = i.take(2);
+    for l in &lines {
         assert_eq!(
             l["requestId"], "rid-1",
             "the id comes from X-Cortex-Request-Id"
         );
     }
+    // The token must never be in a capture: the request capture has none of the twelve.
+    let req = lines
+        .iter()
+        .find(|l| l["event"]["phase"] == "request")
+        .expect("a request line");
+    let hdr = req["capture"]["header"]
+        .as_object()
+        .expect("the request line has a capture header");
+    assert!(hdr.contains_key("X-Cortex-Custom"), "{hdr:?}");
+    for (n, _) in hop.iter().filter(|(n, _)| *n != "X-Cortex-Custom") {
+        assert!(
+            !hdr.keys().any(|k| k.eq_ignore_ascii_case(n)),
+            "{n} is in the capture"
+        );
+    }
 }
 
 #[test]