package wire

import (
	"context"
	"crypto/rand"
	"encoding/base64"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"log/slog"
	mathrand "math/rand/v2"
	"net/http"
	"net/http/cookiejar"
	"net/url"
	"os"
	"runtime"
	"strings"
	"sync"
	"time"

	"github.com/gorilla/websocket"
	"go.yaml.in/yaml/v3"

	"github.com/tidbcloud/AgentTeam/desktop/internal/agent"
	"github.com/tidbcloud/AgentTeam/desktop/internal/agent/runtimeconfig"
	"github.com/tidbcloud/AgentTeam/desktop/internal/metrics"
	"github.com/tidbcloud/AgentTeam/desktop/internal/profile"
	"github.com/tidbcloud/AgentTeam/desktop/internal/types"
	"github.com/tidbcloud/AgentTeam/desktop/internal/upgrade"
	"github.com/tidbcloud/AgentTeam/desktop/internal/version"
	"github.com/tidbcloud/AgentTeam/desktop/internal/wire/apiclient"
	"github.com/tidbcloud/AgentTeam/internal/core/identifiers"
)

// ---------------------------------------------------------------------------
// Constants
// ---------------------------------------------------------------------------

const (
	textMaxBytes = 12 * 1024

	minBackoff = 1 * time.Second
	maxBackoff = 30 * time.Second

	webSocketDialTimeout = 10 * time.Second

	sendBufSize             = 256
	telemetrySendQueueSpare = 32
	writeWait               = 10 * time.Second
	pingPeriod              = 50 * time.Second
	pongWait                = 60 * time.Second

	criticalSendTimeout = 2 * time.Second

	dedupSetCap = 10000

	connectionStatusInvalidMachineAPIKey = "invalid_machine_api_key"

	transientLegacyWakeStatusServerCapability = "transient_legacy_wake_status"
	transientWakeStatusDesktopCapability      = "wake:status-transient"
)

var (
	errSendBufferFull        = errors.New("agentteam websocket send buffer full")
	errMachineAPIKeyRejected = errors.New("agentteam machine api key rejected")
)

// ---------------------------------------------------------------------------
// Config decoding
// ---------------------------------------------------------------------------

// decodeConfig unmarshals raw YAML bytes into a typed Config.
func decodeConfig(raw types.RawMessage) (Config, error) {
	if len(raw) == 0 {
		return Config{}, nil
	}
	var cfg Config
	if err := yaml.Unmarshal(raw, &cfg); err != nil {
		return Config{}, fmt.Errorf("unmarshal agentteam config: %w", err)
	}
	return cfg, nil
}

// ---------------------------------------------------------------------------
// Protocol
// ---------------------------------------------------------------------------

// Protocol implements the Gateway Channel interface for AgentTeam.
// It connects to a AgentTeam server via WebSocket for inbound messages
// and uses the HTTP API for outbound messages.
type Protocol struct {
	id          string
	enabled     bool
	serverURL   string
	apiKey      string
	workspaceID string
	machineID   string
	channelConnectionStatusState
	agentID    string
	authToken  string
	httpClient *http.Client
	// targetResolver owns the lifecycle-scoped /resolve-channel cache shared by
	// every temporary APIClient returned for this Protocol. Protected by mu.
	targetResolver *apiclient.TargetResolver
	// gatewayClient is the lifecycle-scoped HTTP client used to proxy web
	// traffic through the Desktop gateway. Built lazily by GatewayHTTPClient
	// so zero-value test Protocols never allocate it. gatewayClientMu
	// serializes lazy construction and idle-connection reclaim at Stop.
	gatewayClient   *http.Client
	gatewayClientMu sync.Mutex
	// streamingAPIClient is the lifecycle-scoped client for large request-body
	// uploads. It has no total timeout and owns a response-header-bounded
	// transport separate from the control-plane client.
	streamingAPIClient   *http.Client
	streamingAPIClientMu sync.Mutex
	// upgradeInfoPrefetchEnabled is set by New together with registering this
	// Protocol's refresher. Zero-value Protocols used by narrow helpers do not
	// own that global registration and therefore must not manufacture an
	// upgrade-info request at connect time.
	upgradeInfoPrefetchEnabled bool
	// agents tracks per-agent runtime state keyed by server agent ID.
	// Protected by mu.
	agents map[string]*agentState

	// serverCapabilities records the capabilities advertised by the Server for
	// the current WebSocket connection. Protected by mu and reset on reconnect.
	serverCapabilities map[string]struct{}

	// agentTokenRefresh tracks in-flight per-agent token refreshes keyed by
	// server agent ID, coalescing concurrent 401 recovery attempts. Protected
	// by mu.
	agentTokenRefresh map[string]*agentTokenRefreshState
	tokenRefreshProof string

	// dedupRing is a bounded ring buffer of hashes for successfully accepted
	// agent:deliver handoffs. When the buffer is full,
	// the oldest entry is evicted automatically. Protected by mu.
	dedupRing *dedupRing

	mu sync.Mutex
	channelLifecycleState

	// turnStatePublicationMu serializes fresh turn-state capture, revision
	// allocation, and enqueue for both ready/reconnect snapshots and incremental
	// publications, so every increment carries a strictly newer revision than the
	// ready snapshot watermark. State capture/read happens INSIDE this boundary —
	// never by stamping a precomputed stale TurnStateEvent after waiting for the
	// lock. turnStateRevision is protected by turnStatePublicationMu.
	turnStatePublicationMu sync.Mutex
	turnStateRevision      int64

	// WebSocket connection state (protected by wsMu).
	wsMu             sync.Mutex
	conn             *websocket.Conn
	connectionChange chan struct{}
	connectionErr    error
	connectionFailed bool
	sendCh           chan []byte
	sendMu           sync.Mutex

	channelAgentSyncState

	// workspaceListRequests tracks pending machine:workspaces:list requests
	// keyed by request id. Protected by mu.
	workspaceListRequests map[string]chan workspaceListResult

	channelDaemonTransportState

	// elasticRequestLinks tracks Desktop-initiated Elastic Request Links.
	// Protected by elasticLinksMu.
	elasticLinksMu         sync.Mutex
	elasticRequestLinks    map[string]*elasticRequestLink
	elasticDialContext     elasticRequestLinkDialContext
	remoteEndpointRegistry *desktopRemoteEndpointRegistry

	// debug enables verbose message logging. Protected by mu.
	debug bool

	channelTelemetryState

	disp dispatcherRuntime

	backoffFactory func() *reconnectBackoff
	channelUserTokenState
	channelAssetsEventState
	lastExtensionCatalogKey string
}

// assetsEventDedupCapacity bounds the per-server seen-event-id ring. It is
// intentionally larger than the server's typical replay window so genuine
// reconnect replays remain idempotent; FIFO eviction keeps memory bounded
// for long-lived daemons.
const assetsEventDedupCapacity = 512

// channelLifecycleState owns the Start/Stop run agentteam handles. It is embedded
// in Protocol so existing tests and internal package helpers can still inspect
// cancel/done directly while lifecycle ownership is no longer mixed into the
// protocol/config fields.
type channelLifecycleState struct {
	// lifecycleMu serializes public Start/Stop calls across the complete Stop
	// wait. A restart cannot publish until the preceding Stop has reached its
	// exact run's done boundary.
	lifecycleMu sync.Mutex
	cancel      context.CancelFunc
	done        chan struct{}
	generation  uint64
}

// channelConnectionStatusState mirrors the local-only reconnect marker
// persisted in the channel config. Empty means no marker is active.
type channelConnectionStatusState struct {
	connectionStatus string
}

// channelAgentSyncState owns machine:agent:list ticker and in-flight sync
// coalescing. Waiters are notified when the next list response is processed.
type channelAgentSyncState struct {
	// syncTickerLifecycleMu serializes replacement across the previous ticker's
	// exact done boundary. c.mu cannot be held while waiting because ticker
	// retirement clears its published pointer under c.mu.
	syncTickerLifecycleMu sync.Mutex
	agentListSyncInFlight bool
	agentListSyncWaiters  []chan bool
	syncTicker            *agentSyncTicker
}

// channelDaemonTransportState is scoped to the current Session Link. Elastic
// Request Links are bound to this generation and are closed when the Session
// Link disconnects or reconnects with a new generation.
type channelDaemonTransportState struct {
	daemonTransportSessionGeneration string
}

// channelTelemetryState tracks AgentTeam-specific operational metrics.
type channelTelemetryState struct {
	metrics metricsCollector
}

// channelUserTokenState caches the browser-facing AgentTeam API session token
// exchanged from the machine API key via the workspace-scoped daemon endpoint.
type channelUserTokenState struct {
	userTokenMu       sync.Mutex
	userToken         string
	userTokenIssuedAt time.Time
	userTokenFailures int
	userTokenRetryAt  time.Time
	// Test-only timing overrides. Zero values use the production clock and
	// retry bounds in session_token.go.
	userTokenNow      func() time.Time
	userTokenRetryMin time.Duration
	userTokenRetryMax time.Duration
}

// channelAssetsEventState guards the seen-event-id set so duplicate
// agent:assets:access-changed pushes (server retries, reconnect replays) do
// not trigger redundant refresher work. Dedup is keyed by event_id, not by a
// channel-level version, because a single server push can cover multiple
// agents/asset types and per-(agent, asset) versions are not monotonic across
// distinct emissions.
type channelAssetsEventState struct {
	assetsEventMu        sync.Mutex
	seenAssetsEventIDs   map[string]struct{}
	seenAssetsEventOrder []string
}

type agentSyncTicker struct {
	cancel context.CancelFunc
	done   chan struct{}
}

// getOrCreateAgentState returns the agentState for the given agent ID,
// creating it if it does not exist. Caller must hold c.mu.
func (c *Protocol) getOrCreateAgentState(agentID string) *agentState {
	st, ok := c.agents[agentID]
	if !ok {
		st = &agentState{}
		c.agents[agentID] = st
	}
	return st
}

// agentToken returns the per-agent auth token for the given agent ID.
// If the agent has no stored token, the channel-level authToken is returned
// as a fallback. This is safe for concurrent use.
func (c *Protocol) agentToken(agentID string) string {
	c.mu.Lock()
	defer c.mu.Unlock()
	if st, ok := c.agents[agentID]; ok && st.Token != "" {
		return st.Token
	}
	return c.authToken
}

