feat(providers): preserve resolved capabilities on switched-model fallback sessions

* perf(providers): add provider capabilities and default turn-session adapter (PR7)

* fix(providers): project effective reasoning efforts into capabilities

* feat(agent): carry optimized turn sessions across mid-run model switches

* test(agent): cover swap-time session open failure

README.md

@@ -358,6 +358,12 @@ go run ./cmd/zero-release smoke
 go run ./cmd/zero-perf-bench
 ```
 
+Experimental: `ZERO_OPENAI_TURN_SESSION=1` enables the optimized OpenAI turn
+session (background connection prewarm + request-prefix telemetry) for headless
+`zero exec` runs against official OpenAI profiles. Off by default; `0`/`false`
+disable. A/B-benchmark it by running the same `zero-perf-bench` suite with the
+variable unset and set.
+
 ### Code Quality and Security Checks
 
 Before committing any changes, ensure all Go code quality and security checks pass. Pinned `go run` commands matching CI constraints can be used directly without prior installation:

internal/agent/loop.go

@@ -136,6 +136,33 @@ func Run(ctx context.Context, prompt string, provider Provider, options Options)
 		}
 	}
 
+	// Establish the run's turn session — the seam an optimized provider session
+	// (connection reuse, prewarm, native compaction) plugs into without touching
+	// this loop. Default: wrap the passed provider so the session's Stream IS
+	// provider.StreamCompletion, byte-identical to calling it directly. Opened
+	// before dispatchSessionStart so a failed open is a clean run-start error
+	// with no session hooks fired and nothing to unwind.
+	turnSessions := options.TurnSessionProvider
+	if turnSessions == nil {
+		turnSessions = zeroruntime.NewProviderTurnSessionProvider(provider, zeroruntime.ProviderCapabilities{})
+	}
+	session, sessionErr := turnSessions.OpenTurnSession(ctx)
+	if sessionErr != nil {
+		return Result{}, fmt.Errorf("agent: open turn session: %w", sessionErr)
+	}
+	// Closed via closure (not `defer session.Close()`) so after a mid-run model
+	// swap the CURRENT session is the one closed at teardown — the swap closes
+	// the old one inline. Close errors are advisory and never fail the run.
+	defer func() { _ = session.Close() }()
+	// Best-effort prewarm: the default adapter no-ops; a real session may prime
+	// a connection. Failure is ignored by contract (never fatal).
+	_ = session.Prewarm(ctx)
+	// Route ALL provider I/O — per-turn streams, mid-stream retries, the
+	// compaction summarizer, and the final-answer request — through the session
+	// by reassigning the loop's provider local. Every existing call site reads
+	// this local, so no signatures change anywhere downstream.
+	provider = sessionProvider{session: session}
+
 	maxTurns := options.MaxTurns
 	if maxTurns <= 0 {
 		maxTurns = 12
@@ -710,25 +737,53 @@ func Run(ctx context.Context, prompt string, provider Provider, options Options)
 		// of the run. At most one switch per turn (first request wins). A switcher
 		// error is NON-FATAL — record a brief note and continue on the current
 		// model. nil switcher ⇒ requests are ignored entirely (escalation off).
-		if turnRequestedModel != "" && options.ModelSwitcher != nil {
-			newProvider, switchErr := options.ModelSwitcher(ctx, turnRequestedModel)
+		if turnRequestedModel != "" && (options.ModelSessionSwitcher != nil || options.ModelSwitcher != nil) {
+			// Resolve the escalated model's session source. The target-aware
+			// ModelSessionSwitcher is preferred: its TurnSessionProvider carries an
+			// optimized session (and capabilities) across the swap. The legacy
+			// ModelSwitcher fallback returns a bare Provider, wrapped in the
+			// default no-op session — identical to pre-seam behavior.
+			var newSessions zeroruntime.TurnSessionProvider
+			var switchErr error
+			if options.ModelSessionSwitcher != nil {
+				newSessions, switchErr = options.ModelSessionSwitcher(ctx, turnRequestedModel)
+			} else {
+				var newProvider Provider
+				newProvider, switchErr = options.ModelSwitcher(ctx, turnRequestedModel)
+				if switchErr == nil && newProvider != nil {
+					newSessions = zeroruntime.NewProviderTurnSessionProvider(newProvider, zeroruntime.ProviderCapabilities{})
+				}
+			}
 			if switchErr != nil {
 				messages = append(messages, zeroruntime.Message{
 					Role:    zeroruntime.MessageRoleUser,
 					Content: escalationFailedNoticePrefix + " (" + turnRequestedModel + "): " + switchErr.Error() + ". Continuing on " + options.Model + ".",
 				})
-			} else if newProvider != nil {
-				// Reassign the local provider so the next turn's StreamCompletion and
-				// compaction use it; update options.Model so subsequent RunOptions.Model,
-				// context-window sizing, and usage attribution follow the new model.
-				provider = newProvider
-				options.Model = turnRequestedModel
-				options.Trace.Counter(trace.CounterModelSwitches, 1)
-				// KNOWN LIMITATION (deferred): the compactor's context-window budget
-				// is fixed at run start from options.ContextWindow and is NOT updated
-				// here, so a switch to a model with a different window keeps compacting
-				// against the original budget. Fixing it needs a ModelSwitcher contract
-				// change (return the new window) — out of scope for this change.
+			} else if newSessions != nil {
+				// Open the new session and close the old one, so the next turn's
+				// streams and compaction use the new model. For the default adapter
+				// the open never errors and Close is a no-op, so this stays
+				// byte-identical to a direct provider reassignment; the guarded open
+				// and prewarm matter once a session holds real state.
+				newSession, openErr := newSessions.OpenTurnSession(ctx)
+				if openErr != nil {
+					messages = append(messages, zeroruntime.Message{
+						Role:    zeroruntime.MessageRoleUser,
+						Content: escalationFailedNoticePrefix + " (" + turnRequestedModel + "): " + openErr.Error() + ". Continuing on " + options.Model + ".",
+					})
+				} else {
+					_ = session.Close()
+					session = newSession
+					_ = session.Prewarm(ctx)
+					provider = sessionProvider{session: session}
+					options.Model = turnRequestedModel
+					options.Trace.Counter(trace.CounterModelSwitches, 1)
+					// KNOWN LIMITATION (deferred): the compactor's context-window budget
+					// is fixed at run start from options.ContextWindow and is NOT updated
+					// here, so a switch to a model with a different window keeps compacting
+					// against the original budget. Fixing it needs the switcher to also
+					// report the new window — out of scope for this change.
+				}
 			}
 		}
 

internal/agent/turn_session.go

@@ -0,0 +1,21 @@
+package agent
+
+import (
+	"context"
+
+	"github.com/Gitlawb/zero/internal/zeroruntime"
+)
+
+// sessionProvider adapts a TurnSession back to the Provider interface so the
+// loop's existing Provider-shaped call sites (streamWithReconnect, the
+// compaction summarizer, finalAnswerAfterMaxTurns) route their stream I/O
+// through the session without any signature change. For the default adapter
+// this reduces to the wrapped provider's StreamCompletion, so behavior is
+// byte-identical; an optimized session (PR8) takes effect here transparently.
+type sessionProvider struct {
+	session zeroruntime.TurnSession
+}
+
+func (s sessionProvider) StreamCompletion(ctx context.Context, request zeroruntime.CompletionRequest) (<-chan zeroruntime.StreamEvent, error) {
+	return s.session.Stream(ctx, request)
+}

internal/agent/turn_session_test.go

@@ -0,0 +1,367 @@
+package agent
+
+import (
+	"context"
+	"errors"
+	"strings"
+	"testing"
+
+	"github.com/Gitlawb/zero/internal/tools"
+	"github.com/Gitlawb/zero/internal/zeroruntime"
+)
+
+// fakeTurnSession wraps an inner provider for streaming while counting the
+// session lifecycle, so tests can observe exactly how the loop drives it.
+type fakeTurnSession struct {
+	inner    Provider
+	prewarms int
+	streams  int
+	closes   int
+	closeErr error
+}
+
+func (s *fakeTurnSession) Prewarm(context.Context) error { s.prewarms++; return nil }
+
+func (s *fakeTurnSession) Stream(ctx context.Context, request zeroruntime.CompletionRequest) (<-chan zeroruntime.StreamEvent, error) {
+	s.streams++
+	return s.inner.StreamCompletion(ctx, request)
+}
+
+func (s *fakeTurnSession) Compact(context.Context, zeroruntime.CompletionRequest) ([]zeroruntime.Message, error) {
+	return nil, zeroruntime.ErrCompactionUnsupported
+}
+
+func (s *fakeTurnSession) Close() error { s.closes++; return s.closeErr }
+
+// fakeTurnSessionProvider opens a scripted session (or fails to).
+type fakeTurnSessionProvider struct {
+	session *fakeTurnSession
+	openErr error
+	opens   int
+}
+
+func (p *fakeTurnSessionProvider) OpenTurnSession(context.Context) (zeroruntime.TurnSession, error) {
+	p.opens++
+	if p.openErr != nil {
+		return nil, p.openErr
+	}
+	return p.session, nil
+}
+
+func (p *fakeTurnSessionProvider) Capabilities() zeroruntime.ProviderCapabilities {
+	return zeroruntime.ProviderCapabilities{}
+}
+
+func singleAnswerTurns(answer string) [][]zeroruntime.StreamEvent {
+	return [][]zeroruntime.StreamEvent{{
+		{Type: zeroruntime.StreamEventText, Content: answer},
+		{Type: zeroruntime.StreamEventDone},
+	}}
+}
+
+// TestRunOpenTurnSessionFailureSurfaces verifies a failed session open is a
+// clean run-start error: no panic, no stream ever issued, nothing to close.
+func TestRunOpenTurnSessionFailureSurfaces(t *testing.T) {
+	positional := &mockProvider{turns: singleAnswerTurns("never reached")}
+	tsp := &fakeTurnSessionProvider{openErr: errors.New("handshake refused")}
+
+	_, err := Run(context.Background(), "go", positional, Options{
+		MaxTurns:            2,
+		TurnSessionProvider: tsp,
+	})
+	if err == nil {
+		t.Fatal("expected an error from a failed session open")
+	}
+	if !strings.Contains(err.Error(), "open turn session") || !strings.Contains(err.Error(), "handshake refused") {
+		t.Fatalf("expected a wrapped open-turn-session error, got %v", err)
+	}
+	if len(positional.requests) != 0 {
+		t.Fatalf("expected no provider request after a failed open, got %d", len(positional.requests))
+	}
+	if tsp.opens != 1 {
+		t.Fatalf("expected exactly one open attempt, got %d", tsp.opens)
+	}
+}
+
+// TestRunCloseFailureIsSafe verifies a Close error at teardown is swallowed:
+// the run still returns its normal result.
+func TestRunCloseFailureIsSafe(t *testing.T) {
+	inner := &mockProvider{turns: singleAnswerTurns("done")}
+	session := &fakeTurnSession{inner: inner, closeErr: errors.New("close blew up")}
+	tsp := &fakeTurnSessionProvider{session: session}
+
+	result, err := Run(context.Background(), "go", inner, Options{
+		MaxTurns:            2,
+		TurnSessionProvider: tsp,
+	})
+	if err != nil {
+		t.Fatalf("expected Close failure to be swallowed, got %v", err)
+	}
+	if result.FinalAnswer != "done" {
+		t.Fatalf("expected normal result despite Close error, got %q", result.FinalAnswer)
+	}
+	if session.closes != 1 {
+		t.Fatalf("expected exactly one Close at teardown, got %d", session.closes)
+	}
+}
+
+// TestRunStreamsThroughTurnSession verifies that when a TurnSessionProvider is
+// wired, every model request of the run flows through the session (and the
+// session is prewarmed once and closed once).
+func TestRunStreamsThroughTurnSession(t *testing.T) {
+	inner := &mockProvider{turns: singleAnswerTurns("via session")}
+	session := &fakeTurnSession{inner: inner}
+	tsp := &fakeTurnSessionProvider{session: session}
+
+	// The positional provider is a DIFFERENT mock: it must stay untouched,
+	// proving the session (not the positional provider) carried the run.
+	positional := &mockProvider{turns: singleAnswerTurns("wrong path")}
+
+	result, err := Run(context.Background(), "go", positional, Options{
+		MaxTurns:            2,
+		TurnSessionProvider: tsp,
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if result.FinalAnswer != "via session" {
+		t.Fatalf("expected the session-backed answer, got %q", result.FinalAnswer)
+	}
+	if len(positional.requests) != 0 {
+		t.Fatalf("expected the positional provider to be bypassed, got %d requests", len(positional.requests))
+	}
+	if session.streams == 0 || session.streams != len(inner.requests) {
+		t.Fatalf("expected every request to flow through the session: streams=%d inner=%d", session.streams, len(inner.requests))
+	}
+	if session.prewarms != 1 {
+		t.Fatalf("expected exactly one prewarm, got %d", session.prewarms)
+	}
+	if session.closes != 1 {
+		t.Fatalf("expected exactly one close, got %d", session.closes)
+	}
+}
+
+// TestRunDefaultTurnSessionMatchesRawProvider verifies the default (nil
+// TurnSessionProvider) path is behavior-identical to passing an explicit
+// default adapter over the same provider: same result, same request count.
+func TestRunDefaultTurnSessionMatchesRawProvider(t *testing.T) {
+	run := func(explicit bool) (Result, *mockProvider) {
+		provider := &mockProvider{turns: escalateThenAnswerTurns("identical")}
+		registry := tools.NewRegistry()
+		registry.Register(escalatingTool{target: ""})
+		options := Options{Registry: registry, MaxTurns: 4}
+		if explicit {
+			options.TurnSessionProvider = zeroruntime.NewProviderTurnSessionProvider(provider, zeroruntime.ProviderCapabilities{})
+		}
+		result, err := Run(context.Background(), "go", provider, options)
+		if err != nil {
+			t.Fatal(err)
+		}
+		return result, provider
+	}
+
+	defaultResult, defaultProvider := run(false)
+	explicitResult, explicitProvider := run(true)
+
+	if defaultResult.FinalAnswer != explicitResult.FinalAnswer {
+		t.Fatalf("final answers diverged: default=%q explicit=%q", defaultResult.FinalAnswer, explicitResult.FinalAnswer)
+	}
+	if defaultResult.Turns != explicitResult.Turns {
+		t.Fatalf("turn counts diverged: default=%d explicit=%d", defaultResult.Turns, explicitResult.Turns)
+	}
+	if len(defaultProvider.requests) != len(explicitProvider.requests) {
+		t.Fatalf("request counts diverged: default=%d explicit=%d", len(defaultProvider.requests), len(explicitProvider.requests))
+	}
+}
+
+// TestRunModelSwitchClosesSessionAndContinues verifies a mid-run model switch
+// closes the active session exactly once and streams the rest of the run on
+// the new provider.
+func TestRunModelSwitchClosesSessionAndContinues(t *testing.T) {
+	registry := tools.NewRegistry()
+	registry.Register(escalatingTool{target: "claude-opus-4.1"})
+
+	firstInner := &mockProvider{turns: escalateThenAnswerTurns("unused")}
+	firstSession := &fakeTurnSession{inner: firstInner}
+	tsp := &fakeTurnSessionProvider{session: firstSession}
+
+	secondProvider := &mockProvider{turns: singleAnswerTurns("answered after switch")}
+
+	result, err := Run(context.Background(), "go", firstInner, Options{
+		Registry:            registry,
+		Model:               "claude-sonnet-4.5",
+		MaxTurns:            4,
+		TurnSessionProvider: tsp,
+		ModelSwitcher: func(_ context.Context, _ string) (Provider, error) {
+			return secondProvider, nil
+		},
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if result.FinalAnswer != "answered after switch" {
+		t.Fatalf("expected the post-switch answer, got %q", result.FinalAnswer)
+	}
+	// The original session handled exactly the escalation turn, then was closed
+	// once by the swap (the teardown defer closes the NEW session, not this one).
+	if firstSession.streams != 1 {
+		t.Fatalf("expected the original session to stream exactly the escalation turn, got %d", firstSession.streams)
+	}
+	if firstSession.closes != 1 {
+		t.Fatalf("expected the swap to close the original session exactly once, got %d", firstSession.closes)
+	}
+	if len(secondProvider.requests) != 1 {
+		t.Fatalf("expected the post-switch turn on the new provider, got %d requests", len(secondProvider.requests))
+	}
+}
+
+// TestRunModelSessionSwitcherCarriesOptimizedSession verifies the target-aware
+// switcher: the switched-to TurnSessionProvider's DISTINCT session receives the
+// post-switch streams, is prewarmed exactly once, and is closed exactly once at
+// teardown — and the legacy ModelSwitcher is never consulted when both are set.
+func TestRunModelSessionSwitcherCarriesOptimizedSession(t *testing.T) {
+	registry := tools.NewRegistry()
+	registry.Register(escalatingTool{target: "claude-opus-4.1"})
+
+	firstInner := &mockProvider{turns: escalateThenAnswerTurns("unused")}
+	firstSession := &fakeTurnSession{inner: firstInner}
+	initial := &fakeTurnSessionProvider{session: firstSession}
+
+	switchedInner := &mockProvider{turns: singleAnswerTurns("answered on the switched session")}
+	switchedSession := &fakeTurnSession{inner: switchedInner}
+	switched := &fakeTurnSessionProvider{session: switchedSession}
+
+	var switchedTo string
+	legacyCalled := false
+	result, err := Run(context.Background(), "go", firstInner, Options{
+		Registry:            registry,
+		Model:               "claude-sonnet-4.5",
+		MaxTurns:            4,
+		TurnSessionProvider: initial,
+		ModelSessionSwitcher: func(_ context.Context, modelID string) (zeroruntime.TurnSessionProvider, error) {
+			switchedTo = modelID
+			return switched, nil
+		},
+		ModelSwitcher: func(_ context.Context, _ string) (Provider, error) {
+			legacyCalled = true
+			return nil, errors.New("legacy switcher must not be consulted")
+		},
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if result.FinalAnswer != "answered on the switched session" {
+		t.Fatalf("expected the switched session's answer, got %q", result.FinalAnswer)
+	}
+	if switchedTo != "claude-opus-4.1" {
+		t.Fatalf("expected switch to claude-opus-4.1, got %q", switchedTo)
+	}
+	if legacyCalled {
+		t.Fatal("legacy ModelSwitcher was consulted despite ModelSessionSwitcher being set")
+	}
+	if switched.opens != 1 {
+		t.Fatalf("expected exactly one open of the switched provider, got %d", switched.opens)
+	}
+	// The switched session carries the rest of the run: it streams the answer
+	// turn, is prewarmed exactly once at the swap, and is closed exactly once by
+	// the teardown defer.
+	if switchedSession.streams != 1 || len(switchedInner.requests) != 1 {
+		t.Fatalf("expected the switched session to stream the post-switch turn, streams=%d inner=%d", switchedSession.streams, len(switchedInner.requests))
+	}
+	if switchedSession.prewarms != 1 {
+		t.Fatalf("expected the switched session to be prewarmed exactly once, got %d", switchedSession.prewarms)
+	}
+	if switchedSession.closes != 1 {
+		t.Fatalf("expected the switched session to be closed exactly once at teardown, got %d", switchedSession.closes)
+	}
+	// The original session ends with the swap: one stream, one close.
+	if firstSession.streams != 1 || firstSession.closes != 1 {
+		t.Fatalf("expected the original session to end at the swap (streams=%d closes=%d)", firstSession.streams, firstSession.closes)
+	}
+}
+
+// TestRunModelSessionSwitcherOpenFailureIsNonFatal verifies the swap-time open
+// failure: the switcher returns a TurnSessionProvider whose OpenTurnSession
+// fails. The run stays non-fatal on the ORIGINAL session (closed only at
+// teardown), and the switch-failure note reaches the next request.
+func TestRunModelSessionSwitcherOpenFailureIsNonFatal(t *testing.T) {
+	registry := tools.NewRegistry()
+	registry.Register(escalatingTool{target: "claude-opus-4.1"})
+
+	inner := &mockProvider{turns: escalateThenAnswerTurns("recovered after failed open")}
+	session := &fakeTurnSession{inner: inner}
+	failingOpen := &fakeTurnSessionProvider{openErr: errors.New("switched handshake refused")}
+
+	result, err := Run(context.Background(), "go", inner, Options{
+		Registry:            registry,
+		Model:               "claude-sonnet-4.5",
+		MaxTurns:            4,
+		TurnSessionProvider: &fakeTurnSessionProvider{session: session},
+		ModelSessionSwitcher: func(_ context.Context, _ string) (zeroruntime.TurnSessionProvider, error) {
+			return failingOpen, nil
+		},
+	})
+	if err != nil {
+		t.Fatalf("expected a failed swap open to be non-fatal, got %v", err)
+	}
+	if result.FinalAnswer != "recovered after failed open" {
+		t.Fatalf("expected the run to recover on the original session, got %q", result.FinalAnswer)
+	}
+	if failingOpen.opens != 1 {
+		t.Fatalf("expected exactly one open attempt on the switched provider, got %d", failingOpen.opens)
+	}
+	// The original session serves both turns and is closed only at teardown —
+	// never by the failed swap.
+	if session.streams != 2 || session.closes != 1 {
+		t.Fatalf("expected the original session to serve both turns and close once (streams=%d closes=%d)", session.streams, session.closes)
+	}
+	var sawNote bool
+	for _, message := range inner.requests[1].Messages {
+		if message.Role == zeroruntime.MessageRoleUser && strings.Contains(strings.ToLower(message.Content), "could not switch") {
+			sawNote = true
+		}
+	}
+	if !sawNote {
+		t.Fatalf("expected a switch-failure note on the next turn, messages: %+v", inner.requests[1].Messages)
+	}
+}
+
+// TestRunModelSessionSwitcherErrorIsNonFatal verifies a ModelSessionSwitcher
+// error keeps the run on the current session with a transcript note — the same
+// non-fatal contract as ModelSwitcher.
+func TestRunModelSessionSwitcherErrorIsNonFatal(t *testing.T) {
+	registry := tools.NewRegistry()
+	registry.Register(escalatingTool{target: "claude-opus-4.1"})
+
+	inner := &mockProvider{turns: escalateThenAnswerTurns("recovered on original session")}
+	session := &fakeTurnSession{inner: inner}
+
+	result, err := Run(context.Background(), "go", inner, Options{
+		Registry:            registry,
+		Model:               "claude-sonnet-4.5",
+		MaxTurns:            4,
+		TurnSessionProvider: &fakeTurnSessionProvider{session: session},
+		ModelSessionSwitcher: func(_ context.Context, _ string) (zeroruntime.TurnSessionProvider, error) {
+			return nil, errors.New("session build blew up")
+		},
+	})
+	if err != nil {
+		t.Fatalf("expected session-switcher error to be non-fatal, got %v", err)
+	}
+	if result.FinalAnswer != "recovered on original session" {
+		t.Fatalf("expected the run to recover on the original session, got %q", result.FinalAnswer)
+	}
+	// Both turns stream through the original session; it closes once at teardown.
+	if session.streams != 2 || session.closes != 1 {
+		t.Fatalf("expected the original session to serve both turns and close once (streams=%d closes=%d)", session.streams, session.closes)
+	}
+	var sawNote bool
+	for _, request := range inner.requests[1].Messages {
+		if request.Role == zeroruntime.MessageRoleUser && strings.Contains(strings.ToLower(request.Content), "could not switch") {
+			sawNote = true
+		}
+	}
+	if !sawNote {
+		t.Fatalf("expected a non-fatal switch-failure note on the next turn, messages: %+v", inner.requests[1].Messages)
+	}
+}

internal/agent/types.go

@@ -283,6 +283,21 @@ type Options struct {
 	// request), so every existing caller is unaffected. A returned error is
 	// non-fatal: the run continues on the current model.
 	ModelSwitcher func(ctx context.Context, modelID string) (Provider, error)
+	// TurnSessionProvider, when set, supplies the turn session the run streams
+	// through — the seam an optimized provider session (connection reuse,
+	// prewarm, native compaction) plugs into without touching the loop. nil
+	// keeps the default: the loop wraps the passed provider in a no-op session
+	// whose Stream IS provider.StreamCompletion, so behavior is byte-identical
+	// and every existing caller is unaffected.
+	TurnSessionProvider zeroruntime.TurnSessionProvider
+	// ModelSessionSwitcher, when set, is the target-aware escalation hook: the
+	// loop prefers it over ModelSwitcher, and its TurnSessionProvider keeps an
+	// optimized session (and its capabilities) across a mid-run model switch.
+	// nil falls back to ModelSwitcher, whose bare Provider is wrapped in the
+	// default no-op session — today's behavior, unchanged. Same non-fatal error
+	// contract as ModelSwitcher: a returned error records a note and the run
+	// continues on the current model and session.
+	ModelSessionSwitcher func(ctx context.Context, modelID string) (zeroruntime.TurnSessionProvider, error)
 	// Trace, when set, records per-turn timing for the run: the loop stamps
 	// spans (prompt build, generation, tool execution, permission wait,
 	// compaction, provider connect) and counters (model requests, tool calls,

internal/cli/exec.go

@@ -412,6 +412,36 @@ func runExec(args []string, stdout io.Writer, stderr io.Writer, deps appDeps) in
 		}
 	}
 
+	// Optimized OpenAI turn sessions (ZERO_OPENAI_TURN_SESSION, default off).
+	// nil when gated off or the profile is ineligible: agent.Run then wraps the
+	// provider in its default adapter — the exact code path of today. The
+	// session switcher is installed only when the run START is optimized, so
+	// the legacy ModelSwitcher path above stays untouched otherwise.
+	turnSessions, _ := providers.OptimizedTurnSessions(resolved.Provider, provider, providers.Options{})
+	var modelSessionSwitcher func(context.Context, string) (zeroruntime.TurnSessionProvider, error)
+	if options.allowEscalation && turnSessions != nil {
+		modelSessionSwitcher = func(_ context.Context, modelID string) (zeroruntime.TurnSessionProvider, error) {
+			switchedProfile := resolved.Provider
+			switchedProfile.Model = modelID
+			switchedProvider, err := deps.newProvider(switchedProfile)
+			if err != nil {
+				return nil, err
+			}
+			if switchedProvider == nil {
+				// The loop treats a nil session source as "no swap" — mirror the
+				// legacy closure's (nil, nil) contract.
+				return nil, nil
+			}
+			currentModel = modelID
+			if optimized, ok := providers.OptimizedTurnSessions(switchedProfile, switchedProvider, providers.Options{}); ok {
+				return optimized, nil
+			}
+			// Ineligible switch target: default adapter, but with the switched
+			// model's resolved capability projection preserved.
+			return providers.DefaultTurnSessions(switchedProfile, switchedProvider, providers.Options{}), nil
+		}
+	}
+
 	runMetadata, err := resolveExecRunMetadata(resolved.Provider)
 	if err != nil {
 		return writeExecProviderError(stdout, stderr, options.outputFormat, "provider_error", err.Error())
@@ -572,29 +602,31 @@ func runExec(args []string, stdout io.Writer, stderr io.Writer, deps appDeps) in
 	hookDispatcher, hookSkip := newHookDispatcherWithExtra(workspaceRoot, pluginActivation.hooks, trustRoot)
 	emitTrustNotice(stderr, hookSkip, pluginActivation.trustSkip, mcpSkip)
 	result, err := agent.Run(runCtx, agentPrompt, provider, agent.Options{
-		MaxTurns:         resolved.MaxTurns,
-		ContextWindow:    resolveAgentContextWindow(runCtx, modelRegistry, resolved.Provider),
-		DeferThreshold:   effectiveDeferThreshold,
-		Specialists:      specialistRuntime.specialistInfos(),
-		Skills:           pluginActivation.skillInfos(deps.skillsDir()),
-		SessionID:        preparedSession.Session.SessionID,
-		CallingSessionID: options.callingSessionID,
-		CallingToolUseID: options.callingToolUseID,
-		Tag:              options.tag,
-		Depth:            options.depth,
-		SessionTitle:     sessionTitle,
-		ProviderName:     resolved.Provider.Name,
-		Model:            resolved.Provider.Model,
-		ModelSwitcher:    modelSwitcher,
-		ReasoningEffort:  forwardEffort,
-		Trace:            traceRecorder,
-		Cwd:              workspaceRoot,
-		Images:           images,
-		Registry:         registry,
-		PermissionMode:   permissionMode,
-		Autonomy:         options.autonomy,
-		SelfCorrect:      selfCorrector,
-		FileDiagnostics:  fileDiagnostics,
+		MaxTurns:             resolved.MaxTurns,
+		ContextWindow:        resolveAgentContextWindow(runCtx, modelRegistry, resolved.Provider),
+		DeferThreshold:       effectiveDeferThreshold,
+		Specialists:          specialistRuntime.specialistInfos(),
+		Skills:               pluginActivation.skillInfos(deps.skillsDir()),
+		SessionID:            preparedSession.Session.SessionID,
+		CallingSessionID:     options.callingSessionID,
+		CallingToolUseID:     options.callingToolUseID,
+		Tag:                  options.tag,
+		Depth:                options.depth,
+		SessionTitle:         sessionTitle,
+		ProviderName:         resolved.Provider.Name,
+		Model:                resolved.Provider.Model,
+		ModelSwitcher:        modelSwitcher,
+		TurnSessionProvider:  turnSessions,
+		ModelSessionSwitcher: modelSessionSwitcher,
+		ReasoningEffort:      forwardEffort,
+		Trace:                traceRecorder,
+		Cwd:                  workspaceRoot,
+		Images:               images,
+		Registry:             registry,
+		PermissionMode:       permissionMode,
+		Autonomy:             options.autonomy,
+		SelfCorrect:          selfCorrector,
+		FileDiagnostics:      fileDiagnostics,
 		// Headless exec: don't accept a no-tool-call turn as "done" while work
 		// clearly remains (pending plan items / a mid-step continuation cue) —
 		// nudge to continue, and finalize as INCOMPLETE rather than false success

internal/cli/exec_turn_session_test.go

@@ -0,0 +1,125 @@
+package cli
+
+import (
+	"bytes"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/Gitlawb/zero/internal/config"
+	"github.com/Gitlawb/zero/internal/providers/openai"
+	"github.com/Gitlawb/zero/internal/zeroruntime"
+)
+
+// TestRunExecOptimizedSessionUnderGate proves the end-to-end wiring: with
+// ZERO_OPENAI_TURN_SESSION on and a real official-OpenAI provider, a headless
+// run streams through the optimized session — the server sees the prewarm HEAD
+// probe in addition to the turn's POST.
+func TestRunExecOptimizedSessionUnderGate(t *testing.T) {
+	t.Setenv("XDG_DATA_HOME", t.TempDir())
+	t.Setenv("ZERO_OPENAI_TURN_SESSION", "1")
+	cwd := t.TempDir()
+
+	var heads, posts atomic.Int64
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		switch {
+		case r.Method == http.MethodHead:
+			heads.Add(1)
+			w.WriteHeader(http.StatusMethodNotAllowed)
+		case r.Method == http.MethodPost && strings.HasSuffix(r.URL.Path, "/chat/completions"):
+			posts.Add(1)
+			w.Header().Set("Content-Type", "text/event-stream")
+			_, _ = w.Write([]byte("data: {\"choices\":[{\"delta\":{\"content\":\"done\"}}]}\n\n"))
+			_, _ = w.Write([]byte("data: [DONE]\n\n"))
+		default:
+			w.WriteHeader(http.StatusNotFound)
+		}
+	}))
+	defer server.Close()
+
+	var stdout, stderr bytes.Buffer
+	exitCode := runWithDeps([]string{"exec", "say done"}, &stdout, &stderr, appDeps{
+		getwd: func() (string, error) { return cwd, nil },
+		resolveConfig: func(string, config.Overrides) (config.ResolvedConfig, error) {
+			return config.ResolvedConfig{
+				ActiveProvider: "official-openai",
+				Provider: config.ProviderProfile{
+					Name:         "official-openai",
+					ProviderKind: config.ProviderKindOpenAI,
+					BaseURL:      server.URL,
+					APIKey:       "sk-exec-test",
+					Model:        "pr8-exec-model",
+				},
+			}, nil
+		},
+		newProvider: func(profile config.ProviderProfile) (zeroruntime.Provider, error) {
+			// Pin a keep-alive transport so the probe fires on every platform:
+			// the shared default transport disables keep-alives on macOS, where
+			// the session correctly skips the prewarm.
+			transport := http.DefaultTransport.(*http.Transport).Clone()
+			transport.DisableKeepAlives = false
+			return openai.New(openai.Options{
+				APIKey:     profile.APIKey,
+				BaseURL:    profile.BaseURL,
+				Model:      profile.Model,
+				HTTPClient: &http.Client{Transport: transport},
+			})
+		},
+	})
+	if exitCode != exitSuccess {
+		t.Fatalf("exitCode = %d stdout=%s stderr=%s", exitCode, stdout.String(), stderr.String())
+	}
+	if got := posts.Load(); got < 1 {
+		t.Fatalf("server saw %d chat-completions POSTs, want >= 1", got)
+	}
+	// The prewarm probe is asynchronous; it fires well before the model turn
+	// finishes, but poll rather than assuming scheduling order. The deadline
+	// exceeds the production prewarmTimeout (3s) so a slow probe cannot flake
+	// this assertion.
+	deadline := time.Now().Add(5 * time.Second)
+	for heads.Load() == 0 && time.Now().Before(deadline) {
+		time.Sleep(10 * time.Millisecond)
+	}
+	if got := heads.Load(); got != 1 {
+		t.Fatalf("server saw %d prewarm HEAD probes, want exactly 1", got)
+	}
+}
+
+// TestRunExecGateOnFallsBackForFakeProvider proves fallback safety: with the
+// gate on but a provider that is not the concrete *openai.Provider, the run
+// proceeds on the default path exactly as today.
+func TestRunExecGateOnFallsBackForFakeProvider(t *testing.T) {
+	t.Setenv("XDG_DATA_HOME", t.TempDir())
+	t.Setenv("ZERO_OPENAI_TURN_SESSION", "1")
+	cwd := t.TempDir()
+
+	var builds int
+	var stdout, stderr bytes.Buffer
+	exitCode := runWithDeps([]string{"exec", "--model", "claude-haiku-4.5", "hello"}, &stdout, &stderr, appDeps{
+		getwd: func() (string, error) { return cwd, nil },
+		resolveConfig: func(_ string, overrides config.Overrides) (config.ResolvedConfig, error) {
+			model := "claude-haiku-4.5"
+			if overrides.Provider.Model != "" {
+				model = overrides.Provider.Model
+			}
+			cfg := execResolvedConfig()
+			cfg.Provider.ProviderKind = config.ProviderKindAnthropic
+			cfg.Provider.Model = model
+			cfg.MaxTurns = 3
+			return cfg, nil
+		},
+		newProvider: func(config.ProviderProfile) (zeroruntime.Provider, error) {
+			builds++
+			return &escalatingExecProvider{}, nil
+		},
+	})
+	if exitCode != exitSuccess {
+		t.Fatalf("exitCode = %d stdout=%s stderr=%s", exitCode, stdout.String(), stderr.String())
+	}
+	if builds != 1 {
+		t.Fatalf("newProvider called %d times, want 1 (default path, no optimized session)", builds)
+	}
+}

internal/providers/factory.go

@@ -108,6 +108,57 @@ func New(profile config.ProviderProfile, options Options) (zeroruntime.Provider,
 	}
 }
 
+// NewTurnSessionProvider builds the default TurnSessionProvider for a resolved
+// profile: it constructs the provider exactly as New does, then wraps it with a
+// ProviderCapabilities projection computed from the same model-registry entry
+// New already resolves. The default session's Stream is the provider's
+// StreamCompletion, so runtime behavior is identical to using New directly.
+func NewTurnSessionProvider(profile config.ProviderProfile, options Options) (zeroruntime.TurnSessionProvider, error) {
+	provider, err := New(profile, options)
+	if err != nil {
+		return nil, err
+	}
+	caps, err := resolveCapabilities(profile, options)
+	if err != nil {
+		return nil, err
+	}
+	return zeroruntime.NewProviderTurnSessionProvider(provider, caps), nil
+}
+
+// resolveCapabilities projects the resolved profile's model-registry entry into
+// the flat zeroruntime.ProviderCapabilities (zeroruntime cannot reference
+// modelregistry types — modelregistry imports zeroruntime, so a typed field
+// would form an import cycle). A model absent from the registry yields only the
+// resolved API model id and max-output tokens; the rest stays zero (unknown).
+func resolveCapabilities(profile config.ProviderProfile, options Options) (zeroruntime.ProviderCapabilities, error) {
+	resolved, err := resolveProfile(profile, options)
+	if err != nil {
+		return zeroruntime.ProviderCapabilities{}, err
+	}
+	caps := zeroruntime.ProviderCapabilities{
+		Model:           resolved.apiModel,
+		MaxOutputTokens: resolved.maxOutputTokens,
+	}
+	registry, err := defaultRegistry(options.ModelRegistry)
+	if err != nil {
+		return zeroruntime.ProviderCapabilities{}, err
+	}
+	if entry, ok := registry.Get(strings.TrimSpace(profile.Model)); ok {
+		caps.ContextWindow = entry.ContextLimits.ContextWindow
+		caps.SupportsVision = entry.Supports(modelregistry.ModelCapabilityVision)
+		caps.SupportsReasoning = entry.Supports(modelregistry.ModelCapabilityReasoning)
+		caps.SupportsPromptCache = entry.Supports(modelregistry.ModelCapabilityPromptCache)
+		// Read efforts through Registry.ReasoningEfforts (not the raw entry) so
+		// the projection matches what the /effort picker and the run-time
+		// resolver advertise — including the name-based fallback for catalog
+		// entries that enumerate no efforts of their own.
+		for _, effort := range registry.ReasoningEfforts(entry.ID) {
+			caps.ReasoningEfforts = append(caps.ReasoningEfforts, string(effort))
+		}
+	}
+	return caps, nil
+}
+
 func parseThinkTagsForProfile(profile config.ProviderProfile, resolved resolvedProfile) bool {
 	if profile.ParseThinkTags != nil {
 		return *profile.ParseThinkTags

internal/providers/factory_turn_session_test.go

@@ -0,0 +1,194 @@
+package providers
+
+import (
+	"context"
+	"errors"
+	"testing"
+
+	"github.com/Gitlawb/zero/internal/config"
+	"github.com/Gitlawb/zero/internal/modelregistry"
+	"github.com/Gitlawb/zero/internal/zeroruntime"
+)
+
+// TestNewTurnSessionProviderForEveryKind asserts every current provider kind
+// wraps into the default TurnSessionProvider: construction succeeds, a session
+// opens, and Compact reports unsupported (the default adapter contract).
+func TestNewTurnSessionProviderForEveryKind(t *testing.T) {
+	kinds := []config.ProviderKind{
+		config.ProviderKindOpenAI,
+		config.ProviderKindOpenAICompatible,
+		config.ProviderKindAnthropic,
+		config.ProviderKindAnthropicCompat,
+		config.ProviderKindGoogle,
+	}
+	for _, kind := range kinds {
+		t.Run(string(kind), func(t *testing.T) {
+			tsp, err := NewTurnSessionProvider(config.ProviderProfile{
+				Name:         "pr7-" + string(kind),
+				ProviderKind: kind,
+				BaseURL:      "https://provider.example/v1",
+				APIKey:       "sk-turn-session-test",
+				// A model absent from the registry: resolution falls back to the
+				// raw id, so this test stays hermetic across catalog changes.
+				Model: "pr7-unregistered-model",
+			}, Options{UserAgent: "zero-turn-session-test"})
+			if err != nil {
+				t.Fatalf("NewTurnSessionProvider(%s): %v", kind, err)
+			}
+			session, err := tsp.OpenTurnSession(context.Background())
+			if err != nil {
+				t.Fatalf("OpenTurnSession(%s): %v", kind, err)
+			}
+			defer func() {
+				if closeErr := session.Close(); closeErr != nil {
+					t.Fatalf("Close(%s): %v", kind, closeErr)
+				}
+			}()
+			if err := session.Prewarm(context.Background()); err != nil {
+				t.Fatalf("Prewarm(%s): %v", kind, err)
+			}
+			if _, compactErr := session.Compact(context.Background(), zeroruntime.CompletionRequest{}); !errors.Is(compactErr, zeroruntime.ErrCompactionUnsupported) {
+				t.Fatalf("Compact(%s) = %v, want ErrCompactionUnsupported", kind, compactErr)
+			}
+			caps := tsp.Capabilities()
+			if caps.Model != "pr7-unregistered-model" {
+				t.Fatalf("Capabilities(%s).Model = %q, want the resolved api model", kind, caps.Model)
+			}
+			if caps.NativeCompaction {
+				t.Fatalf("Capabilities(%s).NativeCompaction = true, want false for the default adapter", kind)
+			}
+		})
+	}
+}
+
+// TestNewTurnSessionProviderProjectsRegistryCapabilities asserts the factory
+// projects the model-registry entry (context limits, capability flags,
+// reasoning efforts) into the flat ProviderCapabilities.
+func TestNewTurnSessionProviderProjectsRegistryCapabilities(t *testing.T) {
+	registry, err := modelregistry.NewRegistry([]modelregistry.ModelEntry{{
+		ID:          "pr7-caps-model",
+		DisplayName: "PR7 Capability Probe",
+		APIModel:    "pr7-caps-api-model",
+		Provider:    modelregistry.ProviderOpenAI,
+		ContextLimits: modelregistry.ContextLimits{
+			ContextWindow:   200_000,
+			MaxOutputTokens: 64_000,
+		},
+		ReasoningEfforts: []modelregistry.ReasoningEffort{
+			modelregistry.ReasoningEffortLow,
+			modelregistry.ReasoningEffortMedium,
+			modelregistry.ReasoningEffortHigh,
+		},
+		Capabilities: modelregistry.ModelCapabilities{
+			modelregistry.ModelCapabilityChat,
+			modelregistry.ModelCapabilityVision,
+			modelregistry.ModelCapabilityReasoning,
+			modelregistry.ModelCapabilityPromptCache,
+		},
+		Cost: modelregistry.ModelCost{
+			Currency:           "USD",
+			Unit:               "per_1m_tokens",
+			InputPerMillion:    1,
+			OutputPerMillion:   2,
+			Source:             "https://provider.example/pricing (test fixture)",
+			SourceLastVerified: "2026-07-18",
+		},
+		Status:  modelregistry.ModelStatusActive,
+		Aliases: []string{"pr7-caps"},
+	}})
+	if err != nil {
+		t.Fatalf("NewRegistry: %v", err)
+	}
+
+	tsp, err := NewTurnSessionProvider(config.ProviderProfile{
+		Name:         "pr7-caps",
+		ProviderKind: config.ProviderKindOpenAI,
+		BaseURL:      "https://provider.example/v1",
+		APIKey:       "sk-turn-session-test",
+		Model:        "pr7-caps-model",
+	}, Options{
+		UserAgent:     "zero-turn-session-test",
+		ModelRegistry: &registry,
+	})
+	if err != nil {
+		t.Fatalf("NewTurnSessionProvider: %v", err)
+	}
+
+	caps := tsp.Capabilities()
+	if caps.Model != "pr7-caps-api-model" {
+		t.Fatalf("Model = %q, want the registry APIModel", caps.Model)
+	}
+	if caps.ContextWindow != 200_000 || caps.MaxOutputTokens != 64_000 {
+		t.Fatalf("limits = (%d, %d), want (200000, 64000)", caps.ContextWindow, caps.MaxOutputTokens)
+	}
+	if !caps.SupportsVision || !caps.SupportsReasoning || !caps.SupportsPromptCache {
+		t.Fatalf("capability flags = %+v, want vision/reasoning/prompt-cache all true", caps)
+	}
+	if len(caps.ReasoningEfforts) != 3 || caps.ReasoningEfforts[0] != "low" || caps.ReasoningEfforts[2] != "high" {
+		t.Fatalf("ReasoningEfforts = %v, want [low medium high]", caps.ReasoningEfforts)
+	}
+	if caps.NativeCompaction {
+		t.Fatal("NativeCompaction = true, want false for the default adapter")
+	}
+}
+
+// TestNewTurnSessionProviderUsesEffectiveReasoningEfforts asserts the
+// projection reads efforts through Registry.ReasoningEfforts, so a catalog
+// entry that enumerates no efforts of its own still reports the name-inferred
+// effective tiers the /effort picker and run-time resolver advertise.
+func TestNewTurnSessionProviderUsesEffectiveReasoningEfforts(t *testing.T) {
+	registry, err := modelregistry.NewRegistry([]modelregistry.ModelEntry{{
+		// A gpt-5-family id with NO ReasoningEfforts listed: the effective
+		// efforts come from name inference, differing from the raw entry.
+		ID:          "gpt-5-pr7-probe",
+		DisplayName: "PR7 Effective Efforts Probe",
+		APIModel:    "gpt-5-pr7-probe-api",
+		Provider:    modelregistry.ProviderOpenAI,
+		ContextLimits: modelregistry.ContextLimits{
+			ContextWindow:   400_000,
+			MaxOutputTokens: 128_000,
+		},
+		Capabilities: modelregistry.ModelCapabilities{
+			modelregistry.ModelCapabilityChat,
+			modelregistry.ModelCapabilityReasoning,
+		},
+		Cost: modelregistry.ModelCost{
+			Currency:           "USD",
+			Unit:               "per_1m_tokens",
+			InputPerMillion:    1,
+			OutputPerMillion:   2,
+			Source:             "https://provider.example/pricing (test fixture)",
+			SourceLastVerified: "2026-07-18",
+		},
+		Status:  modelregistry.ModelStatusActive,
+		Aliases: []string{"pr7-effective-efforts"},
+	}})
+	if err != nil {
+		t.Fatalf("NewRegistry: %v", err)
+	}
+
+	tsp, err := NewTurnSessionProvider(config.ProviderProfile{
+		Name:         "pr7-effective",
+		ProviderKind: config.ProviderKindOpenAI,
+		BaseURL:      "https://provider.example/v1",
+		APIKey:       "sk-turn-session-test",
+		Model:        "gpt-5-pr7-probe",
+	}, Options{
+		UserAgent:     "zero-turn-session-test",
+		ModelRegistry: &registry,
+	})
+	if err != nil {
+		t.Fatalf("NewTurnSessionProvider: %v", err)
+	}
+
+	got := tsp.Capabilities().ReasoningEfforts
+	want := []string{"minimal", "low", "medium", "high"}
+	if len(got) != len(want) {
+		t.Fatalf("ReasoningEfforts = %v, want %v (name-inferred fallback)", got, want)
+	}
+	for i := range want {
+		if got[i] != want[i] {
+			t.Fatalf("ReasoningEfforts[%d] = %q, want %q (full: %v)", i, got[i], want[i], got)
+		}
+	}
+}

internal/providers/openai/session.go

@@ -0,0 +1,205 @@
+package openai
+
+import (
+	"context"
+	"crypto/sha256"
+	"encoding/hex"
+	"encoding/json"
+	"fmt"
+	"net/http"
+	"strings"
+	"sync"
+	"time"
+
+	"github.com/Gitlawb/zero/internal/trace"
+	"github.com/Gitlawb/zero/internal/zeroruntime"
+)
+
+// prewarmTimeout bounds the best-effort prewarm probe. The probe runs in the
+// background, so this cap only limits how long the goroutine may linger — it
+// can never delay the first turn.
+const prewarmTimeout = 3 * time.Second
+
+// NewTurnSessionProvider wraps an already-constructed *Provider in the
+// optimized OpenAI turn session: a best-effort connection prewarm at run start
+// plus request-prefix stability telemetry. Stream is Provider.StreamCompletion
+// verbatim, so runtime request behavior is identical to the default adapter —
+// the session adds observation and pool priming, never a different request.
+func NewTurnSessionProvider(provider *Provider, caps zeroruntime.ProviderCapabilities) zeroruntime.TurnSessionProvider {
+	return &turnSessionProvider{provider: provider, caps: caps}
+}
+
+type turnSessionProvider struct {
+	provider *Provider
+	caps     zeroruntime.ProviderCapabilities
+}
+
+func (p *turnSessionProvider) OpenTurnSession(context.Context) (zeroruntime.TurnSession, error) {
+	return &turnSession{provider: p.provider}, nil
+}
+
+func (p *turnSessionProvider) Capabilities() zeroruntime.ProviderCapabilities {
+	return p.caps
+}
+
+// turnSession is one run's optimized OpenAI session. The agent loop serializes
+// all provider I/O through its session shim, so fields need no mutex. The
+// fingerprint counters exist so a future stateful-reuse session (Responses API)
+// inherits a proven prefix-stability detector; on chat completions there is no
+// server-side response state to invalidate, so drift is telemetry only.
+type turnSession struct {
+	provider        *Provider
+	lastFingerprint string
+	prewarmOnce     sync.Once
+	// prewarmDone closes when the background probe settles; tests wait on it.
+	prewarmDone chan struct{}
+}
+
+// Prewarm launches one bounded, unauthenticated HEAD probe to the provider's
+// base URL in the background so the TCP+TLS handshake lands in the shared
+// connection pool while the loop assembles the first prompt. One attempt, no
+// retries, no bearer token — the goal is the handshake, not a 2xx (a 401/404/
+// 405 response still primes the pool). Always returns nil: prewarm is advisory
+// by contract and never required for correctness.
+//
+// The probe is issued directly through the client (NOT providerio.SendWithRetry)
+// so it stamps ONLY the provider_prewarm span — a nested provider_connect stamp
+// would contaminate the exact A/B metric this feature is measured by.
+//
+// When the client's transport cannot retain idle connections (macOS disables
+// keep-alives on the shared transport because degraded pooled connections are
+// indistinguishable from backend slowness there), the probe is skipped entirely
+// rather than spending a request that can never be reused. The pool's 30s idle
+// timeout bounds the warm window; a slower start simply falls back to a cold
+// dial, never worse than today.
+func (s *turnSession) Prewarm(ctx context.Context) error {
+	s.prewarmOnce.Do(func() {
+		done := make(chan struct{})
+		s.prewarmDone = done
+		provider := s.provider
+		if !transportRetainsIdleConns(provider.httpClient) {
+			close(done)
+			return
+		}
+		recorder := trace.FromContext(ctx)
+		go func() {
+			defer close(done)
+			probeCtx, cancel := context.WithTimeout(ctx, prewarmTimeout)
+			defer cancel()
+			span := recorder.Span(trace.SpanProviderPrewarm)
+			defer span.End()
+			recorder.Counter(trace.CounterPrewarmAttempts, 1)
+			request, err := http.NewRequestWithContext(probeCtx, http.MethodHead, provider.baseURL, nil)
+			if err != nil {
+				return
+			}
+			if provider.userAgent != "" {
+				request.Header.Set("User-Agent", provider.userAgent)
+			}
+			response, err := provider.httpClient.Do(request)
+			if err == nil {
+				_ = response.Body.Close()
+			}
+		}()
+	})
+	return nil
+}
+
+// transportRetainsIdleConns reports whether a warmed connection can outlive the
+// probe. A *http.Transport with DisableKeepAlives (the shared transport on
+// macOS) closes every connection after its request, so a prewarm there is pure
+// overhead. A nil transport is net/http's default (keep-alives on); an unknown
+// custom transport is assumed reusable — the probe stays a harmless best-effort.
+func transportRetainsIdleConns(client *http.Client) bool {
+	if client == nil {
+		return false
+	}
+	if transport, ok := client.Transport.(*http.Transport); ok {
+		return !transport.DisableKeepAlives
+	}
+	return true
+}
+
+func (s *turnSession) Stream(ctx context.Context, request zeroruntime.CompletionRequest) (<-chan zeroruntime.StreamEvent, error) {
+	s.observePrefix(ctx, request)
+	return s.provider.StreamCompletion(ctx, request)
+}
+
+func (s *turnSession) Compact(context.Context, zeroruntime.CompletionRequest) ([]zeroruntime.Message, error) {
+	return nil, zeroruntime.ErrCompactionUnsupported
+}
+
+// Close is an idempotent no-op: the shared transport pool owns the connections.
+func (s *turnSession) Close() error { return nil }
+
+// observePrefix tracks whether the request-prefix parameters stayed stable
+// between this session's streams. The first stream seeds the fingerprint and
+// counts as neither stable nor drift.
+func (s *turnSession) observePrefix(ctx context.Context, request zeroruntime.CompletionRequest) {
+	fingerprint := s.computeFingerprint(request)
+	if s.lastFingerprint == "" {
+		s.lastFingerprint = fingerprint
+		return
+	}
+	recorder := trace.FromContext(ctx)
+	if fingerprint == s.lastFingerprint {
+		recorder.Counter(trace.CounterPrefixStable, 1)
+		return
+	}
+	recorder.Counter(trace.CounterPrefixDrift, 1)
+	s.lastFingerprint = fingerprint
+}
+
+// computeFingerprint digests the wire-affecting request parameters: the
+// session's model and max-tokens (fixed at construction — CompletionRequest
+// carries no model field), the normalized reasoning effort as it would appear
+// on the wire, the prompt-cache key (constant within a session, and part of
+// the request the provider serializes), and an order-preserving digest of the
+// advertised tools (request serialization preserves tool order, so a reorder
+// changes the wire bytes and counts as drift). Messages are excluded — they
+// grow every turn; this fingerprints request parameters, not conversation
+// content.
+//
+// SCOPE: this is per-provider request-parameter TELEMETRY, not a complete
+// compatibility fingerprint. It deliberately does not incorporate the prompt
+// prefix hashes (base instructions / project context / skills — the
+// complete_prefix signal the agent loop already emits per turn). Any future
+// stateful response reuse MUST combine both signals plus the remaining
+// wire-affecting fields before reusing provider-side state; these counters
+// alone are not sufficient compatibility protection.
+func (s *turnSession) computeFingerprint(request zeroruntime.CompletionRequest) string {
+	var builder strings.Builder
+	builder.WriteString(s.provider.model)
+	builder.WriteByte('|')
+	fmt.Fprintf(&builder, "%d", s.provider.maxTokens)
+	builder.WriteByte('|')
+	builder.WriteString(openAIReasoningEffort(request.ReasoningEffort))
+	builder.WriteByte('|')
+	builder.WriteString(request.PromptCacheKey)
+	builder.WriteByte('|')
+	builder.WriteString(wireToolsDigest(request.Tools))
+	sum := sha256.Sum256([]byte(builder.String()))
+	return hex.EncodeToString(sum[:])
+}
+
+// wireToolsDigest renders the tool set in WIRE ORDER: each tool as name +
+// description + its JSON-marshaled parameters (encoding/json sorts map keys,
+// so each render is stable). Order is preserved deliberately — the OpenAI
+// request serializes tools in the order given, so a reorder changes the
+// request bytes and must count as drift. Descriptions are model-visible
+// request bytes too. A schema that fails to marshal is recorded under a
+// stable sentinel — this digest feeds telemetry counters, so cruder handling
+// of that pathological case is acceptable.
+func wireToolsDigest(tools []zeroruntime.ToolDefinition) string {
+	rendered := make([]string, 0, len(tools))
+	for _, tool := range tools {
+		schema, err := json.Marshal(tool.Parameters)
+		if err != nil {
+			rendered = append(rendered, tool.Name+"\n"+tool.Description+"\n__non_json:"+tool.Name)
+			continue
+		}
+		rendered = append(rendered, tool.Name+"\n"+tool.Description+"\n"+string(schema))
+	}
+	sum := sha256.Sum256([]byte(strings.Join(rendered, "\x00")))
+	return hex.EncodeToString(sum[:])
+}

internal/providers/openai/session_test.go

@@ -0,0 +1,387 @@
+package openai
+
+import (
+	"context"
+	"errors"
+	"net/http"
+	"net/http/httptest"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/Gitlawb/zero/internal/trace"
+	"github.com/Gitlawb/zero/internal/zeroruntime"
+)
+
+// keepAliveClient returns a client whose transport retains idle connections on
+// EVERY platform. The shared default transport disables keep-alives on macOS,
+// where the session correctly skips the probe — tests that assert the probe
+// fires must therefore pin a reusable transport to stay platform-deterministic.
+func keepAliveClient() *http.Client {
+	transport := http.DefaultTransport.(*http.Transport).Clone()
+	transport.DisableKeepAlives = false
+	return &http.Client{Transport: transport}
+}
+
+// newPrewarmTestProvider builds a provider whose transport always allows the
+// prewarm probe, regardless of host platform.
+func newPrewarmTestProvider(t *testing.T, handler http.HandlerFunc) *Provider {
+	t.Helper()
+	return newTestProviderWithOptions(t, Options{APIKey: "sk-test", HTTPClient: keepAliveClient()}, handler)
+}
+
+func waitPrewarmDone(t *testing.T, session *turnSession) {
+	t.Helper()
+	if session.prewarmDone == nil {
+		t.Fatal("prewarm was never launched")
+	}
+	select {
+	case <-session.prewarmDone:
+	case <-time.After(10 * time.Second):
+		t.Fatal("prewarm probe did not settle in time")
+	}
+}
+
+func openOptimizedSession(t *testing.T, provider *Provider) *turnSession {
+	t.Helper()
+	session, err := NewTurnSessionProvider(provider, zeroruntime.ProviderCapabilities{}).OpenTurnSession(context.Background())
+	if err != nil {
+		t.Fatalf("OpenTurnSession: %v", err)
+	}
+	return session.(*turnSession)
+}
+
+func TestTurnSessionPrewarmSendsSingleHEAD(t *testing.T) {
+	var requests atomic.Int64
+	var lastMethod atomic.Value
+	provider := newPrewarmTestProvider(t, func(w http.ResponseWriter, r *http.Request) {
+		requests.Add(1)
+		lastMethod.Store(r.Method)
+		w.WriteHeader(http.StatusMethodNotAllowed)
+	})
+
+	session := openOptimizedSession(t, provider)
+	if err := session.Prewarm(context.Background()); err != nil {
+		t.Fatalf("Prewarm: %v", err)
+	}
+	waitPrewarmDone(t, session)
+
+	if got := requests.Load(); got != 1 {
+		t.Fatalf("prewarm sent %d requests, want exactly 1 (no retries even on 405)", got)
+	}
+	if method, _ := lastMethod.Load().(string); method != http.MethodHead {
+		t.Fatalf("prewarm method = %q, want HEAD", method)
+	}
+
+	// A second Prewarm (e.g. after a mid-run swap re-open) must not re-probe.
+	if err := session.Prewarm(context.Background()); err != nil {
+		t.Fatalf("second Prewarm: %v", err)
+	}
+	waitPrewarmDone(t, session)
+	if got := requests.Load(); got != 1 {
+		t.Fatalf("second Prewarm re-probed: %d requests, want 1", got)
+	}
+}
+
+func TestTurnSessionPrewarmNonFatalOnConnectError(t *testing.T) {
+	// A server that is immediately closed leaves a port that refuses connections.
+	server := httptest.NewServer(http.HandlerFunc(func(http.ResponseWriter, *http.Request) {}))
+	deadURL := server.URL
+	server.Close()
+
+	provider, err := New(Options{APIKey: "sk-test", BaseURL: deadURL, Model: "gpt-test", HTTPClient: keepAliveClient()})
+	if err != nil {
+		t.Fatalf("New: %v", err)
+	}
+	session := openOptimizedSession(t, provider)
+	if err := session.Prewarm(context.Background()); err != nil {
+		t.Fatalf("Prewarm must be advisory even when the probe cannot connect, got %v", err)
+	}
+	waitPrewarmDone(t, session)
+}
+
+func TestTurnSessionPrewarmTraceStamps(t *testing.T) {
+	provider := newPrewarmTestProvider(t, func(w http.ResponseWriter, _ *http.Request) {
+		w.WriteHeader(http.StatusMethodNotAllowed)
+	})
+	recorder := trace.NewRecorder("session-test", "run-1", "test")
+	recorder.Start()
+	ctx := trace.WithContext(context.Background(), recorder)
+
+	session := openOptimizedSession(t, provider)
+	if err := session.Prewarm(ctx); err != nil {
+		t.Fatalf("Prewarm: %v", err)
+	}
+	waitPrewarmDone(t, session)
+
+	tr := recorder.Finish()
+	if got := tr.Counter(trace.CounterPrewarmAttempts); got != 1 {
+		t.Fatalf("prewarm_attempts = %d, want 1", got)
+	}
+	var sawSpan bool
+	for _, span := range tr.Spans {
+		if span.Name == trace.SpanProviderPrewarm {
+			sawSpan = true
+		}
+	}
+	if !sawSpan {
+		t.Fatalf("missing %s span in %+v", trace.SpanProviderPrewarm, tr.Spans)
+	}
+}
+
+func TestTurnSessionStreamDelegatesToProvider(t *testing.T) {
+	handler := func(w http.ResponseWriter, r *http.Request) {
+		w.Header().Set("Content-Type", "text/event-stream")
+		_, _ = w.Write([]byte("data: {\"choices\":[{\"delta\":{\"content\":\"hello\"}}]}\n\n"))
+		_, _ = w.Write([]byte("data: [DONE]\n\n"))
+	}
+
+	direct := newTestProvider(t, handler)
+	viaSession := newTestProvider(t, handler)
+	session := openOptimizedSession(t, viaSession)
+
+	request := zeroruntime.CompletionRequest{
+		Messages: []zeroruntime.Message{{Role: zeroruntime.MessageRoleUser, Content: "hi"}},
+	}
+	collect := func(stream <-chan zeroruntime.StreamEvent) []zeroruntime.StreamEvent {
+		var events []zeroruntime.StreamEvent
+		for event := range stream {
+			events = append(events, event)
+		}
+		return events
+	}
+
+	directStream, err := direct.StreamCompletion(context.Background(), request)
+	if err != nil {
+		t.Fatalf("direct StreamCompletion: %v", err)
+	}
+	sessionStream, err := session.Stream(context.Background(), request)
+	if err != nil {
+		t.Fatalf("session Stream: %v", err)
+	}
+
+	directEvents := collect(directStream)
+	sessionEvents := collect(sessionStream)
+	if len(directEvents) != len(sessionEvents) {
+		t.Fatalf("event counts diverged: direct=%d session=%d", len(directEvents), len(sessionEvents))
+	}
+	for i := range directEvents {
+		if directEvents[i].Type != sessionEvents[i].Type || directEvents[i].Content != sessionEvents[i].Content {
+			t.Fatalf("event %d diverged: direct=%+v session=%+v", i, directEvents[i], sessionEvents[i])
+		}
+	}
+}
+
+func TestTurnSessionFingerprintIgnoresMessages(t *testing.T) {
+	provider := newTestProvider(t, func(http.ResponseWriter, *http.Request) {})
+	session := openOptimizedSession(t, provider)
+
+	tools := []zeroruntime.ToolDefinition{{Name: "read_file", Parameters: map[string]any{"type": "object"}}}
+	base := session.computeFingerprint(zeroruntime.CompletionRequest{
+		Messages:       []zeroruntime.Message{{Role: zeroruntime.MessageRoleUser, Content: "turn one"}},
+		Tools:          tools,
+		PromptCacheKey: "session-a",
+	})
+	grown := session.computeFingerprint(zeroruntime.CompletionRequest{
+		Messages: []zeroruntime.Message{
+			{Role: zeroruntime.MessageRoleUser, Content: "turn one"},
+			{Role: zeroruntime.MessageRoleAssistant, Content: "reply"},
+			{Role: zeroruntime.MessageRoleUser, Content: "turn two"},
+		},
+		Tools:          tools,
+		PromptCacheKey: "session-a",
+	})
+	if base != grown {
+		t.Fatal("fingerprint changed when only messages grew — conversation content must not be an input")
+	}
+}
+
+func TestTurnSessionFingerprintDriftOnCacheKey(t *testing.T) {
+	provider := newTestProvider(t, func(http.ResponseWriter, *http.Request) {})
+	session := openOptimizedSession(t, provider)
+
+	tools := []zeroruntime.ToolDefinition{{Name: "read_file", Parameters: map[string]any{"type": "object"}}}
+	base := session.computeFingerprint(zeroruntime.CompletionRequest{Tools: tools, PromptCacheKey: "session-a"})
+	changed := session.computeFingerprint(zeroruntime.CompletionRequest{Tools: tools, PromptCacheKey: "session-b"})
+	if base == changed {
+		t.Fatal("fingerprint did not drift on a changed prompt-cache key — it is a serialized request field")
+	}
+}
+
+func TestTurnSessionFingerprintDriftOnToolsAndEffort(t *testing.T) {
+	provider := newTestProvider(t, func(http.ResponseWriter, *http.Request) {})
+	session := openOptimizedSession(t, provider)
+
+	base := session.computeFingerprint(zeroruntime.CompletionRequest{
+		Tools: []zeroruntime.ToolDefinition{{Name: "read_file", Parameters: map[string]any{"type": "object"}}},
+	})
+	changedTools := session.computeFingerprint(zeroruntime.CompletionRequest{
+		Tools: []zeroruntime.ToolDefinition{{Name: "write_file", Parameters: map[string]any{"type": "object"}}},
+	})
+	if base == changedTools {
+		t.Fatal("fingerprint did not drift on a changed tool set")
+	}
+	changedEffort := session.computeFingerprint(zeroruntime.CompletionRequest{
+		Tools:           []zeroruntime.ToolDefinition{{Name: "read_file", Parameters: map[string]any{"type": "object"}}},
+		ReasoningEffort: "high",
+	})
+	if base == changedEffort {
+		t.Fatal("fingerprint did not drift on a changed reasoning effort")
+	}
+	// Unrecognized efforts normalize to omitted — same as empty.
+	droppedEffort := session.computeFingerprint(zeroruntime.CompletionRequest{
+		Tools:           []zeroruntime.ToolDefinition{{Name: "read_file", Parameters: map[string]any{"type": "object"}}},
+		ReasoningEffort: "xhigh",
+	})
+	if base != droppedEffort {
+		t.Fatal("an effort the wire would omit must fingerprint like no effort")
+	}
+}
+
+func TestTurnSessionFingerprintDriftOnToolDescription(t *testing.T) {
+	provider := newTestProvider(t, func(http.ResponseWriter, *http.Request) {})
+	session := openOptimizedSession(t, provider)
+
+	base := session.computeFingerprint(zeroruntime.CompletionRequest{
+		Tools: []zeroruntime.ToolDefinition{{Name: "read_file", Description: "Reads a file.", Parameters: map[string]any{"type": "object"}}},
+	})
+	changedDescription := session.computeFingerprint(zeroruntime.CompletionRequest{
+		Tools: []zeroruntime.ToolDefinition{{Name: "read_file", Description: "Reads a file, now with ranges.", Parameters: map[string]any{"type": "object"}}},
+	})
+	if base == changedDescription {
+		t.Fatal("fingerprint did not drift on a description-only tool change — descriptions are model-visible request bytes")
+	}
+}
+
+func TestTurnSessionFingerprintDriftOnToolReorder(t *testing.T) {
+	provider := newTestProvider(t, func(http.ResponseWriter, *http.Request) {})
+	session := openOptimizedSession(t, provider)
+
+	forward := session.computeFingerprint(zeroruntime.CompletionRequest{Tools: []zeroruntime.ToolDefinition{
+		{Name: "alpha", Parameters: map[string]any{"type": "object"}},
+		{Name: "beta", Parameters: map[string]any{"type": "object"}},
+	}})
+	reversed := session.computeFingerprint(zeroruntime.CompletionRequest{Tools: []zeroruntime.ToolDefinition{
+		{Name: "beta", Parameters: map[string]any{"type": "object"}},
+		{Name: "alpha", Parameters: map[string]any{"type": "object"}},
+	}})
+	if forward == reversed {
+		t.Fatal("fingerprint did not drift on a tool reorder — request serialization preserves tool order, so a reorder changes the wire bytes")
+	}
+}
+
+// TestTurnSessionPrewarmDoesNotStampProviderConnect guards the A/B metric: the
+// prewarm probe must appear ONLY under provider_prewarm — a provider_connect
+// stamp from the probe would contaminate the exact span the benchmark compares.
+func TestTurnSessionPrewarmDoesNotStampProviderConnect(t *testing.T) {
+	provider := newPrewarmTestProvider(t, func(w http.ResponseWriter, _ *http.Request) {
+		w.WriteHeader(http.StatusMethodNotAllowed)
+	})
+	recorder := trace.NewRecorder("session-test", "run-1", "test")
+	recorder.Start()
+	ctx := trace.WithContext(context.Background(), recorder)
+
+	session := openOptimizedSession(t, provider)
+	if err := session.Prewarm(ctx); err != nil {
+		t.Fatalf("Prewarm: %v", err)
+	}
+	waitPrewarmDone(t, session)
+
+	tr := recorder.Finish()
+	for _, span := range tr.Spans {
+		if span.Name == trace.SpanProviderConnect {
+			t.Fatalf("prewarm stamped a %s span — it must stay solely under %s", trace.SpanProviderConnect, trace.SpanProviderPrewarm)
+		}
+	}
+}
+
+// TestTurnSessionPrewarmSkippedWhenKeepAlivesDisabled verifies the probe is not
+// spent when the transport cannot retain idle connections (the macOS shared
+// transport): no request, no trace stamps, and the done channel still closes.
+func TestTurnSessionPrewarmSkippedWhenKeepAlivesDisabled(t *testing.T) {
+	var requests atomic.Int64
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+		requests.Add(1)
+		w.WriteHeader(http.StatusMethodNotAllowed)
+	}))
+	defer server.Close()
+
+	transport := http.DefaultTransport.(*http.Transport).Clone()
+	transport.DisableKeepAlives = true
+	provider, err := New(Options{
+		APIKey:     "sk-test",
+		BaseURL:    server.URL,
+		Model:      "gpt-test",
+		HTTPClient: &http.Client{Transport: transport},
+	})
+	if err != nil {
+		t.Fatalf("New: %v", err)
+	}
+
+	recorder := trace.NewRecorder("session-test", "run-1", "test")
+	recorder.Start()
+	ctx := trace.WithContext(context.Background(), recorder)
+
+	session := openOptimizedSession(t, provider)
+	if err := session.Prewarm(ctx); err != nil {
+		t.Fatalf("Prewarm: %v", err)
+	}
+	waitPrewarmDone(t, session)
+
+	if got := requests.Load(); got != 0 {
+		t.Fatalf("keep-alive-disabled transport still received %d probe requests, want 0", got)
+	}
+	tr := recorder.Finish()
+	if got := tr.Counter(trace.CounterPrewarmAttempts); got != 0 {
+		t.Fatalf("prewarm_attempts = %d, want 0 for a skipped probe", got)
+	}
+}
+
+func TestTurnSessionPrefixCounters(t *testing.T) {
+	handler := func(w http.ResponseWriter, _ *http.Request) {
+		w.Header().Set("Content-Type", "text/event-stream")
+		_, _ = w.Write([]byte("data: [DONE]\n\n"))
+	}
+	provider := newTestProvider(t, handler)
+	session := openOptimizedSession(t, provider)
+
+	recorder := trace.NewRecorder("session-test", "run-1", "test")
+	recorder.Start()
+	ctx := trace.WithContext(context.Background(), recorder)
+
+	stable := zeroruntime.CompletionRequest{Tools: []zeroruntime.ToolDefinition{{Name: "read_file", Parameters: map[string]any{"type": "object"}}}}
+	drain := func(request zeroruntime.CompletionRequest) {
+		stream, err := session.Stream(ctx, request)
+		if err != nil {
+			t.Fatalf("Stream: %v", err)
+		}
+		for range stream {
+		}
+	}
+
+	drain(stable)                                                                                                                                 // seeds — neither stable nor drift
+	drain(stable)                                                                                                                                 // stable
+	drain(zeroruntime.CompletionRequest{Tools: []zeroruntime.ToolDefinition{{Name: "write_file", Parameters: map[string]any{"type": "object"}}}}) // drift
+
+	tr := recorder.Finish()
+	if got := tr.Counter(trace.CounterPrefixStable); got != 1 {
+		t.Fatalf("prefix_stable = %d, want 1", got)
+	}
+	if got := tr.Counter(trace.CounterPrefixDrift); got != 1 {
+		t.Fatalf("prefix_drift = %d, want 1", got)
+	}
+}
+
+func TestTurnSessionCloseIdempotentAndCompactUnsupported(t *testing.T) {
+	provider := newTestProvider(t, func(http.ResponseWriter, *http.Request) {})
+	session := openOptimizedSession(t, provider)
+	if err := session.Close(); err != nil {
+		t.Fatalf("Close: %v", err)
+	}
+	if err := session.Close(); err != nil {
+		t.Fatalf("second Close: %v", err)
+	}
+	if _, err := session.Compact(context.Background(), zeroruntime.CompletionRequest{}); !errors.Is(err, zeroruntime.ErrCompactionUnsupported) {
+		t.Fatalf("Compact error = %v, want ErrCompactionUnsupported", err)
+	}
+}

internal/providers/turn_session.go

@@ -0,0 +1,72 @@
+package providers
+
+import (
+	"os"
+	"strings"
+
+	"github.com/Gitlawb/zero/internal/config"
+	"github.com/Gitlawb/zero/internal/providers/openai"
+	"github.com/Gitlawb/zero/internal/zeroruntime"
+)
+
+// openaiTurnSessionEnv gates the optimized OpenAI turn session (prewarm +
+// prefix telemetry). Default OFF: unset, "0", or "false" (any case) leave the
+// agent loop's default adapter path untouched. Set to any other non-empty
+// value (e.g. "1") to enable — the same boolean idiom as ZERO_FORMAT_ON_WRITE.
+const openaiTurnSessionEnv = "ZERO_OPENAI_TURN_SESSION"
+
+func openaiTurnSessionEnabled() bool {
+	value := strings.TrimSpace(os.Getenv(openaiTurnSessionEnv))
+	return value != "" && value != "0" && !strings.EqualFold(value, "false")
+}
+
+// OptimizedTurnSessions returns the gated optimized TurnSessionProvider for an
+// already-built provider, or (nil, false) when the gate is off or the profile
+// is ineligible. Callers leave agent Options.TurnSessionProvider nil on false,
+// so the loop's default wrap keeps today's behavior byte-identical.
+//
+// Eligibility is deliberately narrow: the resolved provider kind must be
+// official OpenAI (NOT openai-compatible gateways — the constructor merges the
+// two, so this branches on the resolved kind explicitly), not the ChatGPT
+// Codex catalog, and the provider value must be the concrete *openai.Provider
+// (a fake, a Codex provider, or nil falls back to the default path safely).
+// No base-URL check on top of the kind: prewarm and fingerprint telemetry are
+// harmless against any host, and kind==openai mirrors the existing
+// official-OpenAI precedent used for prompt_cache_key.
+func OptimizedTurnSessions(profile config.ProviderProfile, provider zeroruntime.Provider, options Options) (zeroruntime.TurnSessionProvider, bool) {
+	if !openaiTurnSessionEnabled() {
+		return nil, false
+	}
+	resolved, err := resolveProfile(profile, options)
+	if err != nil || resolved.providerKind != config.ProviderKindOpenAI || isCodexCatalog(profile, resolved) {
+		return nil, false
+	}
+	concrete, ok := provider.(*openai.Provider)
+	if !ok {
+		return nil, false
+	}
+	caps, err := resolveCapabilities(profile, options)
+	if err != nil {
+		// resolveProfile above already succeeded, so this is effectively
+		// unreachable — but never ship an optimized session with unknown
+		// capabilities; the default path is the safe answer.
+		return nil, false
+	}
+	return openai.NewTurnSessionProvider(concrete, caps), true
+}
+
+// DefaultTurnSessions wraps an already-built provider in the DEFAULT (no-op)
+// turn-session adapter with the profile's resolved capability projection. It is
+// the non-optimized sibling of OptimizedTurnSessions for callers that still
+// want capabilities populated — e.g. the mid-run model-switch fallback when the
+// switched model is not eligible for the optimized session. Capability
+// resolution failure degrades to an unknown (zero) projection rather than
+// failing the wrap: the default session has no behavior that depends on
+// capabilities, so a swap must never be blocked by a projection error.
+func DefaultTurnSessions(profile config.ProviderProfile, provider zeroruntime.Provider, options Options) zeroruntime.TurnSessionProvider {
+	caps, err := resolveCapabilities(profile, options)
+	if err != nil {
+		caps = zeroruntime.ProviderCapabilities{}
+	}
+	return zeroruntime.NewProviderTurnSessionProvider(provider, caps)
+}

internal/providers/turn_session_gate_test.go

@@ -0,0 +1,128 @@
+package providers
+
+import (
+	"context"
+	"testing"
+
+	"github.com/Gitlawb/zero/internal/config"
+	"github.com/Gitlawb/zero/internal/zeroruntime"
+)
+
+func openaiEligibleProfile() config.ProviderProfile {
+	return config.ProviderProfile{
+		Name:         "official-openai",
+		ProviderKind: config.ProviderKindOpenAI,
+		BaseURL:      "https://provider.example/v1",
+		APIKey:       "sk-gate-test",
+		Model:        "pr8-unregistered-model",
+	}
+}
+
+func buildGateProvider(t *testing.T, profile config.ProviderProfile) zeroruntime.Provider {
+	t.Helper()
+	provider, err := New(profile, Options{UserAgent: "zero-gate-test"})
+	if err != nil {
+		t.Fatalf("New: %v", err)
+	}
+	return provider
+}
+
+type fakeGateProvider struct{}
+
+func (fakeGateProvider) StreamCompletion(context.Context, zeroruntime.CompletionRequest) (<-chan zeroruntime.StreamEvent, error) {
+	return nil, nil
+}
+
+func TestOptimizedTurnSessionsDefaultOff(t *testing.T) {
+	// No env set: the gate must be off even for a fully eligible profile.
+	profile := openaiEligibleProfile()
+	tsp, ok := OptimizedTurnSessions(profile, buildGateProvider(t, profile), Options{})
+	if ok || tsp != nil {
+		t.Fatal("optimized turn sessions must be disabled by default")
+	}
+}
+
+func TestOptimizedTurnSessionsFalseyValues(t *testing.T) {
+	profile := openaiEligibleProfile()
+	provider := buildGateProvider(t, profile)
+	for _, value := range []string{"0", "false", "FALSE", "False", " ", ""} {
+		t.Setenv(openaiTurnSessionEnv, value)
+		if _, ok := OptimizedTurnSessions(profile, provider, Options{}); ok {
+			t.Fatalf("gate enabled for falsey value %q", value)
+		}
+	}
+	for _, value := range []string{"1", "true", "on"} {
+		t.Setenv(openaiTurnSessionEnv, value)
+		if _, ok := OptimizedTurnSessions(profile, provider, Options{}); !ok {
+			t.Fatalf("gate disabled for truthy value %q", value)
+		}
+	}
+}
+
+func TestOptimizedTurnSessionsOpenAIEligible(t *testing.T) {
+	t.Setenv(openaiTurnSessionEnv, "1")
+	profile := openaiEligibleProfile()
+	tsp, ok := OptimizedTurnSessions(profile, buildGateProvider(t, profile), Options{})
+	if !ok || tsp == nil {
+		t.Fatal("expected the optimized session for an official-OpenAI profile with the gate on")
+	}
+	session, err := tsp.OpenTurnSession(context.Background())
+	if err != nil {
+		t.Fatalf("OpenTurnSession: %v", err)
+	}
+	if err := session.Close(); err != nil {
+		t.Fatalf("Close: %v", err)
+	}
+	if caps := tsp.Capabilities(); caps.Model != "pr8-unregistered-model" {
+		t.Fatalf("Capabilities().Model = %q, want the resolved api model", caps.Model)
+	}
+}
+
+func TestOptimizedTurnSessionsRejectsCompatible(t *testing.T) {
+	t.Setenv(openaiTurnSessionEnv, "1")
+	profile := openaiEligibleProfile()
+	profile.ProviderKind = config.ProviderKindOpenAICompatible
+	if _, ok := OptimizedTurnSessions(profile, buildGateProvider(t, profile), Options{}); ok {
+		t.Fatal("openai-compatible gateways must not get the optimized session")
+	}
+}
+
+func TestOptimizedTurnSessionsRejectsCodexCatalog(t *testing.T) {
+	t.Setenv(openaiTurnSessionEnv, "1")
+	profile := openaiEligibleProfile()
+	profile.CatalogID = "chatgpt"
+	// Build the provider from the base profile: the point under test is the
+	// catalog check, not Codex construction (which needs OAuth wiring).
+	if _, ok := OptimizedTurnSessions(profile, buildGateProvider(t, openaiEligibleProfile()), Options{}); ok {
+		t.Fatal("the ChatGPT Codex catalog must not get the optimized session")
+	}
+}
+
+func TestDefaultTurnSessionsPreservesResolvedCapabilities(t *testing.T) {
+	profile := openaiEligibleProfile()
+	tsp := DefaultTurnSessions(profile, buildGateProvider(t, profile), Options{})
+	if tsp == nil {
+		t.Fatal("DefaultTurnSessions returned nil")
+	}
+	caps := tsp.Capabilities()
+	if caps.Model != "pr8-unregistered-model" {
+		t.Fatalf("Capabilities().Model = %q, want the resolved api model", caps.Model)
+	}
+	session, err := tsp.OpenTurnSession(context.Background())
+	if err != nil {
+		t.Fatalf("OpenTurnSession: %v", err)
+	}
+	if err := session.Close(); err != nil {
+		t.Fatalf("Close: %v", err)
+	}
+}
+
+func TestOptimizedTurnSessionsRejectsForeignProviderValue(t *testing.T) {
+	t.Setenv(openaiTurnSessionEnv, "1")
+	if _, ok := OptimizedTurnSessions(openaiEligibleProfile(), fakeGateProvider{}, Options{}); ok {
+		t.Fatal("a non-*openai.Provider value must fall back to the default path")
+	}
+	if _, ok := OptimizedTurnSessions(openaiEligibleProfile(), nil, Options{}); ok {
+		t.Fatal("a nil provider must fall back to the default path")
+	}
+}

internal/trace/trace.go

@@ -32,6 +32,7 @@ const (
 	SpanToolPartition   = "tool_partition"   // partitioning the tool set for the prompt
 	SpanProviderConnect = "provider_connect" // client.Do in the provider seam
 	SpanProviderQueue   = "provider_queue"   // pre-send OAuth/token resolve
+	SpanProviderPrewarm = "provider_prewarm" // best-effort turn-session prewarm probe
 	SpanGeneration      = "generation"       // streaming a model completion
 	SpanToolExecution   = "tool_execution"   // executing tool calls
 	SpanPermissionWait  = "permission_wait"  // waiting on a permission prompt
@@ -53,6 +54,9 @@ const (
 	CounterInputTokens       = "input_tokens"
 	CounterCachedInputTokens = "cached_input_tokens"
 	CounterOutputTokens      = "output_tokens"
+	CounterPrewarmAttempts   = "prewarm_attempts"
+	CounterPrefixStable      = "prefix_stable"
+	CounterPrefixDrift       = "prefix_drift"
 )
 
 // Span is one named wall interval attributed to part of a run. Each stamp is
@@ -281,6 +285,10 @@ func OptionalEventKeys() []string {
 		"counter:" + CounterAcceptanceChecks,
 		"counter:" + CounterPollingTurn,
 		"counter:" + CounterModelSwitches,
+		"span:" + SpanProviderPrewarm,
+		"counter:" + CounterPrewarmAttempts,
+		"counter:" + CounterPrefixStable,
+		"counter:" + CounterPrefixDrift,
 		"event:prefix_hash",
 	}
 }

internal/zeroruntime/session.go

@@ -0,0 +1,125 @@
+package zeroruntime
+
+import (
+	"context"
+	"errors"
+)
+
+// ProviderCapabilities is a flat, provider-agnostic projection of a resolved
+// model's static capabilities. It is deliberately plain data — no modelregistry
+// or reasoning types — because modelregistry imports zeroruntime, so a typed
+// reference here would form an import cycle. The providers factory populates it
+// from the model-registry entry it already resolves; a zero value means the
+// capabilities are unknown, which every consumer must treat as "assume nothing".
+type ProviderCapabilities struct {
+	// Model is the resolved API model id (informational; may be empty).
+	Model string
+	// ContextWindow is the model's max input context in tokens; 0 = unknown.
+	ContextWindow int
+	// MaxOutputTokens is the model's max output tokens per turn; 0 = unknown.
+	MaxOutputTokens int
+	// SupportsVision mirrors the registry's vision capability.
+	SupportsVision bool
+	// SupportsReasoning mirrors the registry's reasoning capability.
+	SupportsReasoning bool
+	// SupportsPromptCache mirrors the registry's prompt-cache capability.
+	SupportsPromptCache bool
+	// ReasoningEfforts lists the accepted effort tiers, weakest to strongest.
+	// nil/empty means the model exposes no effort control.
+	ReasoningEfforts []string
+	// NativeCompaction reports server-side compaction support. Always false for
+	// the default adapter; a future native-compaction session sets it, and the
+	// loop branches on it to call TurnSession.Compact instead of the local
+	// summarizer.
+	NativeCompaction bool
+}
+
+// ErrCompactionUnsupported is returned by TurnSession.Compact when the provider
+// has no native server-side compaction (every default adapter). Callers treat
+// it as "use the local summarizer path", not as a run failure.
+var ErrCompactionUnsupported = errors.New("zeroruntime: native compaction unsupported")
+
+// TurnSession is one provider conversation for the lifetime of a single agent
+// run: many turns, compaction summaries, and any mid-run model swaps each open
+// a fresh session. It owns whatever per-run provider state an optimized
+// implementation keeps warm (a pooled connection, a cached-prefix handle). The
+// default adapter holds nothing and every method degrades to today's one-shot
+// request path.
+type TurnSession interface {
+	// Prewarm optionally primes provider-side state before the first Stream.
+	// It must be safe to skip entirely: the default adapter no-ops and returns
+	// nil. A non-nil error is advisory — callers proceed without prewarming
+	// (best-effort, never fatal).
+	Prewarm(ctx context.Context) error
+
+	// Stream issues one completion request. The signature is identical to
+	// Provider.StreamCompletion so the default adapter forwards verbatim and
+	// callers' stream handling stays byte-for-byte unchanged.
+	Stream(ctx context.Context, request CompletionRequest) (<-chan StreamEvent, error)
+
+	// Compact asks the provider to compact the conversation server-side and
+	// returns the replacement messages. The default adapter returns
+	// ErrCompactionUnsupported; callers then keep their local summarizer path.
+	Compact(ctx context.Context, request CompletionRequest) ([]Message, error)
+
+	// Close releases per-run provider state. It must be idempotent and safe to
+	// call after a mid-run swap as well as at run teardown. Default: no-op.
+	Close() error
+}
+
+// TurnSessionProvider opens a TurnSession for one run and exposes the resolved
+// model's static capabilities. The providers factory builds the default
+// implementation by wrapping an existing Provider.
+type TurnSessionProvider interface {
+	// OpenTurnSession starts a session for one run. Everything a session needs
+	// per request (messages, tools, reasoning effort, prompt-cache key) already
+	// arrives on Stream, so nothing beyond ctx is passed here — keeping the
+	// seam narrow. The default adapter never errors; a real implementation may
+	// fail a handshake, which callers surface as a clean run-start error.
+	OpenTurnSession(ctx context.Context) (TurnSession, error)
+
+	// Capabilities returns the resolved model's static capability projection.
+	Capabilities() ProviderCapabilities
+}
+
+// NewProviderTurnSessionProvider wraps an existing Provider as a default
+// TurnSessionProvider. caps may be the zero value (unknown) for callers that
+// only need streaming behavior; the providers factory supplies a populated
+// projection.
+func NewProviderTurnSessionProvider(provider Provider, caps ProviderCapabilities) TurnSessionProvider {
+	return providerTurnSessionProvider{provider: provider, caps: caps}
+}
+
+// providerTurnSessionProvider is the default TurnSessionProvider: it wraps a
+// Provider so every current provider plugs into the session seam unchanged.
+type providerTurnSessionProvider struct {
+	provider Provider
+	caps     ProviderCapabilities
+}
+
+func (d providerTurnSessionProvider) OpenTurnSession(context.Context) (TurnSession, error) {
+	return providerTurnSession{provider: d.provider}, nil
+}
+
+func (d providerTurnSessionProvider) Capabilities() ProviderCapabilities {
+	return d.caps
+}
+
+// providerTurnSession forwards Stream to the wrapped Provider; Prewarm and
+// Close are no-ops and Compact is unsupported. The value receiver keeps it
+// stateless, so double-Close and post-Close Stream are inherently safe.
+type providerTurnSession struct {
+	provider Provider
+}
+
+func (s providerTurnSession) Prewarm(context.Context) error { return nil }
+
+func (s providerTurnSession) Stream(ctx context.Context, request CompletionRequest) (<-chan StreamEvent, error) {
+	return s.provider.StreamCompletion(ctx, request)
+}
+
+func (s providerTurnSession) Compact(context.Context, CompletionRequest) ([]Message, error) {
+	return nil, ErrCompactionUnsupported
+}
+
+func (s providerTurnSession) Close() error { return nil }

internal/zeroruntime/session_test.go

@@ -0,0 +1,126 @@
+package zeroruntime
+
+import (
+	"context"
+	"errors"
+	"testing"
+)
+
+// recordingProvider captures every request and replays a scripted event stream,
+// so tests can assert the default session forwards verbatim.
+type recordingProvider struct {
+	requests []CompletionRequest
+	events   []StreamEvent
+	err      error
+}
+
+func (p *recordingProvider) StreamCompletion(_ context.Context, request CompletionRequest) (<-chan StreamEvent, error) {
+	p.requests = append(p.requests, request)
+	if p.err != nil {
+		return nil, p.err
+	}
+	ch := make(chan StreamEvent, len(p.events))
+	for _, event := range p.events {
+		ch <- event
+	}
+	close(ch)
+	return ch, nil
+}
+
+func TestDefaultTurnSessionStreamForwardsVerbatim(t *testing.T) {
+	provider := &recordingProvider{events: []StreamEvent{
+		{Type: StreamEventText, Content: "hello"},
+		{Type: StreamEventDone, FinishReason: "stop"},
+	}}
+	session, err := NewProviderTurnSessionProvider(provider, ProviderCapabilities{}).OpenTurnSession(context.Background())
+	if err != nil {
+		t.Fatalf("OpenTurnSession: %v", err)
+	}
+
+	request := CompletionRequest{
+		Messages:        []Message{{Role: MessageRoleUser, Content: "hi"}},
+		ReasoningEffort: "high",
+		PromptCacheKey:  "session-123",
+	}
+	stream, err := session.Stream(context.Background(), request)
+	if err != nil {
+		t.Fatalf("Stream: %v", err)
+	}
+
+	var got []StreamEvent
+	for event := range stream {
+		got = append(got, event)
+	}
+	if len(got) != 2 || got[0].Content != "hello" || got[1].FinishReason != "stop" {
+		t.Fatalf("stream events not forwarded verbatim: %#v", got)
+	}
+	if len(provider.requests) != 1 {
+		t.Fatalf("provider saw %d requests, want 1", len(provider.requests))
+	}
+	recorded := provider.requests[0]
+	if recorded.PromptCacheKey != request.PromptCacheKey ||
+		recorded.ReasoningEffort != request.ReasoningEffort ||
+		len(recorded.Messages) != 1 || recorded.Messages[0].Content != "hi" {
+		t.Fatalf("request not forwarded verbatim: %#v", recorded)
+	}
+}
+
+func TestDefaultTurnSessionStreamPropagatesError(t *testing.T) {
+	wantErr := errors.New("provider unavailable")
+	provider := &recordingProvider{err: wantErr}
+	session, err := NewProviderTurnSessionProvider(provider, ProviderCapabilities{}).OpenTurnSession(context.Background())
+	if err != nil {
+		t.Fatalf("OpenTurnSession: %v", err)
+	}
+	if _, streamErr := session.Stream(context.Background(), CompletionRequest{}); !errors.Is(streamErr, wantErr) {
+		t.Fatalf("Stream error = %v, want %v", streamErr, wantErr)
+	}
+}
+
+func TestDefaultTurnSessionPrewarmCloseNoop(t *testing.T) {
+	session, err := NewProviderTurnSessionProvider(&recordingProvider{}, ProviderCapabilities{}).OpenTurnSession(context.Background())
+	if err != nil {
+		t.Fatalf("OpenTurnSession: %v", err)
+	}
+	if err := session.Prewarm(context.Background()); err != nil {
+		t.Fatalf("Prewarm: %v", err)
+	}
+	if err := session.Close(); err != nil {
+		t.Fatalf("Close: %v", err)
+	}
+	// Idempotent: a second Close (e.g. after a mid-run swap already closed it)
+	// must also succeed.
+	if err := session.Close(); err != nil {
+		t.Fatalf("second Close: %v", err)
+	}
+}
+
+func TestDefaultTurnSessionCompactUnsupported(t *testing.T) {
+	session, err := NewProviderTurnSessionProvider(&recordingProvider{}, ProviderCapabilities{}).OpenTurnSession(context.Background())
+	if err != nil {
+		t.Fatalf("OpenTurnSession: %v", err)
+	}
+	if _, compactErr := session.Compact(context.Background(), CompletionRequest{}); !errors.Is(compactErr, ErrCompactionUnsupported) {
+		t.Fatalf("Compact error = %v, want ErrCompactionUnsupported", compactErr)
+	}
+}
+
+func TestDefaultTurnSessionProviderCapabilities(t *testing.T) {
+	caps := ProviderCapabilities{
+		Model:               "test-model",
+		ContextWindow:       200_000,
+		MaxOutputTokens:     64_000,
+		SupportsVision:      true,
+		SupportsReasoning:   true,
+		SupportsPromptCache: true,
+		ReasoningEfforts:    []string{"low", "medium", "high"},
+	}
+	tsp := NewProviderTurnSessionProvider(&recordingProvider{}, caps)
+	got := tsp.Capabilities()
+	if got.Model != caps.Model || got.ContextWindow != caps.ContextWindow ||
+		got.MaxOutputTokens != caps.MaxOutputTokens || !got.SupportsVision ||
+		!got.SupportsReasoning || !got.SupportsPromptCache ||
+		len(got.ReasoningEfforts) != 3 || got.NativeCompaction {
+		t.Fatalf("Capabilities() = %#v, want %#v", got, caps)
+	}
+}