feat(logging): log Codex media forwarding start with detailed fields

- Implemented logging for Codex remote media forwarding start events, including detailed connection and credential metadata. - Added `formatLogFieldValue` for quoting specific log fields and ensured newline safety in log output. - Enhanced unit tests to validate log content, escaping, and field inclusion.
2026-09-03 06:35:00 +08:00 · 2026-07-25 07:51:43 +08:00
parent 49be36aef6
commit f8dffa0522
8 changed files with 401 additions and 37 deletions
--- a/internal/client/codex/live/live.go
+++ b/internal/client/codex/live/live.go
@@ -11,6 +11,7 @@ import (
 	"mime"
 	"mime/multipart"
 	"net/http"
+	"path/filepath"
 	"reflect"
 	"strings"
 	"sync"
@@ -222,7 +223,8 @@ func (h *Handler) Handle(c *gin.Context) {
 		ctx = attemptCtx
 		defer releaseAttempt()
 	}
-	logging.SetGinCPATraceID(c, selected.EnsureIndex())
+	selectedIndex := selected.EnsureIndex()
+	logging.SetGinCPATraceID(c, selectedIndex)
 	if selection != nil {
 		defer func() {
 			if selection.Active() && !selection.Retained() {
@@ -238,7 +240,11 @@ func (h *Handler) Handle(c *gin.Context) {
 			return
 		}
 		var upstreamOffer string
-		mediaSession, upstreamOffer, errSDP = mediaRelay.NewSession(ctx, clientOffer, proxyURLForAuth(runtimeConfig, selected))
+		mediaSession, upstreamOffer, errSDP = mediaRelay.NewSession(ctx, clientOffer, mediaSessionRoute{
+			proxyURL:   proxyURLForAuth(runtimeConfig, selected),
+			credential: mediaCredentialName(selected, selectedIndex),
+			authIndex:  selectedIndex,
+		})
 		if errSDP != nil {
 			c.JSON(http.StatusBadGateway, gin.H{"error": errSDP.Error()})
 			return
@@ -334,6 +340,17 @@ func (h *Handler) Handle(c *gin.Context) {
 	helps.AppendAPIResponseChunk(ctx, runtimeConfig, responseBody)
 	responseBodyToWrite := responseBody
 	success := resp.StatusCode >= http.StatusOK && resp.StatusCode < http.StatusMultipleChoices
+	callID := ""
+	if success {
+		callID = callIDFromLocation(resp.Header.Get("Location"))
+		if callID == "" && mediaSession != nil {
+			c.JSON(http.StatusBadGateway, gin.H{"error": "Codex live response is missing a valid call ID"})
+			return
+		}
+		if mediaSession != nil {
+			mediaSession.SetCallID(callID)
+		}
+	}
 	if success && mediaSession != nil {
 		upstreamAnswer, errSDP := callResponseSDP(responseBody, resp.Header.Get("Content-Type"))
 		if errSDP != nil {
@@ -351,15 +368,7 @@ func (h *Handler) Handle(c *gin.Context) {
 	var storedSession liveSession
 	sessionStored := false
 	if success && h.sessions != nil {
-		callID := callIDFromLocation(resp.Header.Get("Location"))
-		if callID == "" && mediaSession != nil {
-			c.JSON(http.StatusBadGateway, gin.H{"error": "Codex live response is missing a valid call ID"})
-			return
-		}
 		if callID != "" {
-			if mediaSession != nil {
-				mediaSession.SetCallID(callID)
-			}
 			session := liveSession{authID: selected.ID, model: model, media: mediaSession}
 			if selection != nil {
 				if mediaSession != nil {
@@ -404,6 +413,21 @@ func (h *Handler) Handle(c *gin.Context) {
 	}
 }

+func mediaCredentialName(selected *auth.Auth, authIndex string) string {
+	if selected == nil {
+		return strings.TrimSpace(authIndex)
+	}
+	if label := strings.TrimSpace(selected.Label); label != "" {
+		return label
+	}
+	if fileName := strings.TrimSpace(selected.FileName); fileName != "" {
+		if baseName := strings.TrimSpace(filepath.Base(fileName)); baseName != "" && baseName != "." {
+			return baseName
+		}
+	}
+	return strings.TrimSpace(authIndex)
+}
+
 func (h *Handler) selectOAuth(ctx context.Context, model string, opts coreexecutor.Options) (*auth.HomeDispatchSelection, *auth.Auth, error) {
 	var selection *auth.HomeDispatchSelection
 	var selected *auth.Auth
--- a/internal/client/codex/live/live_test.go
+++ b/internal/client/codex/live/live_test.go
@@ -149,20 +149,21 @@ func (b *trackedResponseBody) Close() error {

 type fakeMediaRelay struct {
 	clientOffer   string
-	proxyURL      string
+	route         mediaSessionRoute
 	upstreamOffer string
 	session       *fakeMediaSession
 	err           error
 }

-func (r *fakeMediaRelay) NewSession(_ context.Context, clientOffer, proxyURL string) (mediaRelaySession, string, error) {
+func (r *fakeMediaRelay) NewSession(_ context.Context, clientOffer string, route mediaSessionRoute) (mediaRelaySession, string, error) {
 	r.clientOffer = clientOffer
-	r.proxyURL = proxyURL
+	r.route = route
 	return r.session, r.upstreamOffer, r.err
 }

 type fakeMediaSession struct {
 	upstreamAnswer string
+	callIDAtAccept string
 	downstreamSDP  string
 	closeHandler   func(string)
 	callID         string
@@ -173,6 +174,7 @@ type fakeMediaSession struct {

 func (s *fakeMediaSession) AcceptUpstreamAnswer(_ context.Context, answer string) (string, error) {
 	s.upstreamAnswer = answer
+	s.callIDAtAccept = s.callID
 	return s.downstreamSDP, s.err
 }

@@ -320,6 +322,36 @@ func TestHandlerRewritesLiveCallAndSchedulesOAuth(t *testing.T) {
 	}
 }

+func TestMediaCredentialNameUsesSafeIdentity(t *testing.T) {
+	for name, testCase := range map[string]struct {
+		selected *auth.Auth
+		index    string
+		want     string
+	}{
+		"label": {
+			selected: &auth.Auth{Label: "Voice credential", FileName: "/auths/codex-user.json", ID: "secret-id"},
+			index:    "auth-index",
+			want:     "Voice credential",
+		},
+		"file basename": {
+			selected: &auth.Auth{FileName: "/auths/codex-user.json", ID: "secret-id"},
+			index:    "auth-index",
+			want:     "codex-user.json",
+		},
+		"opaque index": {
+			selected: &auth.Auth{ID: "secret-id"},
+			index:    "auth-index",
+			want:     "auth-index",
+		},
+	} {
+		t.Run(name, func(t *testing.T) {
+			if got := mediaCredentialName(testCase.selected, testCase.index); got != testCase.want {
+				t.Fatalf("mediaCredentialName() = %q, want %q", got, testCase.want)
+			}
+		})
+	}
+}
+
 func TestProxyURLForAuthPrefersCredentialOverride(t *testing.T) {
 	cfg := &config.Config{}
 	cfg.ProxyURL = "http://global.example:8080"
@@ -346,6 +378,7 @@ func TestHandlerRelaysWebRTCMediaSDP(t *testing.T) {
 		ID:       "codex-oauth",
 		Provider: "codex",
 		Status:   auth.StatusActive,
+		Label:    "Voice credential",
 		ProxyURL: "socks5://credential-proxy.example:1080",
 		Metadata: map[string]any{"access_token": "oauth-token"},
 	})
@@ -374,8 +407,11 @@ func TestHandlerRelaysWebRTCMediaSDP(t *testing.T) {
 	if mediaRelay.clientOffer != "v=0\r\no=desktop-offer\r\n" {
 		t.Fatalf("media client offer = %q", mediaRelay.clientOffer)
 	}
-	if mediaRelay.proxyURL != "socks5://credential-proxy.example:1080" {
-		t.Fatalf("media proxy URL = %q, want credential override", mediaRelay.proxyURL)
+	if mediaRelay.route.proxyURL != "socks5://credential-proxy.example:1080" {
+		t.Fatalf("media proxy URL = %q, want credential override", mediaRelay.route.proxyURL)
+	}
+	if mediaRelay.route.credential != "Voice credential" || mediaRelay.route.authIndex == "" {
+		t.Fatalf("media credential route = %#v", mediaRelay.route)
 	}
 	var upstreamPayload struct {
 		SDP string `json:"sdp"`
@@ -392,6 +428,9 @@ func TestHandlerRelaysWebRTCMediaSDP(t *testing.T) {
 	if mediaSession.callID != "call-123" {
 		t.Fatalf("media call ID = %q, want call-123", mediaSession.callID)
 	}
+	if mediaSession.callIDAtAccept != "call-123" {
+		t.Fatalf("media call ID at answer acceptance = %q, want call-123", mediaSession.callIDAtAccept)
+	}
 	if got := recorder.Body.String(); got != mediaSession.downstreamSDP {
 		t.Fatalf("downstream SDP = %q, want %q", got, mediaSession.downstreamSDP)
 	}
--- a/internal/client/codex/live/media.go
+++ b/internal/client/codex/live/media.go
@@ -42,7 +42,13 @@ type mediaRelaySession interface {
 }

 type mediaRelayFactory interface {
-	NewSession(context.Context, string, string) (mediaRelaySession, string, error)
+	NewSession(context.Context, string, mediaSessionRoute) (mediaRelaySession, string, error)
+}
+
+type mediaSessionRoute struct {
+	proxyURL   string
+	credential string
+	authIndex  string
 }

 type pionMediaRelay struct {
@@ -76,11 +82,14 @@ type pionMediaSession struct {
 	callID         string
 	releaseSlot    func()

-	proxyDialer proxy.ContextDialer
-	proxyScheme string
-	localOffer  string
-	tunnelsMu   sync.Mutex
-	tunnels     []*tcpCandidateTunnel
+	proxyDialer       proxy.ContextDialer
+	proxyScheme       string
+	credential        string
+	authIndex         string
+	forwardingLogOnce sync.Once
+	localOffer        string
+	tunnelsMu         sync.Mutex
+	tunnels           []*tcpCandidateTunnel
 }

 type dataChannelMessage struct {
@@ -248,14 +257,14 @@ func isPublicRemoteIP(ip net.IP) bool {
 		!ip.IsLinkLocalUnicast() && !ip.IsLinkLocalMulticast() && !ip.IsMulticast()
 }

-func (r *pionMediaRelay) NewSession(ctx context.Context, clientOffer, proxyURL string) (mediaRelaySession, string, error) {
+func (r *pionMediaRelay) NewSession(ctx context.Context, clientOffer string, route mediaSessionRoute) (mediaRelaySession, string, error) {
 	if r == nil || r.downstreamAPI == nil || r.upstreamAPI == nil || r.proxyUpstreamAPI == nil || r.limiter == nil {
 		return nil, "", errors.New("Codex live media relay unavailable")
 	}
 	if errContext := ctx.Err(); errContext != nil {
 		return nil, "", errContext
 	}
-	builtProxyDialer, proxyMode, errProxy := proxyutil.BuildDialer(proxyURL)
+	builtProxyDialer, proxyMode, errProxy := proxyutil.BuildDialer(route.proxyURL)
 	if errProxy != nil {
 		return nil, "", fmt.Errorf("configure Codex live remote TCP proxy: %w", errProxy)
 	}
@@ -299,7 +308,9 @@ func (r *pionMediaRelay) NewSession(ctx context.Context, clientOffer, proxyURL s
 		mediaSessionID: uuid.NewString(),
 		releaseSlot:    releaseSlot,
 		proxyDialer:    proxyDialer,
-		proxyScheme:    proxyScheme(proxyURL),
+		proxyScheme:    proxyScheme(route.proxyURL),
+		credential:     strings.TrimSpace(route.credential),
+		authIndex:      strings.TrimSpace(route.authIndex),
 	}
 	session.bridge = newDataChannelBridge(session.done, func(err error) {
 		session.fail("data_channel_failed", err)
@@ -402,6 +413,9 @@ func (s *pionMediaSession) AcceptUpstreamAnswer(ctx context.Context, upstreamAns
 		if errProxy != nil {
 			return "", errProxy
 		}
+		for _, tunnel := range tunnels {
+			tunnel.setForwardingStartedHandler(s.logForwardingStarted)
+		}
 		if !s.installCandidateTunnels(tunnels) {
 			errClosed := errors.New("Codex live media session closed while configuring TCP proxy")
 			if errClose := closeCandidateTunnels(tunnels); errClose != nil {
@@ -494,6 +508,36 @@ func (s *pionMediaSession) logFields(peer string) log.Fields {
 	return fields
 }

+func (s *pionMediaSession) forwardingLogFields() log.Fields {
+	fields := s.logFields("remote")
+	if s.authIndex != "" {
+		fields["auth_index"] = s.authIndex
+	}
+	if s.credential != "" {
+		fields["credential"] = s.credential
+	}
+	if s.proxyDialer != nil {
+		fields["connection"] = "via " + s.proxyScheme + " proxy"
+		fields["remote_transport"] = "tcp"
+	} else {
+		fields["connection"] = "direct"
+		fields["remote_transport"] = "ice"
+	}
+	if s.upstream != nil {
+		fields["state"] = s.upstream.ConnectionState().String()
+	}
+	return fields
+}
+
+func (s *pionMediaSession) logForwardingStarted() {
+	if s == nil {
+		return
+	}
+	s.forwardingLogOnce.Do(func() {
+		log.WithFields(s.forwardingLogFields()).Info("codex live remote media forwarding started")
+	})
+}
+
 func (s *pionMediaSession) SetCloseHandler(handler func(string)) {
 	if s == nil {
 		return
@@ -576,6 +620,9 @@ func (s *pionMediaSession) installStateHandlers() {
 				log.WithFields(fields).Info("codex live WebRTC peer connecting")
 			case webrtc.PeerConnectionStateConnected:
 				log.WithFields(fields).Info("codex live WebRTC peer connected")
+				if peer == "remote" {
+					s.logForwardingStarted()
+				}
 			case webrtc.PeerConnectionStateDisconnected:
 				log.WithFields(fields).Warn("codex live WebRTC peer disconnected")
 			case webrtc.PeerConnectionStateFailed:
--- a/internal/client/codex/live/media_test.go
+++ b/internal/client/codex/live/media_test.go
@@ -2,6 +2,7 @@ package live

 import (
 	"context"
+	"fmt"
 	"net"
 	"strings"
 	"testing"
@@ -46,7 +47,7 @@ func TestPionMediaRelaySelectsRemoteProxyMode(t *testing.T) {
 		"SOCKS5H": {proxyURL: "socks5h://proxy.example:1080", proxied: true},
 	} {
 		t.Run(name, func(t *testing.T) {
-			session, upstreamOffer, errSession := relay.NewSession(context.Background(), clientOffer, testCase.proxyURL)
+			session, upstreamOffer, errSession := relay.NewSession(context.Background(), clientOffer, mediaSessionRoute{proxyURL: testCase.proxyURL})
 			if errSession != nil {
 				t.Fatalf("create media session: %v", errSession)
 			}
@@ -66,11 +67,79 @@ func TestPionMediaRelaySelectsRemoteProxyMode(t *testing.T) {
 		})
 	}

-	if _, _, errSession := relay.NewSession(context.Background(), clientOffer, "invalid-proxy"); errSession == nil {
+	if _, _, errSession := relay.NewSession(context.Background(), clientOffer, mediaSessionRoute{proxyURL: "invalid-proxy"}); errSession == nil {
 		t.Fatal("expected invalid proxy URL to fail media session creation")
 	}
 }

+func TestMediaForwardingStartedLogRedactsProxyCredentials(t *testing.T) {
+	logger := log.StandardLogger()
+	previousHooks := logger.ReplaceHooks(make(log.LevelHooks))
+	hook := logtest.NewLocal(logger)
+	defer logger.ReplaceHooks(previousHooks)
+
+	for name, testCase := range map[string]struct {
+		proxyURL   string
+		connection string
+		credential string
+	}{
+		"direct": {
+			connection: "direct",
+			credential: "Voice credential",
+		},
+		"HTTP": {
+			proxyURL:   "http://user:secret@proxy.example:8080",
+			connection: "via http proxy",
+			credential: "Voice credential",
+		},
+		"SOCKS5 without label": {
+			proxyURL:   "socks5://user:secret@proxy.example:1080",
+			connection: "via socks5 proxy",
+			credential: "auth-index",
+		},
+	} {
+		t.Run(name, func(t *testing.T) {
+			session := &pionMediaSession{
+				mediaSessionID: "media-session-" + name,
+				proxyScheme:    proxyScheme(testCase.proxyURL),
+				credential:     testCase.credential,
+				authIndex:      "auth-index",
+			}
+			if testCase.proxyURL != "" {
+				session.proxyDialer = &recordingProxyDialer{dials: make(chan recordedProxyDial, 1)}
+			}
+			earlyFields := session.logFields("session")
+			for _, field := range []string{"auth_id", "auth_label", "auth_index", "credential", "connection"} {
+				if _, exists := earlyFields[field]; exists {
+					t.Fatalf("session log exposed forwarding-only field %q before forwarding started: %#v", field, earlyFields)
+				}
+			}
+			session.logForwardingStarted()
+			session.logForwardingStarted()
+
+			matching := 0
+			for _, entry := range hook.AllEntries() {
+				if entry.Message != "codex live remote media forwarding started" || entry.Data["media_session_id"] != session.mediaSessionID {
+					continue
+				}
+				matching++
+				if entry.Data["connection"] != testCase.connection || entry.Data["credential"] != testCase.credential {
+					t.Fatalf("forwarding fields = %#v", entry.Data)
+				}
+				serialized := fmt.Sprint(entry.Data)
+				for _, secret := range []string{"user", "secret", "proxy.example"} {
+					if strings.Contains(serialized, secret) {
+						t.Fatalf("forwarding log leaked %q: %s", secret, serialized)
+					}
+				}
+			}
+			if matching != 1 {
+				t.Fatalf("forwarding log count = %d, want 1", matching)
+			}
+		})
+	}
+}
+
 func TestPionMediaRelayBridgesAudioAndDataChannel(t *testing.T) {
 	logger := log.StandardLogger()
 	previousHooks := logger.ReplaceHooks(make(log.LevelHooks))
@@ -126,7 +195,10 @@ func TestPionMediaRelayBridgesAudioAndDataChannel(t *testing.T) {
 	if errRelay != nil {
 		t.Fatalf("create media relay: %v", errRelay)
 	}
-	session, relayOffer, errSession := relay.NewSession(context.Background(), clientOffer, "")
+	session, relayOffer, errSession := relay.NewSession(context.Background(), clientOffer, mediaSessionRoute{
+		credential: "Voice credential",
+		authIndex:  "auth-index",
+	})
 	if errSession != nil {
 		t.Fatalf("create media relay session: %v", errSession)
 	}
@@ -140,7 +212,7 @@ func TestPionMediaRelayBridgesAudioAndDataChannel(t *testing.T) {
 	if errRelay != nil {
 		t.Fatalf("reload media relay: %v", errRelay)
 	}
-	if _, _, errCapacity := reloadedRelay.NewSession(context.Background(), clientOffer, ""); errCapacity == nil {
+	if _, _, errCapacity := reloadedRelay.NewSession(context.Background(), clientOffer, mediaSessionRoute{}); errCapacity == nil {
 		t.Fatal("reloaded media relay bypassed the shared session capacity")
 	}

@@ -222,7 +294,7 @@ func TestPionMediaRelayBridgesAudioAndDataChannel(t *testing.T) {
 	if errClose := session.Close(); errClose != nil {
 		t.Fatalf("close media relay session for logging: %v", errClose)
 	}
-	replacementSession, _, errReplacement := reloadedRelay.NewSession(context.Background(), clientOffer, "")
+	replacementSession, _, errReplacement := reloadedRelay.NewSession(context.Background(), clientOffer, mediaSessionRoute{})
 	if errReplacement != nil {
 		t.Fatalf("shared capacity was not released: %v", errReplacement)
 	}
@@ -233,6 +305,8 @@ func TestPionMediaRelayBridgesAudioAndDataChannel(t *testing.T) {
 		assertPeerLog(t, hook, "codex live WebRTC peer connected", peer, "call-log-test")
 		assertPeerLog(t, hook, "codex live WebRTC peer closed", peer, "call-log-test")
 	}
+	assertForwardingLog(t, hook, "direct", "Voice credential", "auth-index", "connected")
+	assertForwardingAfterRemoteConnected(t, hook)
 	assertSessionLog(t, hook, "codex live WebRTC media session closed", "closed", "call-log-test")
 }

@@ -397,6 +471,41 @@ func sendTestRTP(t *testing.T, track *webrtc.TrackLocalStaticRTP, payload []byte
 	t.Fatal("RTP packet was not relayed")
 }

+func assertForwardingAfterRemoteConnected(t *testing.T, hook *logtest.Hook) {
+	t.Helper()
+	connectedIndex := -1
+	forwardingIndex := -1
+	for index, entry := range hook.AllEntries() {
+		if entry.Message == "codex live WebRTC peer connected" && entry.Data["peer"] == "remote" && connectedIndex == -1 {
+			connectedIndex = index
+		}
+		if entry.Message == "codex live remote media forwarding started" && forwardingIndex == -1 {
+			forwardingIndex = index
+		}
+	}
+	if connectedIndex == -1 || forwardingIndex <= connectedIndex {
+		t.Fatalf("remote connected index=%d, forwarding index=%d", connectedIndex, forwardingIndex)
+	}
+}
+
+func assertForwardingLog(t *testing.T, hook *logtest.Hook, connection, credential, authIndex, state string) {
+	t.Helper()
+	deadline := time.Now().Add(time.Second)
+	for time.Now().Before(deadline) {
+		for _, entry := range hook.AllEntries() {
+			if entry.Message == "codex live remote media forwarding started" &&
+				entry.Data["connection"] == connection &&
+				entry.Data["credential"] == credential &&
+				entry.Data["auth_index"] == authIndex &&
+				entry.Data["state"] == state {
+				return
+			}
+		}
+		time.Sleep(time.Millisecond)
+	}
+	t.Fatalf("missing forwarding log for connection %q and credential %q", connection, credential)
+}
+
 func assertSessionLog(t *testing.T, hook *logtest.Hook, message, reason, callID string) {
 	t.Helper()
 	deadline := time.Now().Add(time.Second)
--- a/internal/client/codex/live/tcp_proxy.go
+++ b/internal/client/codex/live/tcp_proxy.go
@@ -71,13 +71,14 @@ type tcpCandidateTunnel struct {
 	expectedUser   string
 	remotePassword string

-	mu              sync.Mutex
-	closed          bool
-	claimed         bool
-	connections     map[net.Conn]struct{}
-	validationSlots chan struct{}
-	ctx             context.Context
-	cancel          context.CancelFunc
+	mu                  sync.Mutex
+	closed              bool
+	claimed             bool
+	connections         map[net.Conn]struct{}
+	validationSlots     chan struct{}
+	onForwardingStarted func()
+	ctx                 context.Context
+	cancel              context.CancelFunc
 }

 type tcpCandidatePlan struct {
@@ -379,6 +380,7 @@ func (t *tcpCandidateTunnel) handleConnection(client net.Conn) {
 		log.WithError(errWrite).Warn("codex live TCP proxy: forward authenticated ICE frame failed")
 		return
 	}
+	t.notifyForwardingStarted()

 	copyDone := make(chan struct{}, 2)
 	copyConnection := func(destination, source net.Conn) {
@@ -393,6 +395,27 @@ func (t *tcpCandidateTunnel) handleConnection(client net.Conn) {
 	<-copyDone
 }

+func (t *tcpCandidateTunnel) setForwardingStartedHandler(handler func()) {
+	if t == nil {
+		return
+	}
+	t.mu.Lock()
+	t.onForwardingStarted = handler
+	t.mu.Unlock()
+}
+
+func (t *tcpCandidateTunnel) notifyForwardingStarted() {
+	if t == nil {
+		return
+	}
+	t.mu.Lock()
+	handler := t.onForwardingStarted
+	t.mu.Unlock()
+	if handler != nil {
+		handler()
+	}
+}
+
 func (t *tcpCandidateTunnel) trackConnection(connection net.Conn) bool {
 	t.mu.Lock()
 	defer t.mu.Unlock()
--- a/internal/client/codex/live/tcp_proxy_test.go
+++ b/internal/client/codex/live/tcp_proxy_test.go
@@ -37,6 +37,14 @@ type blockingContextDialer struct {
 	canceled chan struct{}
 }

+type closedUpstreamDialer struct{}
+
+func (*closedUpstreamDialer) DialContext(context.Context, string, string) (net.Conn, error) {
+	client, server := net.Pipe()
+	_ = server.Close()
+	return client, nil
+}
+
 func (d *blockingContextDialer) DialContext(ctx context.Context, _, _ string) (net.Conn, error) {
 	close(d.started)
 	<-ctx.Done()
@@ -228,6 +236,10 @@ func TestTCPCandidateTunnelAuthenticatesBeforeFixedTargetDial(t *testing.T) {
 		t.Fatalf("newTCPCandidateTunnel returned error: %v", errTunnel)
 	}
 	defer func() { _ = tunnel.Close() }()
+	forwardingStarted := make(chan struct{}, 1)
+	tunnel.setForwardingStartedHandler(func() {
+		forwardingStarted <- struct{}{}
+	})

 	client, errDial := net.Dial("tcp", tunnel.listener.Addr().String())
 	if errDial != nil {
@@ -256,6 +268,11 @@ func TestTCPCandidateTunnelAuthenticatesBeforeFixedTargetDial(t *testing.T) {
 	if !bytes.Equal(forwarded, frame) {
 		t.Fatal("forwarded STUN frame changed")
 	}
+	select {
+	case <-forwardingStarted:
+	case <-time.After(time.Second):
+		t.Fatal("forwarding start handler was not called")
+	}
 	if errWrite := writeAll(dial.connection, []byte("reply")); errWrite != nil {
 		t.Fatalf("write tunnel reply: %v", errWrite)
 	}
@@ -295,6 +312,8 @@ func TestTCPCandidateTunnelCloseCancelsProxyDial(t *testing.T) {
 	case <-time.After(time.Second):
 		t.Fatal("proxy dial did not start")
 	}
+	forwardingStarted := make(chan struct{}, 1)
+	tunnel.setForwardingStartedHandler(func() { forwardingStarted <- struct{}{} })
 	if errClose := tunnel.Close(); errClose != nil {
 		t.Fatalf("close tunnel: %v", errClose)
 	}
@@ -303,6 +322,7 @@ func TestTCPCandidateTunnelCloseCancelsProxyDial(t *testing.T) {
 	case <-time.After(time.Second):
 		t.Fatal("tunnel close did not cancel proxy dial")
 	}
+	assertNoForwardingStart(t, forwardingStarted)
 }

 func TestTCPCandidateTunnelProxyFailureDoesNotFallBack(t *testing.T) {
@@ -320,6 +340,8 @@ func TestTCPCandidateTunnelProxyFailureDoesNotFallBack(t *testing.T) {
 		t.Fatalf("newTCPCandidateTunnel returned error: %v", errTunnel)
 	}
 	defer func() { _ = tunnel.Close() }()
+	forwardingStarted := make(chan struct{}, 1)
+	tunnel.setForwardingStartedHandler(func() { forwardingStarted <- struct{}{} })
 	client, errDial := net.Dial("tcp", tunnel.listener.Addr().String())
 	if errDial != nil {
 		t.Fatalf("dial candidate listener: %v", errDial)
@@ -339,6 +361,31 @@ func TestTCPCandidateTunnelProxyFailureDoesNotFallBack(t *testing.T) {
 	if _, errSecondDial := net.Dial("tcp", tunnel.listener.Addr().String()); errSecondDial == nil {
 		t.Fatal("candidate listener remained available after proxy failure")
 	}
+	assertNoForwardingStart(t, forwardingStarted)
+}
+
+func TestTCPCandidateTunnelWriteFailureDoesNotLogForwardingStart(t *testing.T) {
+	tunnel, errTunnel := newTCPCandidateTunnel(
+		netip.MustParseAddrPort("20.42.0.20:443"),
+		&closedUpstreamDialer{},
+		"remote:local",
+		"remote-password",
+	)
+	if errTunnel != nil {
+		t.Fatalf("newTCPCandidateTunnel returned error: %v", errTunnel)
+	}
+	defer func() { _ = tunnel.Close() }()
+	forwardingStarted := make(chan struct{}, 1)
+	tunnel.setForwardingStartedHandler(func() { forwardingStarted <- struct{}{} })
+	client, errDial := net.Dial("tcp", tunnel.listener.Addr().String())
+	if errDial != nil {
+		t.Fatalf("dial candidate listener: %v", errDial)
+	}
+	if errWrite := writeAll(client, buildTestICEFrame(t, "remote:local", "remote-password", true)); errWrite != nil {
+		t.Fatalf("write authenticated frame: %v", errWrite)
+	}
+	_ = client.Close()
+	assertNoForwardingStart(t, forwardingStarted)
 }

 func TestTCPCandidateTunnelRejectsUnauthenticatedConnectionWithoutDial(t *testing.T) {
@@ -353,6 +400,8 @@ func TestTCPCandidateTunnelRejectsUnauthenticatedConnectionWithoutDial(t *testin
 		t.Fatalf("newTCPCandidateTunnel returned error: %v", errTunnel)
 	}
 	defer func() { _ = tunnel.Close() }()
+	forwardingStarted := make(chan struct{}, 1)
+	tunnel.setForwardingStartedHandler(func() { forwardingStarted <- struct{}{} })

 	client, errDial := net.Dial("tcp", tunnel.listener.Addr().String())
 	if errDial != nil {
@@ -368,6 +417,16 @@ func TestTCPCandidateTunnelRejectsUnauthenticatedConnectionWithoutDial(t *testin
 		t.Fatalf("unauthenticated connection triggered proxy dial to %q", dial.address)
 	case <-time.After(100 * time.Millisecond):
 	}
+	assertNoForwardingStart(t, forwardingStarted)
+}
+
+func assertNoForwardingStart(t *testing.T, started <-chan struct{}) {
+	t.Helper()
+	select {
+	case <-started:
+		t.Fatal("forwarding start handler was called for an unestablished tunnel")
+	case <-time.After(100 * time.Millisecond):
+	}
 }

 func TestPionActiveTCPCandidatePassesTunnelAuthentication(t *testing.T) {
--- a/internal/logging/global_logger.go
+++ b/internal/logging/global_logger.go
@@ -6,6 +6,7 @@ import (
 	"io"
 	"os"
 	"path/filepath"
+	"strconv"
 	"strings"
 	"sync"

@@ -35,10 +36,33 @@ var logFieldOrder = []string{
 	"plugin_id", "plugin_name", "source_id",
 	"version", "active_version", "retired_version", "overwritten",
 	"mode", "budget", "level", "original_mode", "original_value", "min", "max", "clamped_to", "error",
+	"credential", "connection", "proxy_scheme", "remote_transport",
+	"media_session_id", "call_id", "peer", "state", "reason",
+}
+
+var quotedLogFields = map[string]struct{}{
+	"credential":       {},
+	"connection":       {},
+	"proxy_scheme":     {},
+	"remote_transport": {},
+	"media_session_id": {},
+	"call_id":          {},
+	"peer":             {},
+	"state":            {},
+	"reason":           {},
 }

 var pluginPathFieldOrder = []string{"path", "active_path", "retired_path"}

+func formatLogFieldValue(key string, value any) string {
+	if _, quoted := quotedLogFields[key]; quoted {
+		if stringValue, ok := value.(string); ok {
+			return strconv.Quote(stringValue)
+		}
+	}
+	return fmt.Sprint(value)
+}
+
 // Format renders a single log entry with custom formatting.
 func (m *LogFormatter) Format(entry *log.Entry) ([]byte, error) {
 	var buffer *bytes.Buffer
@@ -68,7 +92,7 @@ func (m *LogFormatter) Format(entry *log.Entry) ([]byte, error) {
 		var fields []string
 		for _, k := range logFieldOrder {
 			if v, ok := entry.Data[k]; ok {
-				fields = append(fields, fmt.Sprintf("%s=%v", k, v))
+				fields = append(fields, fmt.Sprintf("%s=%s", k, formatLogFieldValue(k, v)))
 			}
 		}
 		if pluginID, ok := entry.Data["plugin_id"]; ok && strings.TrimSpace(fmt.Sprint(pluginID)) != "" {
--- a/internal/logging/global_logger_test.go
+++ b/internal/logging/global_logger_test.go
@@ -26,6 +26,45 @@ func TestLogFormatterPrintsVersionField(t *testing.T) {
 	}
 }

+func TestLogFormatterPrintsMediaForwardingFields(t *testing.T) {
+	entry := log.NewEntry(log.New())
+	entry.Time = time.Date(2026, 7, 25, 7, 36, 4, 0, time.Local)
+	entry.Level = log.InfoLevel
+	entry.Message = "codex live remote media forwarding started"
+	entry.Data["credential"] = "Voice credential\nsecondary"
+	entry.Data["connection"] = "via socks5 proxy"
+	entry.Data["proxy_scheme"] = "socks5"
+	entry.Data["remote_transport"] = "tcp"
+	entry.Data["media_session_id"] = "media-session-id"
+	entry.Data["call_id"] = "call-id"
+	entry.Data["peer"] = "remote"
+	entry.Data["state"] = "connected"
+
+	formatted, errFormat := (&LogFormatter{}).Format(entry)
+	if errFormat != nil {
+		t.Fatalf("Format() error = %v", errFormat)
+	}
+
+	line := string(formatted)
+	for _, want := range []string{
+		`credential="Voice credential\nsecondary"`,
+		`connection="via socks5 proxy"`,
+		`proxy_scheme="socks5"`,
+		`remote_transport="tcp"`,
+		`media_session_id="media-session-id"`,
+		`call_id="call-id"`,
+		`peer="remote"`,
+		`state="connected"`,
+	} {
+		if !strings.Contains(line, want) {
+			t.Fatalf("formatted line %q missing %s", line, want)
+		}
+	}
+	if strings.Count(line, "\n") != 1 {
+		t.Fatalf("formatted line contains an unescaped newline: %q", line)
+	}
+}
+
 func TestLogFormatterPrintsPluginFields(t *testing.T) {
 	entry := log.NewEntry(log.New())
 	entry.Time = time.Date(2026, 6, 25, 20, 10, 0, 0, time.Local)