// ---------------------------------------------------------------------------
// Constructor
// ---------------------------------------------------------------------------

// New constructs a wire Protocol from a saved AgentTeam server entry.
// runtime provides the narrow server connection runtime interface; the Protocol internally
// type-asserts to the local dispatcherRuntime interface for agentteam-specific methods.
func New(chCfg profile.AgentTeamServer, runtime types.ServerConnectionContext) (types.ServerConnection, error) {
	cfg, err := decodeConfig(chCfg.Config)
	if err != nil {
		return nil, fmt.Errorf("decode agentteam config: %w", err)
	}

	id := strings.TrimSpace(chCfg.ID)
	if id == "" {
		id = "agentteam"
	}

	serverURL := strings.TrimSpace(cfg.ServerURL)
	if serverURL == "" {
		serverURL = strings.TrimSpace(chCfg.ServerURL)
	}
	apiKey := strings.TrimSpace(cfg.APIKey)
	if apiKey == "" {
		apiKey = strings.TrimSpace(chCfg.APIKey)
	}
	workspaceID := strings.TrimSpace(cfg.WorkspaceID)
	if workspaceID == "" {
		workspaceID = strings.TrimSpace(chCfg.WorkspaceID)
	}
	// MachineID is resolved by the server during /connect and stored on the
	// canonical profile.AgentTeamServer field; tools that need a machine identity
	// read it back via Protocol.MachineID() / apiclient.Client.MachineID.
	machineID := strings.TrimSpace(chCfg.MachineID)
	connectionStatus := strings.TrimSpace(cfg.ConnectionStatus)
	if connectionStatus == "" {
		connectionStatus = strings.TrimSpace(chCfg.ConnectionStatus)
	}
	agentID := strings.TrimSpace(cfg.AgentID)
	if agentID == "" {
		agentID = strings.TrimSpace(chCfg.AgentID)
	}

	if serverURL == "" || apiKey == "" {
		return nil, fmt.Errorf("agentteam: server_url and api_key are required")
	}
	if workspaceID == "" {
		return nil, fmt.Errorf("agentteam: workspace_id is required; reconnect or log in again to refresh this Desktop profile")
	}
	if !identifiers.ValidWorkspaceID(workspaceID) {
		return nil, fmt.Errorf("agentteam: workspace_id must be a resolved workspace identifier")
	}

	authToken := strings.TrimSpace(cfg.AuthToken)
	if authToken == "" {
		authToken = strings.TrimSpace(chCfg.AuthToken)
	}
	if authToken == "" {
		authToken = apiKey
	}

	disp, ok := runtime.(dispatcherRuntime)
	if !ok {
		return nil, fmt.Errorf("agentteam: runtime does not implement dispatcherRuntime")
	}

	cookieJar, err := cookiejar.New(nil)
	if err != nil {
		return nil, fmt.Errorf("agentteam: create server cookie jar: %w", err)
	}

	ch := &Protocol{
		id:          id,
		enabled:     chCfg.Enabled,
		serverURL:   serverURL,
		apiKey:      apiKey,
		workspaceID: workspaceID,
		machineID:   machineID,
		channelConnectionStatusState: channelConnectionStatusState{
			connectionStatus: connectionStatus,
		},
		agentID:                    agentID,
		authToken:                  authToken,
		httpClient:                 &http.Client{Timeout: 30 * time.Second, Jar: cookieJar},
		targetResolver:             apiclient.NewTargetResolver(),
		upgradeInfoPrefetchEnabled: true,
		sendCh:                     make(chan []byte, sendBufSize),
		agents:                     make(map[string]*agentState),
		agentTokenRefresh:          make(map[string]*agentTokenRefreshState),
		dedupRing:                  newDedupRing(dedupSetCap),
		workspaceListRequests:      make(map[string]chan workspaceListResult),
		elasticRequestLinks:        make(map[string]*elasticRequestLink),
		remoteEndpointRegistry:     defaultDesktopRemoteEndpointRegistry(),
		backoffFactory:             defaultReconnectBackoff,
		disp:                       disp,
	}

	// Register the upgrade-info refresher so CheckVersion can fetch
	// server-driven guidance on demand when the cache is empty. The callback
	// captures this channel and only serves its own server key, preventing
	// guidance from leaking across servers.
	upgrade.RegisterRecommendationRefresher(ch.fetchUpgradeInfo)

	return ch, nil
}

// agentConfigFields holds the subset of Config fields that originate from
// the AgentTeam server and may be updated during sync. Local-only fields such
// as Policy and Enabled are never overwritten.
func agentConfigFields(ag types.MachineAgentInfo, runtimeName string, channelID string) agent.Config {
	displayName := ag.DisplayName
	if displayName == "" {
		displayName = ag.Name
	}
	llmMode, baseURL, apiKey, headers, wireAPI := agent.NormalizeAgentLLMFields(ag.LLMMode, ag.BaseURL, ag.APIKey, ag.Headers, ag.WireAPI)

	return agent.Config{
		ID:                  ag.ID,
		Name:                displayName,
		Runtime:             runtimeName,
		PermissionMode:      ag.PermissionMode,
		RuntimePolicy:       agentRuntimePolicyFromWire(ag.RuntimePolicy),
		RuntimeDebugEnabled: runtimeDebugEnabledFromWire(ag.RuntimeDebug),
		Description:         ag.Description,
		LLMMode:             llmMode,
		BaseURL:             baseURL,
		APIKey:              apiKey,
		Headers:             headers,
		WireAPI:             wireAPI,
		CredentialKind:      strings.TrimSpace(ag.CredentialKind),
		Model:               ag.Model,
		ReasoningEffort:     ag.ReasoningEffort,
		Source:              agent.NewAgentTeamSource(channelID),
	}
}

func agentRuntimePolicyFromWire(policy *types.RuntimePolicySnapshot) agent.RuntimePolicySnapshot {
	if policy == nil {
		return agent.RuntimePolicySnapshot{}
	}
	return agent.RuntimePolicySnapshot{
		SchemaVersion:  policy.SchemaVersion,
		Generation:     policy.Generation,
		Fingerprint:    strings.TrimSpace(policy.Fingerprint),
		Runtime:        strings.TrimSpace(policy.Runtime),
		RuntimeType:    strings.TrimSpace(policy.RuntimeType),
		PermissionMode: strings.TrimSpace(policy.PermissionMode),
	}
}

func runtimePolicySnapshotIsZero(policy agent.RuntimePolicySnapshot) bool {
	return policy.SchemaVersion == 0 &&
		policy.Generation == 0 &&
		strings.TrimSpace(policy.Fingerprint) == "" &&
		strings.TrimSpace(policy.Runtime) == "" &&
		strings.TrimSpace(policy.RuntimeType) == "" &&
		strings.TrimSpace(policy.PermissionMode) == ""
}

func runtimeDebugEnabledFromWire(config *types.RuntimeDebugConfig) bool {
	return config != nil && config.Enabled
}

// syncAgents ensures local agent configs stay in sync with the AgentTeam server.
// For agents that don't exist locally, a new config is created.
// For agents that already exist, only server-originated fields are updated;
// local-only fields (Policy, Enabled) are preserved.
// Local agents sourced from agentteam that no longer exist on the server are removed.
func (c *Protocol) syncAgents(agents []types.MachineAgentInfo, contexts ...context.Context) {
	mgr := profile.Default()
	ctx := firstConnectionTaskContext(contexts)

	// Build a set of remote agent IDs.
	remoteIDs := make(map[string]struct{}, len(agents))
	for _, ag := range agents {
		remoteIDs[ag.ID] = struct{}{}
	}

	// Remove local agentteam agents owned by this channel that no longer exist on the server.
	for _, local := range mgr.Agents().All() {
		if local.Source.Type != "agentteam" || local.Source.ID != c.id {
			continue
		}
		if _, exists := remoteIDs[local.ID]; exists {
			continue
		}
		// The server list is authoritative for target-cache identity even when
		// deleting the corresponding local profile later fails.
		c.removeResolvedTargetsForAgent(local.ID)
		if err := mgr.Agents().Delete(local.ID); err != nil {
			slog.Warn("failed to remove stale agentteam agent",
				"agent_id", local.ID, "agent_name", local.Name, "error", err)
		} else {
			slog.Info("removed stale agentteam agent from local config",
				"agent_id", local.ID, "agent_name", local.Name)
		}
	}

	// Build a map from runtime type → first matching runtime name.
	runtimes := mgr.Runtimes().All()
	typeToName := make(map[string]string, len(runtimes))
	for _, rt := range runtimes {
		if _, exists := typeToName[string(rt.Type)]; !exists {
			typeToName[string(rt.Type)] = rt.Name
		}
	}

	for _, ag := range agents {
		runtimeName, ok := typeToName[ag.Runtime]
		if !ok {
			slog.Warn("skipping agentteam agent sync: no local runtime with matching type",
				"agent_id", ag.ID,
				"agent_name", ag.DisplayName,
				"runtime_type", ag.Runtime,
			)
			continue
		}

		remote := agentConfigFields(ag, runtimeName, c.id)

		if existing, found := mgr.Agents().Get(ag.ID); found {
			// Merge: only update server-originated fields, preserve local extras.
			// For each human-readable field, only overwrite when the remote
			// side provides a non-empty value. Model/ReasoningEffort also
			// accept a present empty value as an explicit server clear while
			// preserving values when older/partial payloads omit the fields.
			merged := existing
			if remote.Name != "" {
				merged.Name = remote.Name
			}
			merged.Runtime = remote.Runtime
			if remote.Description != "" {
				merged.Description = remote.Description
			}
			if remote.Model != "" || ag.ModelSpecified {
				merged.Model = remote.Model
			}
			if remote.ReasoningEffort != "" || ag.ReasoningEffortSpecified {
				merged.ReasoningEffort = remote.ReasoningEffort
			}
			if remote.PermissionMode != "" {
				merged.PermissionMode = remote.PermissionMode
			}
			if !runtimePolicySnapshotIsZero(remote.RuntimePolicy) {
				merged.RuntimePolicy = remote.RuntimePolicy
			}
			merged.RuntimeDebugEnabled = remote.RuntimeDebugEnabled
			merged.LLMMode = remote.LLMMode
			merged.BaseURL = remote.BaseURL
			merged.APIKey = remote.APIKey
			merged.Headers = remote.Headers
			merged.WireAPI = remote.WireAPI
			merged.CredentialKind = remote.CredentialKind
			merged.Source = remote.Source

			if err := mgr.Agents().Update(merged); err != nil {
				slog.Warn("failed to merge agentteam agent update",
					"agent_id", ag.ID, "agent_name", remote.Name, "error", err)
			} else {
				slog.Info("merged agentteam agent update",
					"agent_id", ag.ID, "agent_name", remote.Name, "runtime", runtimeName)
				// agent:update_identity is best-effort: it can be missed while the
				// daemon is offline or while an older server is being upgraded. The
				// authoritative catalog must therefore repair daemon-local resume
				// identity as well as agents.yaml when it observes a rename.
				if c.disp != nil {
					c.applyCatalogAgentIdentity(ctx, ag.ID, ag.Name, ag.DisplayName)
				}
				c.refreshRunningAgentConfig(merged)
			}
			continue
		}

		if err := mgr.Agents().Add(remote); err != nil {
			slog.Warn("failed to sync agentteam agent",
				"agent_id", ag.ID, "agent_name", remote.Name, "error", err)
			continue
		}

		slog.Info("synced agentteam agent to local config",
			"agent_id", ag.ID, "agent_name", remote.Name, "runtime", runtimeName)
		if c.disp != nil {
			c.applyCatalogAgentIdentity(ctx, ag.ID, ag.Name, ag.DisplayName)
		}
		c.refreshRunningAgentConfig(remote)
	}
}

