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

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/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/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",
 	}
 }