func (c *Protocol) applyCatalogAgentIdentity(ctx context.Context, agentID, name, displayName string) {
	c.rememberAgentIdentity(agentID, name, displayName)
	msg := agentUpdateIdentityMsg{AgentID: agentID, Name: name, DisplayName: displayName}
	if connectionRunFromContext(ctx) == nil {
		// Narrow direct-handler tests and process-scoped callers historically
		// observe catalog repair synchronously. Production connection frames use
		// the receipt-ordered lifecycle chain below.
		c.applyPreparedAgentIdentityUpdate(msg)
		return
	}
	c.startAgentLifecycleTask(ctx, agentID, func(context.Context) {
		c.applyPreparedAgentIdentityUpdate(msg)
	})
}

func (c *Protocol) refreshRunningAgentConfig(cfg agent.Config) {
	if c == nil || c.disp == nil {
		return
	}
	refresher, ok := c.disp.(runningAgentConfigRefresher)
	if !ok || refresher == nil {
		return
	}
	refresher.RefreshRunningAgentConfig(cfg)
}

// ---------------------------------------------------------------------------
// Channel interface
// ---------------------------------------------------------------------------

// ID returns the unique channel identifier.
func (c *Protocol) ID() string {
	return c.id
}

// Enabled reports whether the channel should be started.
func (c *Protocol) Enabled() bool {
	return c.enabled
}

// ServerURL returns the AgentTeam gateway base URL this server connection connects to.
func (c *Protocol) ServerURL() string {
	serverURL, _, _, _, _, _ := c.identitySnapshot()
	return serverURL
}

// AuthToken returns the Bearer token used for outbound API calls.
func (c *Protocol) AuthToken() string {
	_, _, _, _, _, authToken := c.identitySnapshot()
	return authToken
}

// WorkspaceID returns the workspace this channel is scoped to.
func (c *Protocol) WorkspaceID() string {
	_, _, workspaceID, _, _, _ := c.identitySnapshot()
	return workspaceID
}

// MachineID returns the machine identity this channel runs on, sourced from
// the channel config at New() time. Empty when the channel hasn't been
// bound to a machine yet (e.g., fresh login before server resolution).
// Tools that require a machine identity MUST guard on this being non-empty
// rather than asking the model to supply one.
func (c *Protocol) MachineID() string {
	_, _, _, machineID, _, _ := c.identitySnapshot()
	return machineID
}

// GatewayResponseHeaderTimeout bounds how long the gateway client waits
// for response headers from the upstream before aborting the request.
// The control-plane *http.Client.Timeout (30s) cannot be reused because
// long-lived SSE / streaming gateway responses are not capped), so this is
// the only header-side guard. Exported so tests sharing
// BuildGatewayHTTPClient can reference the same production constant.
const GatewayResponseHeaderTimeout = 30 * time.Second

// ErrUnsupportedGatewayTransport is returned by the gateway client when
// BuildGatewayHTTPClient cannot install a header-side watchdog over the
// control-plane transport. Production wire.New always feeds Transport=nil
// (cloned from http.DefaultTransport) or an explicit *http.Transport,
// both of which honor ResponseHeaderTimeout. A custom RoundTripper cannot
// be supported because:
//
//   - The interface has no Clone, so the gateway would have to share the
//     control-plane pool (defeating pool isolation).
//   - A custom RoundTripper that does not respect request context
//     cancellation cannot be interrupted by any external header timer;
//     the wrapper would block forever and the watchdog has no escape hatch.
//
// Rather than silently degrade to a header-timeout-less client, the builder
// installs a rejecting transport that fails every request with this
// sentinel. The Desktop gateway surfaces the failure as REMOTE_UNREACHABLE;
// the underlying custom RoundTripper is never invoked.
var ErrUnsupportedGatewayTransport = errors.New(
	"gateway client requires an *http.Transport (or nil) control-plane transport; " +
		"custom RoundTripper is not supported because ResponseHeaderTimeout cannot be enforced",
)

// rejectingGatewayRoundTripper implements http.RoundTripper by always
// returning ErrUnsupportedGatewayTransport. BuildGatewayHTTPClient
// installs it when the control-plane client carries a non-*http.Transport
// RoundTripper, so requests fail loudly at Do time instead of silently
// degrading to a header-timeout-less client.
type rejectingGatewayRoundTripper struct{}

func (rejectingGatewayRoundTripper) RoundTrip(*http.Request) (*http.Response, error) {
	return nil, ErrUnsupportedGatewayTransport
}

// BuildGatewayHTTPClient is the production-owned policy that turns a
// control-plane *http.Client into a gateway proxy client. The result has:
//
//   - Timeout=0 so long-lived SSE / streaming responses are not capped by
//     the control-plane total deadline.
//   - Jar=nil so upstream cookies do not leak into (and are not polluted
//     by) daemon control-plane calls.
//   - CheckRedirect=ErrUseLastResponse so the local Desktop gateway owns
//     all redirect rewriting.
//   - Transport cloned (not shared) from the control-plane transport so
//     in-flight streams do not consume slots from the control-plane pool.
//     Proxy / dial / TLS policies are inherited via Transport.Clone().
//     When the control-plane client has Transport==nil (the production
//     wire.New shape) or a typed-nil *http.Transport, the gateway client
//     clones http.DefaultTransport so it still gets a private pool — but
//     ONLY when the global default is itself a non-nil *http.Transport.
//     If the global has been replaced with a custom RoundTripper, the
//     builder installs the rejecting transport instead of silently
//     inheriting that custom RT via http.Client.Do's implicit fallback.
//   - ResponseHeaderTimeout=headerTimeout on the cloned transport to bound
//     the "accepted but no response" failure mode. Pass
//     GatewayResponseHeaderTimeout for production; tests may pass a
//     shorter value to keep the suite fast.
//
// If the control-plane client uses a non-*http.Transport RoundTripper,
// the gateway client cannot enforce ResponseHeaderTimeout (the
// *http.Transport field is the only header-side watchdog the standard
// library offers). Rather than silently degrade, the builder installs a
// rejecting transport that fails every Do with
// ErrUnsupportedGatewayTransport. The original custom RoundTripper is
// never invoked. The same rejection applies when source.Transport is
// nil/typed-nil AND the process has replaced http.DefaultTransport with
// a non-*http.Transport value: without the rejection, http.Client.Do
// would fall back to that same global default and bypass the watchdog.
//
// Production callers go through Protocol.GatewayHTTPClient, which caches
// one client per Protocol; tests that exercise the gateway handler use
// BuildGatewayHTTPClient directly so the policy shape cannot drift
// between production and fake connections.
func BuildGatewayHTTPClient(control *http.Client, headerTimeout time.Duration) *http.Client {
	source := control
	if source == nil {
		source = http.DefaultClient
	}
	gateway := &http.Client{
		Timeout: 0,
		Jar:     nil,
		CheckRedirect: func(*http.Request, []*http.Request) error {
			return http.ErrUseLastResponse
		},
	}
	switch st := source.Transport.(type) {
	case *http.Transport:
		if st != nil {
			cloned := st.Clone()
			cloned.ResponseHeaderTimeout = headerTimeout
			gateway.Transport = cloned
		} else {
			gateway.Transport = cloneDefaultGatewayTransport(headerTimeout)
		}
	default:
		if source.Transport == nil {
			gateway.Transport = cloneDefaultGatewayTransport(headerTimeout)
		} else {
			gateway.Transport = rejectingGatewayRoundTripper{}
		}
	}
	return gateway
}

// cloneDefaultGatewayTransport returns a gateway-suitable RoundTripper
// derived from http.DefaultTransport. When the process global is a
// non-nil *http.Transport, the result is a private clone with the
// header-side watchdog applied. When the global has been replaced with a
// custom RoundTripper (or is typed-nil), the result is the rejecting
// transport: http.Client.Do would otherwise fall back to the same global
// default and silently bypass the fail-fast contract.
func cloneDefaultGatewayTransport(headerTimeout time.Duration) http.RoundTripper {
	if dt, ok := http.DefaultTransport.(*http.Transport); ok && dt != nil {
		cloned := dt.Clone()
		cloned.ResponseHeaderTimeout = headerTimeout
		return cloned
	}
	return rejectingGatewayRoundTripper{}
}

// GatewayHTTPClient returns the Protocol's lifecycle-scoped HTTP client
// for proxying web traffic through the Desktop gateway. The first call
// constructs a single client via BuildGatewayHTTPClient; all subsequent
// calls return the same instance so keep-alive connections are reused
// across gateway requests. Stop reclaims the pool's idle connections.
func (c *Protocol) GatewayHTTPClient() *http.Client {
	c.gatewayClientMu.Lock()
	defer c.gatewayClientMu.Unlock()
	if c.gatewayClient != nil {
		return c.gatewayClient
	}
	c.gatewayClient = BuildGatewayHTTPClient(c.httpClient, GatewayResponseHeaderTimeout)
	return c.gatewayClient
}

// closeGatewayClientIdleConnections releases any pooled gateway transport
// idle connections. Called from Stop so shutdown does not strand
// keep-alive connections to the upstream regardless of whether the
// Protocol was ever Start-ed.
func (c *Protocol) closeGatewayClientIdleConnections() {
	c.gatewayClientMu.Lock()
	client := c.gatewayClient
	c.gatewayClientMu.Unlock()
	if client == nil {
		return
	}
	if t, ok := client.Transport.(*http.Transport); ok {
		t.CloseIdleConnections()
	}
}

// ApplyServerBinding atomically changes the live workspace binding without
// waiting for the Protocol run agentteam. The profile mutation owner holds its
// per-server transaction until this method acknowledges the CAS-style apply.
func (c *Protocol) ApplyServerBinding(previous, current profile.AgentTeamServer) error {
	if c == nil || strings.TrimSpace(previous.ID) != c.id || strings.TrimSpace(current.ID) != c.id {
		return fmt.Errorf("server binding does not belong to protocol %q", c.id)
	}
	previousWorkspaceID := strings.TrimSpace(previous.WorkspaceID)
	currentWorkspaceID := strings.TrimSpace(current.WorkspaceID)
	if !identifiers.ValidWorkspaceID(previousWorkspaceID) {
		return fmt.Errorf("previous server workspace binding must be canonical")
	}
	if !identifiers.ValidWorkspaceID(currentWorkspaceID) {
		return fmt.Errorf("current server workspace binding must be canonical")
	}
	c.mu.Lock()
	defer c.mu.Unlock()
	liveWorkspaceID := strings.TrimSpace(c.workspaceID)
	if liveWorkspaceID != previousWorkspaceID && liveWorkspaceID != currentWorkspaceID {
		return fmt.Errorf(
			"server workspace binding changed concurrently: have %q, expected %q or %q",
			c.workspaceID,
			previous.WorkspaceID,
			current.WorkspaceID,
		)
	}
	c.workspaceID = currentWorkspaceID
	c.machineID = strings.TrimSpace(current.MachineID)
	return nil
}

func (c *Protocol) identitySnapshot() (serverURL, apiKey, workspaceID, machineID, agentID, authToken string) {
	c.mu.Lock()
	defer c.mu.Unlock()
	return c.serverURL, c.apiKey, c.workspaceID, c.machineID, c.agentID, c.authToken
}

// ServerIdentitySnapshot returns the live identity used to verify an
// explicit repair before the profile owner clears quarantine.
func (c *Protocol) ServerIdentitySnapshot() profile.AgentTeamServer {
	serverURL, apiKey, workspaceID, machineID, agentID, authToken := c.identitySnapshot()
	return profile.AgentTeamServer{
		ID:          c.id,
		Enabled:     c.enabled,
		ServerURL:   serverURL,
		APIKey:      apiKey,
		WorkspaceID: workspaceID,
		MachineID:   machineID,
		AgentID:     agentID,
		AuthToken:   authToken,
	}
}

// recordSentStats records a sent message event in the server message stats singleton.
func (c *Protocol) recordSentStats() {
	if cs := metrics.DefaultServerMessageCollector(); cs != nil {
		cs.RecordSent(c.id)
	}
}

func (c *Protocol) recordAgentSentStats(agentID string) {
	if cs := metrics.DefaultServerMessageCollector(); cs != nil {
		cs.RecordAgentSent(agentID)
	}
}

// Start launches the WebSocket connection agentteam in the background.
func (c *Protocol) Start(ctx context.Context) error {
	if !c.enabled {
		return nil
	}
	if ctx == nil {
		ctx = context.Background()
	}

	c.lifecycleMu.Lock()
	defer c.lifecycleMu.Unlock()
	for {
		c.mu.Lock()
		if c.cancel == nil {
			agentteamCtx, cancel := context.WithCancel(ctx)
			done := make(chan struct{})
			c.generation++
			generation := c.generation
			c.cancel = cancel
			c.done = done
			c.mu.Unlock()

			go func() {
				c.serve(agentteamCtx)
				// Publish completion before clearing the exact run identity. A
				// concurrent Stop can never mistake a newer run for this one, and
				// a concurrent Start is serialized until Stop returns.
				close(done)
				c.clearRunState(generation, done)
			}()
			break
		}

		// A naturally exited run may have closed done but not yet cleared its
		// fields. Help finish that exact generation, then retry publication.
		done := c.done
		generation := c.generation
		c.mu.Unlock()
		select {
		case <-done:
			c.clearRunState(generation, done)
			continue
		default:
			return nil
		}
	}

	slog.Info("started AgentTeam WebSocket connection", "channel", c.id)
	return nil
}

// Stop requests shutdown and waits for the connection to exit.
func (c *Protocol) Stop(ctx context.Context) error {
	if ctx == nil {
		ctx = context.Background()
	}
	defer c.closeGatewayClientIdleConnections()
	defer c.closeStreamingAPIClientIdleConnections()
	c.lifecycleMu.Lock()
	defer c.lifecycleMu.Unlock()

	c.mu.Lock()
	cancel := c.cancel
	done := c.done
	generation := c.generation
	syncTicker := c.syncTicker
	c.mu.Unlock()

	if cancel == nil || done == nil {
		return nil
	}

	if syncTicker != nil {
		syncTicker.cancel()
	}
	cancel()

	select {
	case <-done:
		c.clearRunState(generation, done)
		return nil
	case <-ctx.Done():
		return ctx.Err()
	}
}

// ToggleDebug toggles the debug mode for the AgentTeam channel, returning
// the new debug state. This implements the types.DebugToggler interface.
func (c *Protocol) ToggleDebug() bool {
	c.mu.Lock()
	defer c.mu.Unlock()
	c.debug = !c.debug
	slog.Info("AgentTeam: debug mode toggled", "channel", c.id, "debug", c.debug)
	return c.debug
}

func (c *Protocol) clearRunState(generation uint64, done chan struct{}) {
	c.mu.Lock()
	defer c.mu.Unlock()
	if c.generation != generation || c.done != done {
		return
	}
	c.cancel = nil
	c.done = nil
}

func (c *Protocol) currentConnectionStatus() string {
	c.mu.Lock()
	defer c.mu.Unlock()
	return c.connectionStatus
}

func (c *Protocol) setConnectionStatus(status string) {
	c.mu.Lock()
	defer c.mu.Unlock()
	c.connectionStatus = strings.TrimSpace(status)
}

// ---------------------------------------------------------------------------
// WebSocket connection agentteam (with exponential backoff reconnection)
// ---------------------------------------------------------------------------

// reconnectLogBatchSize is the number of consecutive failures that each get an
// individual log line before the agentteam switches to one summary line per batch.
// This keeps a dead server entry from flooding the log every backoff tick.
const reconnectLogBatchSize = 20

// shouldLogReconnectAttempt reports whether an attempt in the current failure
// streak deserves a log line: the first 10 attempts log individually, then one
// line per batch of 20 keeps the streak observable without the noise.
func shouldLogReconnectAttempt(attempt int) bool {
	return attempt <= 10 || attempt%reconnectLogBatchSize == 0
}

func (c *Protocol) serve(ctx context.Context) {
	backoff := c.newReconnectBackoff()
	for {
		attempt := backoff.Attempt() + 1
		serverURL := c.ServerURL()
		if shouldLogReconnectAttempt(attempt) {
			slog.Info("AgentTeam: dialing WebSocket", "channel", c.id, "server_url", serverURL, "attempt", attempt, "timeout", webSocketDialTimeout)
		}

		err := c.connect(ctx)
		if err != nil {
			terminal := errors.Is(err, errMachineAPIKeyRejected)
			c.recordConnectionFailure(err, terminal)
			if terminal {
				if persistErr := c.persistConnectionStatus(connectionStatusInvalidMachineAPIKey); persistErr != nil {
					slog.Warn("AgentTeam: failed to persist machine reconnect state", "channel", c.id, "error", persistErr)
				}
				slog.Error("AgentTeam: machine API key rejected; reconnect this daemon",
					"channel", c.id,
					"server_url", serverURL,
					"action", "create a new machine connection in AgentTeam Web or run agentteam daemon with a fresh --api-key",
					"error", err)
				return
			}
			delay := backoff.Next()
			if shouldLogReconnectAttempt(attempt) {
				slog.Warn("AgentTeam: WebSocket connection failed; retrying", "channel", c.id, "server_url", serverURL, "attempt", attempt, "backoff", delay, "error", err)
			}
			if !sleepWithContext(ctx, delay) {
				return
			}
			continue
		}

		slog.Info("connected to AgentTeam server", "channel", c.id, "server_url", c.ServerURL())
		if c.currentConnectionStatus() != "" {
			if persistErr := c.persistConnectionStatus(""); persistErr != nil {
				slog.Warn("AgentTeam: failed to clear machine reconnect state", "channel", c.id, "error", persistErr)
			}
		}
		if backoff.Attempt() > 0 {
			c.metrics.reconnects.Add(1)
		}
		backoff.Reset()
		c.metrics.connectedAt.Store(time.Now().Unix())

		if dropped := c.drainSendQueue(); dropped > 0 {
			slog.Warn("AgentTeam: dropped stale outbound messages before ready handshake", "channel", c.id, "count", dropped)
		}

		connectionRun := newConnectionRun(ctx)
		connCtx := connectionRun.context()
		pumpsStarted := make(chan struct{})
		pumpsDone := make(chan struct{})
		if !connectionRun.start(connCtx, func(context.Context) {
			defer close(pumpsDone)
			c.runPumps(connCtx, pumpsStarted)
		}) {
			// A fresh run cannot be closed before its first admission, but keep the
			// failure path exact if construction changes in the future.
			close(pumpsDone)
			return
		}

		select {
		case <-pumpsStarted:
		case <-ctx.Done():
			c.finishConnectionRun(connectionRun, nil, "session_stopped_before_connect")
			return
		}

		c.onConnect(connCtx)
		c.onServerConnect(connCtx)
		stopSessionTelemetry := c.startSessionStateTelemetry(connCtx)
		<-pumpsDone
		// Stop publication first, then cancel and exact-join every subordinate
		// owner before clearing connection state or allowing the reconnect agentteam to
		// publish a new generation. The final drain is deliberately after the join:
		// an admitted handler may observe cancellation and enqueue one last response.
		c.finishConnectionRun(connectionRun, stopSessionTelemetry, "session_disconnected")
		c.clearDaemonTransportSessionGeneration()
		c.onServerDisconnect()
		slog.Info("disconnected from AgentTeam server", "channel", c.id, "server_url", c.ServerURL())

		if ctx.Err() != nil {
			return
		}

		delay := backoff.Next()
		slog.Info("AgentTeam: scheduling WebSocket reconnect", "channel", c.id, "server_url", c.ServerURL(), "backoff", delay)
		if !sleepWithContext(ctx, delay) {
			return
		}
	}
}

func (c *Protocol) newReconnectBackoff() *reconnectBackoff {
	if c.backoffFactory != nil {
		return c.backoffFactory()
	}
	return defaultReconnectBackoff()
}

type reconnectBackoff struct {
	attempt int
	min     time.Duration
	max     time.Duration
}

func defaultReconnectBackoff() *reconnectBackoff {
	return &reconnectBackoff{min: minBackoff, max: maxBackoff}
}

func (b *reconnectBackoff) Attempt() int {
	return b.attempt
}

func (b *reconnectBackoff) Reset() {
	b.attempt = 0
}

// Next returns the next reconnect delay. The exponential ramp keeps its exact
// deterministic bounds for tests (min == max means a fixed delay), while every
// production ramp carries a ±20% jitter so independently restarted daemons do
// not resynchronize their retries into a thundering herd against one server.
// The jittered delay stays clamped to the configured [min, max] envelope.
func (b *reconnectBackoff) Next() time.Duration {
	if b.min <= 0 {
		b.min = minBackoff
	}
	if b.max <= 0 {
		b.max = maxBackoff
	}
	delay := boundedExponentialDelay(b.min, b.max, b.attempt)
	b.attempt++
	if b.max > b.min {
		delay = clampDelay(jitterDelay(delay), b.min, b.max)
	}
	return delay
}

// clampDelay keeps a jittered delay inside the configured [min, max]
// envelope. Pure so the bounds invariant can be tested deterministically.
func clampDelay(delay, minDelay, maxDelay time.Duration) time.Duration {
	if delay < minDelay {
		return minDelay
	}
	if delay > maxDelay {
		return maxDelay
	}
	return delay
}

func boundedExponentialDelay(minDelay, maxDelay time.Duration, attempt int) time.Duration {
	if maxDelay < minDelay {
		maxDelay = minDelay
	}
	delay := minDelay
	for i := 0; i < attempt && delay < maxDelay; i++ {
		if delay > maxDelay/2 {
			return maxDelay
		}
		delay *= 2
	}
	if delay > maxDelay {
		return maxDelay
	}
	return delay
}

// jitterDelay randomizes delay within ±20%. The caller clamps the result back
// into the configured [min, max] envelope.
func jitterDelay(delay time.Duration) time.Duration {
	if delay <= 0 {
		return delay
	}
	spread := delay / 5
	if spread <= 0 {
		return delay
	}
	return delay - spread + time.Duration(mathrand.Int64N(int64(2*spread)+1))
}

func sleepWithContext(ctx context.Context, delay time.Duration) bool {
	timer := time.NewTimer(delay)
	defer timer.Stop()
	select {
	case <-ctx.Done():
		return false
	case <-timer.C:
		return true
	}
}

func (c *Protocol) connect(ctx context.Context) error {
	dialCtx, cancel := context.WithTimeout(ctx, webSocketDialTimeout)
	defer cancel()

	dialer := websocket.Dialer{HandshakeTimeout: webSocketDialTimeout, EnableCompression: true}
	serverURL, apiKey, workspaceID, _, _, _ := c.identitySnapshot()
	wsURL := buildWSURL(serverURL, apiKey, workspaceID)

	conn, resp, err := dialer.DialContext(dialCtx, wsURL, routeWorkspaceHeaders(workspaceID))
	if err != nil {
		if isInvalidMachineAPIKeyResponse(resp) {
			return fmt.Errorf("%w: %w", errMachineAPIKeyRejected, formatHandshakeError(err, resp, wsURL))
		}
		return formatHandshakeError(err, resp, wsURL)
	}

	// The capability set belongs to exactly one authenticated WebSocket. Reset
	// it after a successful handshake and before publishing the connection or
	// starting read/write pumps, so an immediate Server capability frame cannot
	// be applied and then accidentally cleared by onConnect.
	c.replaceServerCapabilities(nil)

	c.wsMu.Lock()
	c.conn = conn
	c.connectionErr = nil
	c.connectionFailed = false
	c.notifyConnectionChangeLocked()
	c.wsMu.Unlock()

	return nil
}

// WaitConnected blocks until this Protocol has completed a WebSocket
// handshake. Desktop onboarding uses this narrow readiness boundary so a
// newly added local Machine is not first rendered as offline.
func (c *Protocol) WaitConnected(ctx context.Context) error {
	if ctx == nil {
		ctx = context.Background()
	}
	for {
		c.wsMu.Lock()
		if c.conn != nil {
			c.wsMu.Unlock()
			return nil
		}
		if c.connectionFailed {
			err := c.connectionErr
			c.wsMu.Unlock()
			return err
		}
		changed := c.connectionChangeLocked()
		c.wsMu.Unlock()

		select {
		case <-ctx.Done():
			c.wsMu.Lock()
			lastErr := c.connectionErr
			c.wsMu.Unlock()
			if lastErr != nil {
				return fmt.Errorf("%w; last WebSocket connection error: %v", ctx.Err(), lastErr)
			}
			return ctx.Err()
		case <-changed:
		}
	}
}

func (c *Protocol) recordConnectionFailure(err error, terminal bool) {
	c.wsMu.Lock()
	c.connectionErr = err
	c.connectionFailed = terminal
	c.notifyConnectionChangeLocked()
	c.wsMu.Unlock()
}

func (c *Protocol) connectionChangeLocked() chan struct{} {
	if c.connectionChange == nil {
		c.connectionChange = make(chan struct{})
	}
	return c.connectionChange
}

func (c *Protocol) notifyConnectionChangeLocked() {
	changed := c.connectionChangeLocked()
	close(changed)
	c.connectionChange = make(chan struct{})
}

// persistConnectionStatus writes the local reconnect state back to servers.yaml.
// An empty status clears the marker after a successful connection.
func (c *Protocol) persistConnectionStatus(status string) error {
	mgr := profile.Default()
	if mgr == nil {
		return fmt.Errorf("profile manager unavailable")
	}
	status = strings.TrimSpace(status)
	if c.currentConnectionStatus() == status {
		return nil
	}
	if _, err := mgr.Servers().PatchConnectionStatus(c.id, status); err != nil {
		return fmt.Errorf("update server config: %w", err)
	}
	c.setConnectionStatus(status)
	return nil
}

func formatHandshakeError(err error, resp *http.Response, wsURL string) error {
	if err == nil {
		return nil
	}
	if resp == nil {
		return fmt.Errorf("websocket dial failed for %s: %w", redactWSURLKey(wsURL), err)
	}

	details := resp.Status
	if resp.Body != nil {
		body, readErr := io.ReadAll(io.LimitReader(resp.Body, 1024))
		if closeErr := resp.Body.Close(); closeErr != nil && readErr == nil {
			readErr = closeErr
		}
		if trimmed := strings.TrimSpace(string(body)); trimmed != "" {
			details += ": " + trimmed
		}
		if readErr != nil {
			details += fmt.Sprintf(" (read body: %v)", readErr)
		}
	}

	return fmt.Errorf("websocket handshake failed for %s: %s: %w", redactWSURLKey(wsURL), details, err)
}

func isInvalidMachineAPIKeyResponse(resp *http.Response) bool {
	if resp == nil || resp.StatusCode != http.StatusUnauthorized || resp.Body == nil {
		return false
	}

	body, readErr := io.ReadAll(io.LimitReader(resp.Body, 1024))
	if closeErr := resp.Body.Close(); closeErr != nil && readErr == nil {
		readErr = closeErr
	}
	resp.Body = io.NopCloser(strings.NewReader(string(body)))
	if readErr != nil {
		return false
	}

	var payload struct {
		Error string `json:"error"`
	}
	if err := json.Unmarshal(body, &payload); err != nil {
		return false
	}
	return strings.EqualFold(strings.TrimSpace(payload.Error), "invalid machine api key")
}

func redactWSURLKey(rawURL string) string {
	parsed, err := url.Parse(rawURL)
	if err != nil {
		return "<invalid websocket url>"
	}
	query := parsed.Query()
	if query.Has("key") {
		query.Set("key", "<redacted>")
		parsed.RawQuery = query.Encode()
	}
	return parsed.String()
}

// detectRuntimes returns the names of available runtimes on this machine.
// Builtin OpenCode and external OpenCode are advertised separately.
func detectRuntimes() []string {
	var found []string
	for _, rt := range runtimeconfig.AllRuntimeTypes {
		binary := runtimeconfig.DefaultBinary(rt)
		probe := runtimeconfig.ProbeRuntimeBinary(rt, binary)
		if !probe.Capable {
			continue
		}
		found = append(found, string(rt))
	}
	return found
}

func (c *Protocol) readyMessage() (map[string]any, []string) {
	// Capture the ready snapshot under the same turn-state publication ordering
	// boundary as incremental publications, so the snapshot watermark equals the
	// revision counter at capture time and every later increment is strictly
	// newer.
	c.turnStatePublicationMu.Lock()
	defer c.turnStatePublicationMu.Unlock()
	hostname, _ := os.Hostname()
	if hostname == "" {
		hostname = "unknown"
	}
	runtimes := detectRuntimes()
	runningSessions := c.runningAgentTeamSessions()

	msg := map[string]any{
		"type": "ready",
		"capabilities": []string{
			"agent:start", "agent:stop", "agent:interrupt", "agent:deliver",
			"agent:sleep", "agent:reset-workspace", "agent:permission-mode",
			"agent:maas-direct", "agent:runtime-session-authority", "agent:local-session-resume",
			"workspace:files", "workspace:target-path",
			"recovery:restore", "recovery:gc",
			"wake:status", transientWakeStatusDesktopCapability,
			daemonTransportCapabilityV1, daemonTransportElasticRequestCapability, daemonEndpointsCapabilityV1,
		},
		"tokenRefreshProof": c.ensureTokenRefreshProof(),
		"runtimes":          runtimes,
		"runtimeDefaults":   profile.DetectRuntimeDefaults(runtimes),
		"runningAgents":     sessionAgentIDs(runningSessions),
		"hostname":          hostname,
		"os":                runtime.GOOS + " " + runtime.GOARCH,
		"daemonVersion":     version.DaemonWireVersion(),
		"daemonTransport":   c.daemonTransportReady(),
	}
	if runningAgentSessions := sessionAgentSessionIDs(runningSessions); len(runningAgentSessions) > 0 {
		msg["runningAgentSessions"] = runningAgentSessions
	}
	if runningAgentTurnStates := sessionAgentTurnStates(runningSessions); len(runningAgentTurnStates) > 0 {
		msg["runningAgentTurnStates"] = runningAgentTurnStates
	}
	// Every modern ready publication carries an explicit snapshot watermark,
	// captured under the same turn-state publication mutex that guards
	// incremental revision allocation, even when runningAgentTurnStates is empty
	// or no agents are running. An empty snapshot is still an authoritative
	// current-epoch snapshot: its watermark (the revision counter at capture
	// time) lets the server retire prior per-agent state while ordering against
	// every increment allocated after this boundary.
	msg["runningAgentTurnStatesWatermark"] = c.turnStateRevision
	return msg, runtimes
}

func (c *Protocol) ensureTokenRefreshProof() string {
	c.mu.Lock()
	defer c.mu.Unlock()

	if token := strings.TrimSpace(c.tokenRefreshProof); token != "" {
		return token
	}
	token, err := newTokenRefreshProof()
	if err != nil {
		slog.Warn("AgentTeam: failed to generate token refresh proof", "channel", c.id, "error", err)
		return ""
	}
	c.tokenRefreshProof = token
	return token
}

func (c *Protocol) currentTokenRefreshProof() string {
	c.mu.Lock()
	defer c.mu.Unlock()
	return strings.TrimSpace(c.tokenRefreshProof)
}

func newTokenRefreshProof() (string, error) {
	var buffer [32]byte
	if _, err := rand.Read(buffer[:]); err != nil {
		return "", err
	}
	return base64.RawURLEncoding.EncodeToString(buffer[:]), nil
}

func sessionAgentIDs(sessions []types.SessionSnapshot) []string {
	if len(sessions) == 0 {
		return nil
	}
	ids := make([]string, 0, len(sessions))
	seen := make(map[string]struct{}, len(sessions))
	for _, session := range sessions {
		agentID := sessionServerAgentID(session)
		if agentID == "" {
			continue
		}
		if _, ok := seen[agentID]; ok {
			continue
		}
		seen[agentID] = struct{}{}
		ids = append(ids, agentID)
	}
	return ids
}

func sessionAgentSessionIDs(sessions []types.SessionSnapshot) map[string]string {
	if len(sessions) == 0 {
		return nil
	}
	ids := make(map[string]string, len(sessions))
	for _, session := range sessions {
		agentID := sessionServerAgentID(session)
		sessionID := strings.TrimSpace(session.RuntimeSessionID)
		if agentID == "" || sessionID == "" {
			continue
		}
		if _, ok := ids[agentID]; ok {
			continue
		}
		ids[agentID] = sessionID
	}
	if len(ids) == 0 {
		return nil
	}
	return ids
}

// sessionAgentTurnStates returns the daemon-derived real turn execution state
// (active/queued/idle) for each running session agent, used to replay the
// recovery working signal across a reconnect.
func sessionAgentTurnStates(sessions []types.SessionSnapshot) map[string]string {
	if len(sessions) == 0 {
		return nil
	}
	states := make(map[string]string, len(sessions))
	for _, session := range sessions {
		agentID := sessionServerAgentID(session)
		state := strings.TrimSpace(session.TurnState)
		if agentID == "" || state == "" {
			continue
		}
		if _, ok := states[agentID]; ok {
			continue
		}
		states[agentID] = state
	}
	if len(states) == 0 {
		return nil
	}
	return states
}

func (c *Protocol) runningAgentTeamSessions() []types.SessionSnapshot {
	lister, ok := c.disp.(agentSessionSnapshotLister)
	if !ok || lister == nil {
		return nil
	}
	snapshots := lister.Sessions()
	if len(snapshots) == 0 {
		return nil
	}
	sessions := make([]types.SessionSnapshot, 0, len(snapshots))
	for _, snapshot := range snapshots {
		if !isRunningAgentTeamSessionSnapshot(c.id, snapshot) {
			continue
		}
		sessions = append(sessions, snapshot)
	}
	return sessions
}

func isRunningAgentTeamSessionSnapshot(channelID string, snapshot types.SessionSnapshot) bool {
	if strings.TrimSpace(snapshot.ChannelID) != strings.TrimSpace(channelID) {
		return false
	}
	if sessionServerAgentID(snapshot) == "" {
		return false
	}
	status := strings.TrimSpace(snapshot.Status)
	return status != "" && status != "stopped"
}

func sessionServerAgentID(snapshot types.SessionSnapshot) string {
	if serverAgentID := strings.TrimSpace(snapshot.ServerAgentID); serverAgentID != "" {
		return serverAgentID
	}
	return strings.TrimSpace(snapshot.AgentID)
}

func (c *Protocol) runningAgentTeamSessionForAgent(agentID string) (types.SessionSnapshot, bool) {
	agentID = strings.TrimSpace(agentID)
	if agentID == "" {
		return types.SessionSnapshot{}, false
	}
	for _, snapshot := range c.runningAgentTeamSessions() {
		if strings.TrimSpace(snapshot.AgentID) == agentID || sessionServerAgentID(snapshot) == agentID {
			return snapshot, true
		}
	}
	return types.SessionSnapshot{}, false
}

func (c *Protocol) sendReadyMessage() ([]string, error) {
	msg, runtimes := c.readyMessage()
	return runtimes, c.sendJSONCritical(msg)
}

func (c *Protocol) replaceServerCapabilities(capabilities []string) {
	next := make(map[string]struct{}, len(capabilities))
	for _, capability := range capabilities {
		if capability = strings.TrimSpace(capability); capability != "" {
			next[capability] = struct{}{}
		}
	}

	c.mu.Lock()
	c.serverCapabilities = next
	c.mu.Unlock()
}

func (c *Protocol) hasServerCapability(capability string) bool {
	capability = strings.TrimSpace(capability)
	if capability == "" {
		return false
	}

	c.mu.Lock()
	_, ok := c.serverCapabilities[capability]
	c.mu.Unlock()
	return ok
}

// ResendMachineCapabilities refreshes the machine capability payload on the
// connected AgentTeam server after local runtime detection changes.
func (c *Protocol) ResendMachineCapabilities() {
	runtimes, err := c.sendReadyMessage()
	if err != nil {
		slog.Warn("AgentTeam: failed to resend machine capabilities", "channel", c.id, "server_url", c.ServerURL(), "error", err)
		return
	}
	slog.Info("AgentTeam: resent machine capabilities", "channel", c.id, "server_url", c.ServerURL(), "runtimes", runtimes)
}

// onConnect sends the ready message and starts the agent sync ticker
// after a successful WebSocket connection.
func (c *Protocol) onConnect(ctx context.Context) {
	c.rotateDaemonTransportSessionGeneration()

	// Eagerly exchange the machine API key for a user session token in the
	// background so downstream operations (memory pre-clone, forwarder calls)
	// can use it without a lazy first-call penalty.
	c.ResetUserTokenFailures()
	c.startConnectionRunTask(ctx, func(ctx context.Context) {
		handshakeTimer := time.NewTimer(2 * time.Second)
		defer handshakeTimer.Stop()
		select {
		case <-handshakeTimer.C:
		case <-ctx.Done():
			return
		}
		tokenCtx, cancel := context.WithTimeout(ctx, 10*time.Second)
		defer cancel()
		if _, err := c.UserToken(tokenCtx); err != nil {
			slog.Warn("AgentTeam: initial user token exchange failed; will retry on demand",
				"channel", c.id, "error", err)
		}
	})

	if _, err := c.sendReadyMessage(); err != nil {
		slog.Warn("AgentTeam: failed to enqueue ready message", "channel", c.id, "error", err)
	}

	// Request the server's agent list for this machine and start periodic sync.
	if err := c.sendJSON(map[string]string{"type": "machine:agent:list"}); err != nil {
		slog.Warn("AgentTeam: failed to enqueue initial agent list request", "channel", c.id, "error", err)
	}
	c.emitRunningAgentTelemetrySnapshot()
	c.startAgentSyncTicker(ctx)
}

func (c *Protocol) emitRunningAgentTelemetrySnapshot() {
	for _, snapshot := range c.runningAgentTeamSessions() {
		c.emitAgentTeamSessionTelemetrySnapshot(sessionServerAgentID(snapshot))
	}
}

func (c *Protocol) emitSessionStateTelemetry(agentID string) {
	c.emitAgentTeamSessionTelemetrySnapshot(agentID)
}

// emitAgentTeamSessionTelemetrySnapshot publishes one agent's session/status/activity/
// turn-state telemetry under the turn-state publication ordering boundary. It
// re-reads the CURRENT session snapshot fresh inside the boundary (never stamps a
// precomputed stale event after waiting for the lock) and allocates a strictly
// newer revision for each turn-state increment, so the server can order it
// against the ready snapshot watermark and earlier increments.
func (c *Protocol) emitAgentTeamSessionTelemetrySnapshot(agentID string) {
	c.turnStatePublicationMu.Lock()
	defer c.turnStatePublicationMu.Unlock()
	snapshot, ok := c.runningAgentTeamSessionForAgent(agentID)
	if !ok {
		return
	}
	serverAgentID := sessionServerAgentID(snapshot)
	if serverAgentID == "" {
		return
	}
	sessionID := strings.TrimSpace(snapshot.RuntimeSessionID)
	if sessionID != "" {
		// Emit the session event FIRST so the server learns the current sessionId
		// before any turn state that belongs to it (same FIFO sendCh). A stale
		// turn_state from a replaced session is then dropped by the server fence.
		c.Emit(types.SessionEvent{
			AgentID:   serverAgentID,
			SessionID: sessionID,
		})
	}
	c.Emit(types.StatusEvent{
		AgentID: serverAgentID,
		Status:  "active",
	})
	activity, detail := sessionActivity(snapshot)
	c.Emit(types.ActivityEvent{
		AgentID:  serverAgentID,
		Activity: activity,
		Detail:   detail,
	})
	// The authoritative turn-state publication point: every ready/reconnect and
	// every session-state-change snapshot carries the current active/queued/idle
	// signal independent of optional driver events. A non-empty state only; empty
	// keeps the server's Hub unknown (fail-safe). The sessionId is included so the
	// server can fence stale turn states from a replaced session, the epoch (this
	// connection's daemon transport session generation) so the server can drop
	// stale-epoch events, and a strictly-newer revision so the server can drop an
	// old/equal increment that must never overwrite a newer snapshot.
	if turnState := strings.TrimSpace(snapshot.TurnState); turnState != "" {
		c.turnStateRevision++
		c.Emit(types.TurnStateEvent{
			AgentID:   serverAgentID,
			TurnState: turnState,
			SessionID: sessionID,
			Epoch:     c.currentDaemonTransportSessionGeneration(),
			Revision:  c.turnStateRevision,
		})
	}
}

func (c *Protocol) startSessionStateTelemetry(ctx context.Context) func() {
	subscriber, ok := c.disp.(sessionStateChangeSubscriber)
	if !ok || subscriber == nil {
		return func() {}
	}
	sub := subscriber.SubscribeSessionStateChanges("agentteam_channel_" + c.id)
	if sub == nil {
		return func() {}
	}

	reporterCtx, cancel := context.WithCancel(ctx)
	done := make(chan struct{})
	started := c.startConnectionRunTask(reporterCtx, func(ctx context.Context) {
		defer close(done)
		defer sub.Unsubscribe()
		for {
			select {
			case <-ctx.Done():
				return
			case ev, ok := <-sub.Events():
				if !ok {
					return
				}
				agentID := strings.TrimSpace(ev.Context.AgentID)
				if agentID == "" {
					c.emitRunningAgentTelemetrySnapshot()
					continue
				}
				c.emitSessionStateTelemetry(agentID)
			}
		}
	})
	if !started {
		cancel()
		sub.Unsubscribe()
		close(done)
	}

	return func() {
		cancel()
		sub.Unsubscribe()
		<-done
	}
}

func sessionActivity(snapshot types.SessionSnapshot) (string, string) {
	activity := strings.TrimSpace(snapshot.Activity)
	detail := strings.TrimSpace(snapshot.ActivityDetail)
	if activity == "" {
		activity = "online"
	}
	return activity, detail
}

const agentSyncInterval = 30 * time.Minute

// startAgentSyncTicker periodically requests the server's agent list
// to keep local agent configs in sync.
func (c *Protocol) startAgentSyncTicker(ctx context.Context) {
	c.syncTickerLifecycleMu.Lock()
	defer c.syncTickerLifecycleMu.Unlock()

	// A replacement may only publish after the exact previous ticker has
	// unwound. Across reconnects the connection-run join already provides this
	// boundary; the explicit wait also keeps direct/repeated calls deterministic.
	c.mu.Lock()
	previous := c.syncTicker
	c.mu.Unlock()
	if previous != nil {
		previous.cancel()
		<-previous.done
	}

	tickerCtx, cancel := context.WithCancel(ctx)
	state := &agentSyncTicker{
		cancel: cancel,
		done:   make(chan struct{}),
	}

	c.mu.Lock()
	c.syncTicker = state
	c.mu.Unlock()

	started := c.startConnectionRunTask(tickerCtx, func(ctx context.Context) {
		defer close(state.done)
		defer c.clearAgentSyncTicker(state)

		ticker := time.NewTicker(agentSyncInterval)
		defer ticker.Stop()
		for {
			select {
			case <-ctx.Done():
				return
			case <-ticker.C:
				c.sendJSON(map[string]string{"type": "machine:agent:list"})
			}
		}
	})
	if !started {
		cancel()
		c.clearAgentSyncTicker(state)
		close(state.done)
	}
}

func (c *Protocol) clearAgentSyncTicker(state *agentSyncTicker) {
	c.mu.Lock()
	defer c.mu.Unlock()
	if c.syncTicker == state {
		c.syncTicker = nil
	}
}

// requestAgentListSync sends a machine:agent:list request to the server and
// blocks until the next machine:agent:list:response is processed (i.e.
// syncAgents completes). Concurrent callers share a single in-flight request
// so one response wakes all waiters from the same local sync generation.
func (c *Protocol) requestAgentListSync(timeout time.Duration, contexts ...context.Context) bool {
	ctx := firstConnectionTaskContext(contexts)
	waiter, shouldSend := c.beginAgentListSync()

	if shouldSend {
		if err := c.sendJSONCritical(map[string]string{"type": "machine:agent:list"}); err != nil {
			c.completeAgentListSync(false)
			slog.Warn("AgentTeam: failed to enqueue agent list sync request", "channel", c.id, "error", err)
			return false
		}
	}

	timer := time.NewTimer(timeout)
	defer timer.Stop()

	select {
	case ok := <-waiter:
		return ok
	case <-ctx.Done():
		c.removeAgentListSyncWaiter(waiter)
		return false
	case <-timer.C:
		c.removeAgentListSyncWaiter(waiter)
		slog.Warn("AgentTeam: agent list sync timed out", "channel", c.id, "timeout", timeout)
		return false
	}
}

func (c *Protocol) beginAgentListSync() (chan bool, bool) {
	waiter := make(chan bool, 1)

	c.mu.Lock()
	defer c.mu.Unlock()

	shouldSend := !c.agentListSyncInFlight
	c.agentListSyncInFlight = true
	c.agentListSyncWaiters = append(c.agentListSyncWaiters, waiter)
	return waiter, shouldSend
}

func (c *Protocol) completeAgentListSync(ok bool) {
	c.mu.Lock()
	waiters := c.agentListSyncWaiters
	c.agentListSyncWaiters = nil
	c.agentListSyncInFlight = false
	c.mu.Unlock()

	for _, waiter := range waiters {
		waiter <- ok
		close(waiter)
	}
}

func (c *Protocol) removeAgentListSyncWaiter(target chan bool) {
	c.mu.Lock()
	defer c.mu.Unlock()

	for i, waiter := range c.agentListSyncWaiters {
		if waiter != target {
			continue
		}
		c.agentListSyncWaiters = append(c.agentListSyncWaiters[:i], c.agentListSyncWaiters[i+1:]...)
		break
	}
	if len(c.agentListSyncWaiters) == 0 {
		c.agentListSyncWaiters = nil
		c.agentListSyncInFlight = false
	}
}

type workspaceListResult struct {
	workspaces []types.AgentTeamWorkspaceSummary
	err        error
}

// WorkspaceList requests the user's AgentTeam workspaces from the server over the
// authenticated daemon websocket.
func (c *Protocol) WorkspaceList(ctx context.Context) ([]types.AgentTeamWorkspaceSummary, error) {
	if c == nil {
		return nil, fmt.Errorf("agentteam channel is nil")
	}
	c.wsMu.Lock()
	connected := c.conn != nil
	c.wsMu.Unlock()
	if !connected {
		return nil, fmt.Errorf("agentteam websocket is not connected")
	}

	requestID := fmt.Sprintf("workspace_list_%d", time.Now().UnixNano())
	resultCh := make(chan workspaceListResult, 1)
	c.mu.Lock()
	if c.workspaceListRequests == nil {
		c.workspaceListRequests = make(map[string]chan workspaceListResult)
	}
	c.workspaceListRequests[requestID] = resultCh
	c.mu.Unlock()
	defer func() {
		c.mu.Lock()
		delete(c.workspaceListRequests, requestID)
		c.mu.Unlock()
	}()

	if err := c.sendJSONBlocking(ctx, map[string]string{
		"type":      "machine:workspaces:list",
		"requestId": requestID,
	}); err != nil {
		return nil, fmt.Errorf("request agentteam workspaces: %w", err)
	}

	select {
	case <-ctx.Done():
		if errors.Is(ctx.Err(), context.DeadlineExceeded) {
			return nil, fmt.Errorf("%w: %w", types.ErrAgentTeamWorkspaceListTimeout, ctx.Err())
		}
		return nil, ctx.Err()
	case result := <-resultCh:
		return result.workspaces, result.err
	}
}

// ---------------------------------------------------------------------------
// Read/write pumps
// ---------------------------------------------------------------------------

// runPumps starts the read and write goroutines for the WebSocket connection.
// When either pump exits, the connection is closed and both goroutines are
// waited upon before returning.
func (c *Protocol) runPumps(ctx context.Context, started ...chan struct{}) {
	var startedCh chan struct{}
	if len(started) > 0 {
		startedCh = started[0]
	}
	notifyStarted := func() {
		if startedCh == nil {
			return
		}
		close(startedCh)
		startedCh = nil
	}

	c.wsMu.Lock()
	conn := c.conn
	c.wsMu.Unlock()
	if conn == nil {
		notifyStarted()
		return
	}

	connCtx, connCancel := context.WithCancel(ctx)
	defer connCancel()

	// Set up read deadline and pong handler so stale connections are detected.
	// The writePump sends pings every pingPeriod; if no pong arrives within
	// pongWait, ReadMessage will fail with a deadline exceeded error,
	// breaking the readPump and triggering reconnection.
	if err := conn.SetReadDeadline(time.Now().Add(pongWait)); err != nil {
		slog.Warn("agentteam websocket: set initial read deadline", "error", err)
		notifyStarted()
		return
	}
	conn.SetPongHandler(func(string) error {
		return conn.SetReadDeadline(time.Now().Add(pongWait))
	})

	type pumpResult struct {
		name string
		err  error
	}
	errCh := make(chan pumpResult, 2)
	var wg sync.WaitGroup
	wg.Add(2)

	go func() {
		defer wg.Done()
		errCh <- pumpResult{name: "read", err: c.readPump(connCtx, conn)}
	}()
	go func() {
		defer wg.Done()
		errCh <- pumpResult{name: "write", err: c.writePump(connCtx, conn)}
	}()

	notifyStarted()

	select {
	case result := <-errCh:
		if result.err != nil && !isNormalClose(result.err) {
			attrs := []any{
				"channel", c.id,
				"server_url", c.ServerURL(),
				"pump", result.name,
				"error", result.err,
			}
			var closeErr *websocket.CloseError
			if errors.As(result.err, &closeErr) {
				attrs = append(attrs, "close_code", closeErr.Code)
			}
			slog.Warn("AgentTeam WebSocket pump stopped unexpectedly", attrs...)
		}
	case <-ctx.Done():
	}

	connCancel()
	// Send the close control frame without racing the write pump. Gorilla
	// allows WriteControl to run concurrently with the single data writer,
	// while WriteMessage would contend with writePump's text/ping writes.
	_ = conn.WriteControl(
		websocket.CloseMessage,
		websocket.FormatCloseMessage(websocket.CloseNormalClosure, ""),
		time.Now().Add(writeWait),
	)
	_ = conn.Close()

	c.wsMu.Lock()
	if c.conn == conn {
		c.conn = nil
		c.notifyConnectionChangeLocked()
	}
	c.wsMu.Unlock()

	wg.Wait()
	c.drainSendQueue()
}

func (c *Protocol) drainSendQueue() int {
	drained := 0
	for {
		select {
		case <-c.sendCh:
			drained++
		default:
			return drained
		}
	}
}

// readPump reads messages from the WebSocket connection and dispatches them.
func (c *Protocol) readPump(ctx context.Context, conn *websocket.Conn) error {
	for {
		_, msg, err := conn.ReadMessage()
		if err != nil {
			return err
		}
		c.dispatch(ctx, msg)
	}
}

// writePump sends messages from the send channel and periodic pings
// over the WebSocket connection.
func (c *Protocol) writePump(ctx context.Context, conn *websocket.Conn) error {
	ticker := time.NewTicker(pingPeriod)
	defer ticker.Stop()

	for {
		select {
		case <-ctx.Done():
			return nil
		case data, ok := <-c.sendCh:
			if !ok {
				return nil
			}
			if ctx.Err() != nil {
				return nil
			}
			_ = conn.SetWriteDeadline(time.Now().Add(writeWait))
			if err := conn.WriteMessage(websocket.TextMessage, data); err != nil {
				return err
			}
		case <-ticker.C:
			if ctx.Err() != nil {
				return nil
			}
			_ = conn.SetWriteDeadline(time.Now().Add(writeWait))
			if err := conn.WriteMessage(websocket.PingMessage, nil); err != nil {
				return err
			}
		}
	}
}

// ---------------------------------------------------------------------------
// WebSocket send helper
// ---------------------------------------------------------------------------

// sendJSON marshals v and enqueues it for sending over the WebSocket without
// blocking. It returns an error when the buffer is full so callers that care
// about delivery can observe the failure instead of silently losing messages.
func (c *Protocol) sendJSON(v any) error {
	data, err := json.Marshal(v)
	if err != nil {
		slog.Debug("AgentTeam: marshal error", "channel", c.id, "error", err)
		return err
	}

	c.sendMu.Lock()
	defer c.sendMu.Unlock()
	select {
	case c.sendCh <- data:
		return nil
	default:
		c.recordSendDrop(data, "queue_full")
		return errSendBufferFull
	}
}

// sendJSONTelemetry preserves queue headroom for wake/runtime/recovery evidence.
// Activity and trajectory are live projections and are safe to shed under
// backpressure because ready/session snapshots rebuild current state after a
// reconnect. They must not crowd out state-transition evidence.
func (c *Protocol) sendJSONTelemetry(v any) error {
	data, err := json.Marshal(v)
	if err != nil {
		slog.Debug("AgentTeam: marshal error", "channel", c.id, "error", err)
		return err
	}

	c.sendMu.Lock()
	defer c.sendMu.Unlock()
	if len(c.sendCh) >= telemetryQueueLimit(cap(c.sendCh)) {
		c.recordSendDrop(data, "telemetry_headroom")
		return errSendBufferFull
	}
	select {
	case c.sendCh <- data:
		return nil
	default:
		c.recordSendDrop(data, "queue_full")
		return errSendBufferFull
	}
}

func telemetryQueueLimit(capacity int) int {
	if capacity <= 1 {
		return capacity
	}
	spare := telemetrySendQueueSpare
	if spare >= capacity {
		spare = max(1, capacity/4)
	}
	return max(1, capacity-spare)
}

func (c *Protocol) recordSendDrop(data []byte, reason string) {
	total := c.metrics.sendDropped.Add(1)
	// Keep a brief burst plus periodic totals. A disconnected multi-agent
	// session can otherwise emit hundreds of identical warning lines and hide
	// the first transport failure that caused the backpressure.
	if total > 3 && total%100 != 0 && !strings.HasPrefix(reason, "critical_") {
		return
	}
	slog.Warn("AgentTeam: outbound message dropped",
		"channel", c.id,
		"message_type", outboundMessageType(data),
		"reason", reason,
		"size", len(data),
		"queue_len", len(c.sendCh),
		"queue_cap", cap(c.sendCh),
		"total_dropped", total,
	)
}

func outboundMessageType(data []byte) string {
	var envelope struct {
		Type string `json:"type"`
	}
	if err := json.Unmarshal(data, &envelope); err != nil {
		return "<invalid-json>"
	}
	if messageType := strings.TrimSpace(envelope.Type); messageType != "" {
		return messageType
	}
	return "<unknown>"
}

// sendJSONCritical enqueues messages that are part of the protocol handshake
// or request/ack path. It waits briefly for writePump to make space so critical
// messages are not dropped just because transient telemetry filled the queue.
func (c *Protocol) sendJSONCritical(v any) error {
	ctx, cancel := context.WithTimeout(context.Background(), criticalSendTimeout)
	defer cancel()
	return c.sendJSONBlocking(ctx, v)
}

func (c *Protocol) sendJSONBlocking(ctx context.Context, v any) error {
	data, err := json.Marshal(v)
	if err != nil {
		slog.Debug("AgentTeam: marshal error", "channel", c.id, "error", err)
		return err
	}

	c.sendMu.Lock()
	defer c.sendMu.Unlock()
	select {
	case <-ctx.Done():
		c.recordSendDrop(data, "critical_context_done")
		return ctx.Err()
	default:
	}

	select {
	case c.sendCh <- data:
		return nil
	case <-ctx.Done():
		c.recordSendDrop(data, "critical_timeout")
		return ctx.Err()
	}
}

// ---------------------------------------------------------------------------
// Agent event reporting (via WebSocket)
// ---------------------------------------------------------------------------

// Emit implements types.EventReporter. It dispatches on the concrete
// AgentEvent type and sends the corresponding WebSocket message to the
// AgentTeam server.
func (c *Protocol) Emit(event types.AgentEvent) {
	switch ev := event.(type) {
	case types.StatusEvent:
		msg := map[string]any{
			"type":    "agent:status",
			"agentId": ev.AgentID,
			"status":  ev.Status,
		}
		if expected := strings.TrimSpace(ev.ExpectedSessionID); expected != "" {
			// A terminal lifecycle frame carries the dying generation's prior
			// session identity so the server applies it only while its current
			// agent session still matches (generation fence).
			msg["expectedSessionId"] = expected
		}
		c.sendJSON(msg)
	case types.ActivityEvent:
		msg := map[string]any{
			"type":     "agent:activity",
			"agentId":  ev.AgentID,
			"activity": ev.Activity,
			"detail":   ev.Detail,
		}
		if expected := strings.TrimSpace(ev.ExpectedSessionID); expected != "" {
			// A terminal lifecycle frame carries the dying generation's prior
			// session identity so the server applies it only while its current
			// agent session still matches (generation fence).
			msg["expectedSessionId"] = expected
		}
		c.sendJSONTelemetry(msg)
	case types.SessionEvent:
		msg := map[string]any{
			"type":      "agent:session",
			"agentId":   ev.AgentID,
			"sessionId": ev.SessionID,
		}
		if expected := strings.TrimSpace(ev.ExpectedSessionID); expected != "" {
			// A terminal clear carries the dying generation's prior session
			// identity so the server applies the clear only while its current
			// agent session still matches (compare-and-clear fence).
			msg["expectedSessionId"] = expected
		}
		c.sendJSON(msg)
	case types.TurnStateEvent:
		msg := map[string]any{
			"type":      "agent:turn_state",
			"agentId":   ev.AgentID,
			"turnState": ev.TurnState,
			"sessionId": ev.SessionID,
		}
		if epoch := strings.TrimSpace(ev.Epoch); epoch != "" {
			msg["epoch"] = epoch
		}
		if ev.Revision > 0 {
			msg["revision"] = ev.Revision
		}
		c.sendJSON(msg)
	case types.TrajectoryEvent:
		if len(ev.Entries) == 0 {
			return
		}
		payload := map[string]any{
			"type":      "agent:trajectory",
			"agentId":   ev.AgentID,
			"sessionId": ev.SessionID,
			"entries":   ev.Entries,
		}
		if ev.Timestamp > 0 {
			payload["timestamp"] = ev.Timestamp
		}
		c.sendJSONTelemetry(payload)
	case types.WakeStatusEvent:
		c.sendJSONCritical(types.AgentWakeStatusMsg{
			Type:           "agent:wake_status",
			AgentID:        ev.AgentID,
			MessageID:      ev.MessageID,
			WakeToken:      ev.WakeToken,
			Target:         ev.Target,
			Status:         ev.Status,
			ReplyMessageID: ev.ReplyMessageID,
			Detail:         ev.Detail,
			Timestamp:      time.Now().UTC().Format(time.RFC3339),
		})
	case types.RecoveryPointPreparedEvent:
		c.sendJSONCritical(map[string]any{
			"type":        "agent:recovery_point_prepared",
			"agentId":     ev.AgentID,
			"messageId":   ev.MessageID,
			"threadId":    ev.ThreadID,
			"strategy":    ev.Strategy,
			"recoveryRef": ev.RecoveryRef,
		})
	case types.RuntimeEvent:
		c.sendJSONCritical(agentRuntimeEventMsg{
			Type:           "agent:runtime_event",
			AgentID:        ev.AgentID,
			MessageID:      ev.MessageID,
			WakeToken:      ev.WakeToken,
			EventClass:     ev.EventClass,
			Target:         ev.Target,
			ReplyMessageID: ev.ReplyMessageID,
			Detail:         ev.Detail,
			Timestamp:      time.Now().UTC().Format(time.RFC3339),
		})
	}
}