package wire

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"log/slog"
	"strings"
	"time"

	"github.com/tidbcloud/AgentTeam/desktop/internal/consts"
	"github.com/tidbcloud/AgentTeam/desktop/internal/profile"
	"github.com/tidbcloud/AgentTeam/desktop/internal/types"
	"github.com/tidbcloud/AgentTeam/desktop/internal/wire/mcpformat"
	"github.com/tidbcloud/AgentTeam/desktop/internal/wire/workspacefs"
	"github.com/tidbcloud/AgentTeam/desktop/internal/workspace"
)

const (
	// The server waits 2s for agent:deliver:ack. Keep daemon-local handoff
	// below that window so failures are rejected instead of timing out into
	// server-side start fallback while this read agentteam remains blocked.
	agentDeliverAckHandoffTimeout = 1500 * time.Millisecond
)

// ---------------------------------------------------------------------------
// Message dispatch
// ---------------------------------------------------------------------------

// dispatch parses the message envelope and routes it to the appropriate handler.
func (c *Protocol) dispatch(ctx context.Context, raw []byte) {
	var env envelope
	if err := json.Unmarshal(raw, &env); err != nil {
		slog.Debug("AgentTeam: bad message", "channel", c.id, "error", err)
		return
	}

	switch env.Type {
	case "ping":
		// Application-level ping/pong is a Session Link lifecycle exception: it is
		// handled directly by the carrier dispatch agentteam, not the typed Command Registry.
		c.sendJSON(map[string]string{"type": "pong"})

	case "server:capabilities":
		var message struct {
			Capabilities []string `json:"capabilities"`
		}
		if err := json.Unmarshal(raw, &message); err != nil {
			slog.Debug("AgentTeam: bad server capabilities message", "channel", c.id, "error", err)
			return
		}
		c.replaceServerCapabilities(message.Capabilities)

	case "agent:deliver":
		c.handleDeliver(ctx, raw)

	case "agent:start":
		prepared, ok := c.prepareAgentStart(raw)
		if ok {
			c.startAgentLifecycleTask(ctx, prepared.message.AgentID, func(ctx context.Context) {
				c.handlePreparedAgentStartResult(ctx, raw, prepared)
			})
		}

	case "agent:stop":
		prepared, ok := c.prepareAgentStop(raw)
		if ok {
			c.startAgentLifecycleTask(ctx, prepared.message.AgentID, func(context.Context) {
				c.handlePreparedAgentStop(prepared)
			})
		}

	case "agent:interrupt":
		c.startConnectionRunTask(ctx, func(ctx context.Context) { c.handleAgentInterrupt(ctx, raw) })

	case "agent:sleep":
		// agent:sleep remains a Desktop-advertised lifecycle control compatibility
		// command until Server exposes a public sender and promotes it into the registry.
		c.handleAgentSleep(ctx, raw)

	case "agent:update_identity":
		msg, ok := c.prepareAgentIdentityUpdate(raw)
		if ok {
			c.startAgentLifecycleTask(ctx, msg.AgentID, func(context.Context) {
				c.applyPreparedAgentIdentityUpdate(msg)
			})
		}

	case "agent:reset-workspace":
		c.startConnectionRunTask(ctx, func(ctx context.Context) { c.handleResetWorkspace(ctx, raw) })

	case "agent:trajectory:history":
		c.startConnectionRunTask(ctx, func(ctx context.Context) { c.handleTrajectoryHistory(ctx, raw) })

	case "agent:workspace:list":
		c.startConnectionRunTask(ctx, func(ctx context.Context) { c.handleWorkspaceList(ctx, raw) })

	case "agent:workspace:read":
		c.startConnectionRunTask(ctx, func(ctx context.Context) { c.handleWorkspaceRead(ctx, raw) })

	case "agent:workspace:apply_seed_files":
		c.startConnectionRunTask(ctx, func(ctx context.Context) { c.handleWorkspaceApplySeedFiles(ctx, raw) })

	case "agent:workspace:delete":
		c.startConnectionRunTask(ctx, func(ctx context.Context) { c.handleWorkspaceDelete(ctx, raw) })

	case "machine:workspace:scan":
		c.startConnectionRunTask(ctx, func(ctx context.Context) { c.handleMachineScan(ctx, raw) })

	case "machine:workspace:delete":
		c.startConnectionRunTask(ctx, func(ctx context.Context) { c.handleMachineDelete(ctx, raw) })

	case "machine:agent:list:response":
		c.handleAgentListResponse(raw, ctx)

	case "machine:workspaces:list:response":
		c.handleWorkspaceListResponse(raw)

	case "elastic_request_link:open":
		c.startConnectionRunTask(ctx, func(ctx context.Context) { c.handleElasticRequestLinkOpen(ctx, raw) })

	case "elastic_request_link:drain":
		c.handleElasticRequestLinkDrain(raw)

	case "elastic_request_link:cancel":
		c.handleElasticRequestLinkCancel(raw)

	case "remote_request:assign":
		c.handleRemoteRequestAssignment(ctx, nil, raw)

	case "agent:restore_recovery_point":
		c.startConnectionRunTask(ctx, func(ctx context.Context) { c.handleAgentRestoreRecoveryPoint(ctx, raw) })

	case "agent:gc_recovery_points":
		c.startConnectionRunTask(ctx, func(ctx context.Context) { c.handleAgentGCRecoveryPoints(ctx, raw) })

	case "agent:review:summary":
		var msg agentReviewSummaryMsg
		if err := json.Unmarshal(raw, &msg); err != nil {
			slog.Debug("AgentTeam: agent:review:summary parse error", "channel", c.id, "error", err)
			return
		}
		c.startConnectionRunTask(ctx, func(ctx context.Context) { c.handleReviewSummary(ctx, msg) })

	case "agent:review:files":
		var msg agentReviewFilesMsg
		if err := json.Unmarshal(raw, &msg); err != nil {
			slog.Debug("AgentTeam: agent:review:files parse error", "channel", c.id, "error", err)
			return
		}
		c.startConnectionRunTask(ctx, func(ctx context.Context) { c.handleReviewFiles(ctx, msg) })

	case "agent:review:diff":
		var msg agentReviewDiffMsg
		if err := json.Unmarshal(raw, &msg); err != nil {
			slog.Debug("AgentTeam: agent:review:diff parse error", "channel", c.id, "error", err)
			return
		}
		c.startConnectionRunTask(ctx, func(ctx context.Context) { c.handleReviewDiff(ctx, msg) })

	case "agent:output-target:validate":
		var msg agentOutputTargetValidateMsg
		if err := json.Unmarshal(raw, &msg); err != nil {
			slog.Debug("AgentTeam: agent:output-target:validate parse error", "channel", c.id, "error", err)
			return
		}
		c.startConnectionRunTask(ctx, func(ctx context.Context) { c.handleOutputTargetValidate(ctx, msg) })

	case "agent:memory:access-changed":
		var msg memoryAccessChangedMsg
		if err := json.Unmarshal(raw, &msg); err != nil {
			slog.Debug("AgentTeam: agent:memory:access-changed parse error", "channel", c.id, "error", err)
			return
		}
		c.startConnectionRunTask(ctx, func(ctx context.Context) { c.handleMemoryAccessChanged(ctx, msg) })

	case "agent:assets:access-changed":
		var msg assetsAccessChangedMsg
		if err := json.Unmarshal(raw, &msg); err != nil {
			slog.Debug("AgentTeam: agent:assets:access-changed parse error", "channel", c.id, "error", err)
			return
		}
		c.startConnectionRunTask(ctx, func(ctx context.Context) { c.handleAssetsAccessChanged(ctx, msg) })

	default:
		slog.Debug("AgentTeam: unhandled message type", "channel", c.id, "type", env.Type)
	}
}

// ---------------------------------------------------------------------------
// agent:deliver handler
// ---------------------------------------------------------------------------

// handleDeliver processes an agent:deliver message: parses the inner payload,
// deduplicates retransmits, and forwards it to the session handler.
func (c *Protocol) handleDeliver(ctx context.Context, raw []byte) {
	c.handleDeliverWithUnread(ctx, raw, nil)
}

type deliveryHandoffDisposition uint8

const (
	deliveryHandoffRejected deliveryHandoffDisposition = iota
	deliveryHandoffAccepted
	deliveryHandoffNeedsLifecycleStart
)

// handleDeliverWithUnread returns the daemon-local ownership disposition.
// Only a strict Process-generation rejection requests lifecycle fallback;
// malformed or unsupported input is terminally rejected instead of being
// converted into a nil-wake Process start.
func (c *Protocol) handleDeliverWithUnread(ctx context.Context, raw []byte, unreadSummary map[string]int) deliveryHandoffDisposition {
	return c.handleDeliverWithOptions(ctx, raw, unreadSummary, false)
}

func (c *Protocol) handleAgentStartDelivery(ctx context.Context, raw []byte, unreadSummary map[string]int) deliveryHandoffDisposition {
	return c.handleDeliverWithOptions(ctx, raw, unreadSummary, true)
}

func (c *Protocol) handleDeliverWithOptions(
	ctx context.Context,
	raw []byte,
	unreadSummary map[string]int,
	suppressRejectedAck bool,
) deliveryHandoffDisposition {
	var msg agentDeliverMsg
	if err := json.Unmarshal(raw, &msg); err != nil {
		slog.Debug("AgentTeam: agent:deliver parse error", "channel", c.id, "error", err)
		return deliveryHandoffRejected
	}
	deliverySeq := msg.Seq
	if deliverySeq == 0 && msg.DeliverySeq != 0 {
		deliverySeq = msg.DeliverySeq
	}

	slog.Info("AgentTeam: agent:deliver received",
		"channel", c.id, "agent_id", msg.AgentID, "seq", deliverySeq)

	// Parse the inner message first so dedup can prefer the globally unique
	// message ID. Thread-local seq values can repeat across unrelated threads.
	var dm deliverMessage
	if err := json.Unmarshal(msg.Message, &dm); err != nil {
		slog.Debug("AgentTeam: deliver message parse error", "channel", c.id, "error", err)
		c.sendAck(msg.AgentID, "", deliverySeq, "", false, "parse error")
		return deliveryHandoffRejected
	}

	dm.TargetSeq = normalizedDeliverTargetSeq(dm, deliverySeq)
	rawInterruptPolicy := firstNonEmptyAgentInterruptPolicy(msg.InterruptPolicy, dm.InterruptPolicy)
	interruptPolicy, ok := normalizeAgentInterruptPolicy(rawInterruptPolicy)
	if !ok {
		c.sendAck(msg.AgentID, dm.MessageID, deliverySeq, rawInterruptPolicy, false, "unsupported interruptPolicy")
		return deliveryHandoffRejected
	}

	// Dedup only records deliveries that the daemon has actually accepted.
	// Prefer message ID when present; fall back to seq for older payloads.
	h := deliveryDedupHashWithInterruptPolicy(msg.AgentID, deliverySeq, dm.MessageID, interruptPolicy)
	if c.deliveryDedupSeen(h) {
		c.metrics.dedupHits.Add(1)
		c.sendAck(msg.AgentID, dm.MessageID, deliverySeq, interruptPolicy, true, "duplicate")
		slog.Debug("AgentTeam: duplicate agent:deliver skipped",
			"channel", c.id, "agent_id", msg.AgentID, "seq", deliverySeq)
		return deliveryHandoffAccepted
	}

	content := strings.TrimSpace(dm.Content)
	if content == "" && len(dm.Attachments) == 0 {
		slog.Debug("AgentTeam: empty content, skipping", "channel", c.id)
		c.commitDeliveryDedup(h, msg.AgentID)
		c.sendAck(msg.AgentID, dm.MessageID, deliverySeq, interruptPolicy, true, "empty content")
		return deliveryHandoffAccepted
	}

	// Build the reply target from channel info.
	target := deliverMessageTarget(dm)

	if c.suppressStaleDelivery(msg.AgentID, dm, target, deliverySeq, interruptPolicy) {
		c.commitDeliveryDedup(h, msg.AgentID)
		c.sendAck(msg.AgentID, dm.MessageID, deliverySeq, interruptPolicy, true, "stale delivery suppressed")
		return deliveryHandoffAccepted
	}

	authorProfile := dm.AuthorProfile
	authorTarget := firstNonEmptyString(authorProfile.TargetID, authorProfile.MemberID, authorProfile.PrincipalID, authorProfile.DisplayName, dm.SenderName)
	if authorProfile == (types.AuthorProfile{}) && authorTarget != "" {
		authorProfile = types.AuthorProfile{
			ActorType:   dm.SenderType,
			DisplayName: authorTarget,
			TargetID:    authorTarget,
		}
	}

	slog.Info("AgentTeam: inbound message",
		"channel", c.id, "agent_id", msg.AgentID, "author", firstNonEmptyString(authorProfile.DisplayName, authorTarget), "target", target,
		"content", previewText(content, 100))

	workflowContext := dm.WorkflowContext.Normalize()
	if !workflowContext.Valid() {
		workflowContext = types.WorkflowContextFromMetadata(dm.Metadata)
	}

	meta := &types.AgentTeamMeta{
		ID:                       dm.MessageID,
		TargetKey:                strings.TrimSpace(dm.TargetKey),
		TargetSeq:                dm.TargetSeq,
		AuthorTarget:             authorTarget,
		Target:                   target,
		TargetChannelType:        strings.TrimSpace(dm.ChannelType),
		TargetChannelName:        strings.TrimSpace(dm.ChannelName),
		ChannelDisplayName:       strings.TrimSpace(dm.ChannelDisplayName),
		ParentTargetChannelType:  strings.TrimSpace(dm.ParentChannelType),
		ParentTargetChannelName:  strings.TrimSpace(dm.ParentChannelName),
		ParentChannelDisplayName: strings.TrimSpace(dm.ParentChannelDisplayName),
		AgentID:                  msg.AgentID,
		AuthorProfile:            authorProfile,
		Timestamp:                dm.Timestamp,
		MessageType:              dm.MessageType,
		InterruptPolicy:          interruptPolicy,
		Source:                   dm.Source,
		Metadata:                 types.MetadataWithoutWorkflowContext(dm.Metadata),
		WorkflowContext:          workflowContext,
		ActiveTaskTarget:         dm.ActiveTaskTarget,
		TeamContexts:             dm.TeamContexts,
	}

	// Store attachment info so the agent can reference them.
	if len(dm.Attachments) > 0 {
		var ids, names []string
		attachments := make([]types.MessageAttachment, 0, len(dm.Attachments))
		for _, a := range dm.Attachments {
			attachments = append(attachments, types.MessageAttachment{
				ID:          strings.TrimSpace(a.ID),
				Filename:    strings.TrimSpace(a.Filename),
				ContentType: strings.TrimSpace(a.ContentType),
				SizeBytes:   a.SizeBytes,
			})
			if strings.TrimSpace(a.ID) != "" {
				ids = append(ids, a.ID)
			}
			if strings.TrimSpace(a.Filename) != "" {
				names = append(names, a.Filename)
			}
		}
		meta.AttachmentIDs = strings.Join(ids, ",")
		meta.AttachmentNames = strings.Join(names, ",")
		meta.Attachments = attachments
	}

	c.mu.Lock()
	if st, ok := c.agents[msg.AgentID]; ok {
		meta.SessionID = st.Config.SessionID
	}
	c.mu.Unlock()
	if len(unreadSummary) > 0 {
		meta.UnreadSummary = unreadSummary
	}

	input := types.MessageInput{
		Prompt:  content,
		Channel: c.id,
		Meta:    meta,
	}
	m := types.NewMessage(newSessionID(c.id), input)

	if ctx == nil {
		ctx = context.Background()
	}
	handoffCtx, cancel := context.WithTimeout(ctx, agentDeliverAckHandoffTimeout)
	defer cancel()
	if err := c.disp.DispatchForDeliveryAck(handoffCtx, c, m); err != nil {
		if !suppressRejectedAck {
			c.sendAck(msg.AgentID, dm.MessageID, deliverySeq, interruptPolicy, false, err.Error())
		}
		return deliveryHandoffNeedsLifecycleStart
	}
	c.commitDeliveryDedup(h, msg.AgentID)
	c.sendAck(msg.AgentID, dm.MessageID, deliverySeq, interruptPolicy, true, "")
	return deliveryHandoffAccepted
}

func (c *Protocol) suppressStaleDelivery(agentID string, dm deliverMessage, target string, deliverySeq int, interruptPolicy string) bool {
	if c == nil || c.disp == nil || strings.TrimSpace(dm.MessageType) == "reaction" || strings.TrimSpace(interruptPolicy) == agentInterruptPolicyCurrentRun {
		return false
	}
	targetKey := strings.TrimSpace(dm.TargetKey)
	if targetKey == "" || dm.TargetSeq <= 0 {
		return false
	}
	tracker, ok := c.disp.(messageFreshnessTracker)
	if !ok {
		return false
	}
	frontier, ok := tracker.MessageFreshnessSnapshot(agentID, targetKey)
	if !ok || frontier.ObservedSeq <= 0 || dm.TargetSeq > frontier.ObservedSeq {
		return false
	}
	slog.Debug("AgentTeam: stale agent:deliver suppressed",
		"channel", c.id,
		"agent_id", strings.TrimSpace(agentID),
		"delivery_id", deliverySeq,
		"message_id", strings.TrimSpace(dm.MessageID),
		"target", strings.TrimSpace(target),
		"target_key", targetKey,
		"target_seq", dm.TargetSeq,
		"observed_seq", frontier.ObservedSeq,
		"latest_known_seq", frontier.LatestKnownSeq,
		"reason", "already_observed_before_channel_dispatch",
	)
	return true
}

type preparedAgentStart struct {
	message   agentStartMsg
	rejection *agentStartRejection
}

type agentStartRejection struct {
	messageID string
	target    string
	detail    string
}

func (c *Protocol) prepareAgentStartConfigRejection(agentID string, wakeMessage json.RawMessage, parseErr error) *agentStartRejection {
	messageID, target := wakeMessageContext(wakeMessage)
	detail := fmt.Sprintf("invalid agent start config: %v", parseErr)
	slog.Warn("AgentTeam: agent:start rejected invalid config",
		"channel", c.id, "agent_id", agentID, "error", parseErr)
	return &agentStartRejection{
		messageID: messageID,
		target:    target,
		detail:    detail,
	}
}

// resolveRejectedAgentStart closes the Server fallback-start lifecycle even
// though no Process is launched. A seq-zero delivery ACK has no start waiter,
// so wake/runtime evidence and a final authoritative status are required to
// clear Server starting state. The final status reflects Process authority: a
// healthy older generation remains active, otherwise the agent is inactive.
func (c *Protocol) resolveRejectedAgentStart(agentID string, rejection *agentStartRejection) {
	if rejection == nil {
		return
	}
	agentID = strings.TrimSpace(agentID)
	detail := strings.TrimSpace(rejection.detail)
	c.emitAgentStartWakeFailure(agentID, rejection.messageID, rejection.target, detail)
	c.Emit(types.RuntimeEvent{
		AgentID:    agentID,
		MessageID:  rejection.messageID,
		EventClass: consts.RuntimeEventClassRuntimeStartFailed,
		Target:     rejection.target,
		Detail:     detail,
	})
	status := "inactive"
	if c.hasRunningAgentSession(agentID) {
		status = "active"
	}
	if err := c.sendJSONCritical(map[string]any{
		"type":    "agent:status",
		"agentId": agentID,
		"status":  status,
	}); err != nil {
		slog.Warn("AgentTeam: failed to resolve rejected agent:start status",
			"channel", c.id, "agent_id", agentID, "status", status, "error", err)
	}
}

func (c *Protocol) emitAgentStartWakeFailure(agentID, messageID, target, detail string) {
	if strings.TrimSpace(messageID) == "" {
		return
	}
	c.Emit(types.WakeStatusEvent{
		AgentID:   strings.TrimSpace(agentID),
		MessageID: messageID,
		WakeToken: "agent-start:" + messageID,
		Target:    target,
		Status:    consts.WakeStatusFailed,
		Detail:    strings.TrimSpace(detail),
	})
}

func (c *Protocol) emitAgentStartWakeFailures(agentID string, starts []agentStartPayload, err error) {
	if err == nil {
		return
	}
	detail := "Start failed: " + strings.TrimSpace(err.Error())
	seen := make(map[string]struct{}, len(starts))
	for _, start := range starts {
		messageID, target := wakeMessageContext(start.WakeMessage)
		if strings.TrimSpace(messageID) == "" {
			continue
		}
		if _, ok := seen[messageID]; ok {
			continue
		}
		seen[messageID] = struct{}{}
		c.emitAgentStartWakeFailure(agentID, messageID, target, detail)
	}
}

func wakeMessageContext(raw json.RawMessage) (messageID string, target string) {
	if len(raw) == 0 || string(raw) == "null" {
		return "", ""
	}
	var dm deliverMessage
	if err := json.Unmarshal(raw, &dm); err != nil {
		return "", ""
	}
	return dm.MessageID, deliverMessageTarget(dm)
}

func agentStartWakeRequiresReplacement(raw json.RawMessage) bool {
	if len(raw) == 0 || string(raw) == "null" {
		return false
	}
	var message deliverMessage
	if err := json.Unmarshal(raw, &message); err != nil {
		return false
	}
	policy, ok := normalizeAgentInterruptPolicy(message.InterruptPolicy)
	return ok && policy == agentInterruptPolicyCurrentRun
}

func deliverMessageTarget(dm deliverMessage) string {
	target := mcpformat.FormatTarget(dm.ChannelType, dm.ChannelName, dm.ParentChannelType, dm.ParentChannelName)
	if target != "#" || strings.TrimSpace(dm.MessageType) != "task_dispatch" {
		return target
	}
	if taskScopeTarget := taskScopeProjectionTarget(dm.TargetKey); taskScopeTarget != "" {
		return taskScopeTarget
	}
	return target
}

func (c *Protocol) messageFromWakePayload(agentID string, raw json.RawMessage, cfg types.AgentStartConfig, unreadSummary map[string]int) (*types.Message, string, string, error) {
	if len(raw) == 0 || string(raw) == "null" {
		return nil, "", "", nil
	}
	var dm deliverMessage
	if err := json.Unmarshal(raw, &dm); err != nil {
		return nil, "", "", err
	}
	dm.TargetSeq = normalizedDeliverTargetSeq(dm, 0)
	interruptPolicy, ok := normalizeAgentInterruptPolicy(dm.InterruptPolicy)
	if !ok {
		return nil, dm.MessageID, "", fmt.Errorf("unsupported interruptPolicy")
	}
	content := strings.TrimSpace(dm.Content)
	if content == "" && len(dm.Attachments) == 0 {
		return nil, dm.MessageID, interruptPolicy, nil
	}
	target := deliverMessageTarget(dm)
	authorProfile := dm.AuthorProfile
	authorTarget := firstNonEmptyString(authorProfile.TargetID, authorProfile.MemberID, authorProfile.PrincipalID, authorProfile.DisplayName, dm.SenderName)
	if authorProfile == (types.AuthorProfile{}) && authorTarget != "" {
		authorProfile = types.AuthorProfile{
			ActorType:   dm.SenderType,
			DisplayName: authorTarget,
			TargetID:    authorTarget,
		}
	}
	workflowContext := dm.WorkflowContext.Normalize()
	if !workflowContext.Valid() {
		workflowContext = types.WorkflowContextFromMetadata(dm.Metadata)
	}

	meta := &types.AgentTeamMeta{
		ID:                       dm.MessageID,
		TargetKey:                strings.TrimSpace(dm.TargetKey),
		TargetSeq:                dm.TargetSeq,
		AuthorTarget:             authorTarget,
		Target:                   target,
		TargetChannelType:        strings.TrimSpace(dm.ChannelType),
		TargetChannelName:        strings.TrimSpace(dm.ChannelName),
		ChannelDisplayName:       strings.TrimSpace(dm.ChannelDisplayName),
		ParentTargetChannelType:  strings.TrimSpace(dm.ParentChannelType),
		ParentTargetChannelName:  strings.TrimSpace(dm.ParentChannelName),
		ParentChannelDisplayName: strings.TrimSpace(dm.ParentChannelDisplayName),
		AgentID:                  agentID,
		Runtime:                  cfg.Runtime,
		Model:                    cfg.Model,
		ReasoningEffort:          cfg.ReasoningEffort,
		SessionID:                cfg.SessionID,
		AuthorProfile:            authorProfile,
		Timestamp:                dm.Timestamp,
		MessageType:              dm.MessageType,
		InterruptPolicy:          interruptPolicy,
		Source:                   dm.Source,
		Metadata:                 types.MetadataWithoutWorkflowContext(dm.Metadata),
		WorkflowContext:          workflowContext,
		ActiveTaskTarget:         dm.ActiveTaskTarget,
		TeamContexts:             dm.TeamContexts,
	}
	if len(unreadSummary) > 0 {
		meta.UnreadSummary = unreadSummary
	}
	if len(dm.Attachments) > 0 {
		var ids, names []string
		attachments := make([]types.MessageAttachment, 0, len(dm.Attachments))
		for _, a := range dm.Attachments {
			attachments = append(attachments, types.MessageAttachment{
				ID:          strings.TrimSpace(a.ID),
				Filename:    strings.TrimSpace(a.Filename),
				ContentType: strings.TrimSpace(a.ContentType),
				SizeBytes:   a.SizeBytes,
			})
			if strings.TrimSpace(a.ID) != "" {
				ids = append(ids, a.ID)
			}
			if strings.TrimSpace(a.Filename) != "" {
				names = append(names, a.Filename)
			}
		}
		meta.AttachmentIDs = strings.Join(ids, ",")
		meta.AttachmentNames = strings.Join(names, ",")
		meta.Attachments = attachments
	}
	input := types.MessageInput{
		Prompt:  content,
		Channel: c.id,
		Meta:    meta,
	}
	return types.NewMessage(newSessionID(c.id), input), dm.MessageID, interruptPolicy, nil
}

func removeInitialWakePayload(starts []agentStartPayload, initial json.RawMessage) []agentStartPayload {
	if len(starts) == 0 || len(initial) == 0 || string(initial) == "null" {
		return starts
	}
	initialMessageID, _ := wakeMessageContext(initial)
	for i, start := range starts {
		messageID, _ := wakeMessageContext(start.WakeMessage)
		if (strings.TrimSpace(initialMessageID) != "" && messageID == initialMessageID) ||
			(strings.TrimSpace(initialMessageID) == "" && string(start.WakeMessage) == string(initial)) {
			return append(starts[:i], starts[i+1:]...)
		}
	}
	return starts
}

func normalizedDeliverTargetSeq(dm deliverMessage, _ int) int64 {
	if strings.TrimSpace(dm.MessageType) == "reaction" ||
		strings.TrimSpace(dm.Source.RefType) == "task_activity" {
		return 0
	}
	if strings.TrimSpace(dm.ChannelType) == "thread" {
		shortID := mcpformat.NormalizeReplyThreadShortID(dm.ChannelName)
		messageID := strings.TrimSpace(dm.MessageID)
		// Legacy started-thread root payloads may carry the parent-channel seq.
		if shortID != "" && strings.HasPrefix(messageID, shortID) {
			return 0
		}
	}
	if dm.TargetSeq > 0 {
		return dm.TargetSeq
	}
	return 0
}

// sendAck sends an agent:deliver:ack message back to the server.
// The server expects accepted/detail fields; if rejected or timed out
// (~2s) it falls back to agent:start.
func (c *Protocol) sendAck(agentID string, messageID string, seq int, interruptPolicy string, accepted bool, detail string) bool {
	if err := c.sendJSONCritical(deliverAckMsg{
		Type:            "agent:deliver:ack",
		AgentID:         agentID,
		MessageID:       strings.TrimSpace(messageID),
		Seq:             seq,
		InterruptPolicy: strings.TrimSpace(interruptPolicy),
		Accepted:        accepted,
		Detail:          detail,
	}); err != nil {
		slog.Warn("AgentTeam: failed to enqueue deliver ack",
			"channel", c.id, "agent_id", agentID, "message_id", messageID, "accepted", accepted, "error", err)
		return false
	}
	return true
}

func (c *Protocol) deliveryDedupSeen(hash uint64) bool {
	c.mu.Lock()
	defer c.mu.Unlock()
	return c.dedupRing != nil && c.dedupRing.Contains(hash)
}

func (c *Protocol) commitDeliveryDedup(hash uint64, agentID string) {
	c.mu.Lock()
	defer c.mu.Unlock()
	if c.dedupRing != nil {
		c.dedupRing.AddWithKey(hash, agentID)
	}
}

func (c *Protocol) clearDeliveryDedup(agentID string) {
	c.mu.Lock()
	defer c.mu.Unlock()
	if c.dedupRing != nil {
		c.dedupRing.RemoveByKey(agentID)
	}
}

func (c *Protocol) hasRunningAgentSession(agentID string) bool {
	if c == nil || c.disp == nil {
		return false
	}
	snapshot, ok := c.disp.Session(agentID)
	return ok && strings.TrimSpace(snapshot.Status) == "running"
}

const agentInterruptPolicyCurrentRun = "interrupt_current_run"

func normalizeAgentInterruptPolicy(raw string) (string, bool) {
	trimmed := strings.TrimSpace(raw)
	if trimmed == "" {
		return "", true
	}
	if trimmed == agentInterruptPolicyCurrentRun {
		return trimmed, true
	}
	return "", false
}

func (c *Protocol) refreshRunningAgentStartConfig(agentID string) bool {
	if c == nil || c.disp == nil {
		return false
	}
	refresher, ok := c.disp.(runningAgentStartConfigRefresher)
	if !ok || refresher == nil {
		return false
	}
	c.mu.Lock()
	cfg := types.AgentStartConfig{}
	if st, ok := c.agents[agentID]; ok {
		cfg = st.Config
	}
	c.mu.Unlock()
	return refresher.RefreshRunningAgentStartConfig(agentID, cfg)
}

func firstNonEmptyAgentInterruptPolicy(values ...string) string {
	for _, value := range values {
		if trimmed := strings.TrimSpace(value); trimmed != "" {
			return trimmed
		}
	}
	return ""
}

// ---------------------------------------------------------------------------
// Server command handlers
// ---------------------------------------------------------------------------

// handleAgentStart processes an agent:start command: parses the config,
// guards against concurrent starts, spawns the agent process, and delivers
// any wake message or unread summary.
func (c *Protocol) handleAgentStart(ctx context.Context, raw []byte) {
	prepared, ok := c.prepareAgentStart(raw)
	if !ok {
		return
	}
	c.handlePreparedAgentStartResult(ctx, raw, prepared)
}

func (c *Protocol) handlePreparedAgentStartResult(ctx context.Context, raw []byte, prepared preparedAgentStart) {
	if prepared.rejection != nil {
		c.resolveRejectedAgentStart(prepared.message.AgentID, prepared.rejection)
		return
	}
	c.handlePreparedAgentStart(ctx, raw)
}

// prepareAgentStart applies receipt-time config/token state before the
// asynchronous lifecycle owner is queued. This preserves immediate token
// rotation while the receipt-ordered owner chain serializes Dispatcher work.
func (c *Protocol) prepareAgentStart(raw []byte) (preparedAgentStart, bool) {
	var msg agentStartMsg
	if err := json.Unmarshal(raw, &msg); err != nil {
		slog.Debug("AgentTeam: agent:start parse error", "channel", c.id, "error", err)
		return preparedAgentStart{}, false
	}
	prepared := preparedAgentStart{message: msg}
	parsedUnread := parseUnreadSummary(msg.UnreadSummary)
	hasWake := len(msg.WakeMessage) > 0 && string(msg.WakeMessage) != "null"
	hasConfig := len(msg.Config) > 0 && string(msg.Config) != "null"
	var cfg types.AgentStartConfig
	if hasConfig {
		if err := json.Unmarshal(msg.Config, &cfg); err != nil {
			c.mu.Lock()
			if c.agents == nil {
				c.agents = make(map[string]*agentState)
			}
			c.getOrCreateAgentState(msg.AgentID).nextReceiptSequence()
			c.mu.Unlock()
			prepared.rejection = c.prepareAgentStartConfigRejection(msg.AgentID, msg.WakeMessage, err)
			return prepared, true
		}
		if !hasWake && len(parsedUnread) > 0 {
			cfg.UnreadSummary = parsedUnread
		}
	}
	c.mu.Lock()
	if c.agents == nil {
		c.agents = make(map[string]*agentState)
	}
	state := c.getOrCreateAgentState(msg.AgentID)
	receiptSequence := state.nextReceiptSequence()
	if hasConfig {
		if cfg.AuthToken != "" {
			state.Token = cfg.AuthToken
			state.tokenReceiptSequence = receiptSequence
		}
		state.Config = cfg
		state.startConfigReceiptSequence = receiptSequence
		state.Sleeping = false
	}
	c.mu.Unlock()
	if hasWake {
		if rejection := c.validateAgentStartWake(msg.AgentID, msg.WakeMessage); rejection != nil {
			prepared.rejection = rejection
			return prepared, true
		}
	}
	return prepared, true
}

func (c *Protocol) validateAgentStartWake(agentID string, raw json.RawMessage) *agentStartRejection {
	var message deliverMessage
	if err := json.Unmarshal(raw, &message); err != nil {
		detail := fmt.Sprintf("invalid agent:start wake: parse error: %v", err)
		c.sendAck(agentID, "", 0, "", false, detail)
		return &agentStartRejection{detail: detail}
	}
	if _, ok := normalizeAgentInterruptPolicy(message.InterruptPolicy); !ok {
		detail := "invalid agent:start wake: unsupported interruptPolicy"
		c.sendAck(agentID, message.MessageID, 0, message.InterruptPolicy, false, detail)
		return &agentStartRejection{
			messageID: message.MessageID,
			target:    deliverMessageTarget(message),
			detail:    detail,
		}
	}
	return nil
}

func (c *Protocol) handlePreparedAgentStart(ctx context.Context, raw []byte) {
	c.metrics.agentStarts.Add(1)
	var msg agentStartMsg
	if err := json.Unmarshal(raw, &msg); err != nil {
		slog.Debug("AgentTeam: agent:start parse error", "channel", c.id, "error", err)
		return
	}

	slog.Info("AgentTeam: agent:start received",
		"channel", c.id, "agent_id", msg.AgentID)

	// Guard against concurrent starts for the same agent.
	parsedUnread := parseUnreadSummary(msg.UnreadSummary)
	hasWake := len(msg.WakeMessage) > 0 && string(msg.WakeMessage) != "null"
	var startPayload agentStartPayload
	if hasWake {
		startPayload = agentStartPayload{
			WakeMessage:   msg.WakeMessage,
			UnreadSummary: parsedUnread,
		}
	}

	if hasWake && !agentStartWakeRequiresReplacement(msg.WakeMessage) && c.hasRunningAgentSession(msg.AgentID) {
		c.refreshRunningAgentStartConfig(msg.AgentID)
		if c.hasRunningAgentSession(msg.AgentID) {
			slog.Info("AgentTeam: agent:start routed to running session",
				"channel", c.id, "agent_id", msg.AgentID, "has_wake", hasWake)
			envelope, err := json.Marshal(agentDeliverMsg{
				Type:    "agent:deliver",
				AgentID: msg.AgentID,
				Seq:     0,
				Message: msg.WakeMessage,
			})
			if err != nil {
				slog.Debug("AgentTeam: agent:start delivery envelope encode failed",
					"channel", c.id, "agent_id", msg.AgentID, "error", err)
				return
			}
			switch c.handleAgentStartDelivery(ctx, envelope, parsedUnread) {
			case deliveryHandoffAccepted:
				c.Emit(types.StatusEvent{
					AgentID: msg.AgentID,
					Status:  "active",
				})
				return
			case deliveryHandoffRejected:
				// The Process is still healthy; only this malformed/unsupported
				// payload was terminally rejected (with a negative ACK above).
				c.Emit(types.StatusEvent{
					AgentID: msg.AgentID,
					Status:  "active",
				})
				return
			case deliveryHandoffNeedsLifecycleStart:
				slog.Info("AgentTeam: running-session agent:start handoff rejected; replacing Process",
					"channel", c.id, "agent_id", msg.AgentID)
			}
		}
		slog.Info("AgentTeam: agent:start replacing stale running session after config refresh",
			"channel", c.id, "agent_id", msg.AgentID)
	}

	if !hasWake && c.hasRunningAgentSession(msg.AgentID) {
		slog.Info("AgentTeam: agent:start replacing running session",
			"channel", c.id, "agent_id", msg.AgentID)
	}

	// Clear all dedup entries for this agent so that deliveries after
	// restart (which may reuse earlier seq numbers) are not mistaken
	// for duplicates from the previous session. Running sessions keep
	// their dedup state so agent:start fallback cannot re-deliver a
	// message that was already accepted by agent:deliver.
	c.clearDeliveryDedup(msg.AgentID)

	replaceQueuedStart := false
	var pendingStarts []agentStartPayload
	for {
		c.mu.Lock()
		st := c.getOrCreateAgentState(msg.AgentID)
		if replaceQueuedStart {
			st.RestartQueued = false
			replaceQueuedStart = false
		} else {
			if st.Starting {
				st.RestartQueued = true
				if hasWake {
					st.PendingStarts = append(st.PendingStarts, startPayload)
				}
				c.mu.Unlock()
				slog.Debug("AgentTeam: agent:start queued, already starting",
					"channel", c.id, "agent_id", msg.AgentID)
				return
			}
			st.Starting = true
			st.RestartQueued = false
			if hasWake {
				st.PendingStarts = append(st.PendingStarts, startPayload)
			}
		}
		c.mu.Unlock()

		// If the agent doesn't exist in local config yet, request a full agent
		// list sync from the server before attempting to start. This handles
		// the race where the server sends agent:start before the periodic sync
		// has had a chance to pull the latest agent list.
		if _, found := profile.Default().Agents().Get(msg.AgentID); !found {
			slog.Info("AgentTeam: agent not found locally, requesting agent list sync before start",
				"channel", c.id, "agent_id", msg.AgentID)
			c.requestAgentListSync(10*time.Second, ctx)
		}
		c.ensureManagedSkillsReadyForLaunch(ctx, msg.AgentID)

		var initialWakePayload agentStartPayload
		var initialWakeRaw json.RawMessage
		var initialWake *types.Message
		var initialWakeMessageID string
		var initialWakeInterruptPolicy string
		var cfg types.AgentStartConfig
		c.mu.Lock()
		if state, ok := c.agents[msg.AgentID]; ok {
			// Identity/config receipt is immediate even though Dispatcher lifecycle
			// work is sequenced. Fold every update received during catalog/skill
			// preflight into this launch; a later update is queued behind it and
			// will retire the published generation if it arrived after this read.
			cfg = state.Config
			if len(state.PendingStarts) > 0 {
				initialWakePayload = state.PendingStarts[0]
				initialWakeRaw = initialWakePayload.WakeMessage
			}
		}
		c.mu.Unlock()
		if len(initialWakeRaw) > 0 {
			var err error
			initialWake, initialWakeMessageID, initialWakeInterruptPolicy, err = c.messageFromWakePayload(msg.AgentID, initialWakeRaw, cfg, initialWakePayload.UnreadSummary)
			if err != nil {
				slog.Debug("AgentTeam: agent:start wake message parse error", "channel", c.id, "agent_id", msg.AgentID, "error", err)
			}
		}

		// Immediately spawn the agent process so it is running when the server
		// expects it. When available, the first wake message is the initial
		// runtime input so cold start does not re-enter active-turn delivery.
		startErr := c.disp.StartAgentForAgentTeam(c.id, msg.AgentID, cfg, initialWake)

		c.mu.Lock()
		restartQueued := false
		if s, ok := c.agents[msg.AgentID]; ok {
			restartQueued = s.RestartQueued
			s.RestartQueued = false
			if !restartQueued {
				s.Starting = false
				pendingStarts = append([]agentStartPayload(nil), s.PendingStarts...)
				s.PendingStarts = nil
			}
		}
		c.mu.Unlock()

		if restartQueued {
			if startErr != nil {
				slog.Warn("AgentTeam: queued agent:start will replace failed in-flight start",
					"channel", c.id, "agent_id", msg.AgentID, "error", startErr)
			} else {
				slog.Info("AgentTeam: queued agent:start replacing in-flight start",
					"channel", c.id, "agent_id", msg.AgentID)
			}
			replaceQueuedStart = true
			continue
		}

		if startErr != nil {
			slog.Warn("AgentTeam: agent:start failed to spawn process",
				"channel", c.id, "agent_id", msg.AgentID, "error", startErr)
			c.emitAgentStartFailure(msg.AgentID, pendingStarts, startErr)
			return
		}
		if initialWake != nil {
			pendingStarts = removeInitialWakePayload(pendingStarts, initialWakeRaw)
			h := deliveryDedupHashWithInterruptPolicy(msg.AgentID, 0, initialWakeMessageID, initialWakeInterruptPolicy)
			c.commitDeliveryDedup(h, msg.AgentID)
			c.sendAck(msg.AgentID, initialWakeMessageID, 0, initialWakeInterruptPolicy, true, "")
		}
		break
	}

	// If there's a wake message, deliver it as a normal user types.
	// The wakeMessage is the inner deliverMessage payload, not the full
	// agentDeliverMsg envelope, so we must wrap it.
	for index, startPayload := range pendingStarts {
		envelope, err := json.Marshal(agentDeliverMsg{
			Type:    "agent:deliver",
			AgentID: msg.AgentID,
			Seq:     0,
			Message: startPayload.WakeMessage,
		})
		if err != nil {
			c.emitAgentStartFailure(msg.AgentID, pendingStarts[index:], err)
			return
		}
		switch c.handleAgentStartDelivery(ctx, envelope, startPayload.UnreadSummary) {
		case deliveryHandoffAccepted:
			continue
		case deliveryHandoffRejected:
			// A negative ACK terminally resolves this payload. Keep draining later
			// valid wakes rather than declaring the already-running Process offline.
			continue
		case deliveryHandoffNeedsLifecycleStart:
			if err := c.startAgentForPendingWake(msg.AgentID, startPayload); err != nil {
				slog.Warn("AgentTeam: queued agent:start wake replacement failed",
					"channel", c.id, "agent_id", msg.AgentID, "error", err)
				c.emitAgentStartFailure(msg.AgentID, pendingStarts[index:], err)
				return
			}
		}
	}

	// The Process is active only after every detached start wake has either been
	// handed to the live generation or transferred through lifecycle replacement.
	c.Emit(types.StatusEvent{
		AgentID: msg.AgentID,
		Status:  "active",
	})
}

func (c *Protocol) startAgentForPendingWake(agentID string, startPayload agentStartPayload) error {
	c.mu.Lock()
	cfg := types.AgentStartConfig{}
	if st, ok := c.agents[agentID]; ok {
		cfg = st.Config
	}
	c.mu.Unlock()

	wake, messageID, interruptPolicy, err := c.messageFromWakePayload(
		agentID,
		startPayload.WakeMessage,
		cfg,
		startPayload.UnreadSummary,
	)
	if err != nil {
		return err
	}
	if wake == nil {
		return errors.New("agent:start wake payload has no deliverable input")
	}
	if err := c.disp.StartAgentForAgentTeam(c.id, agentID, cfg, wake); err != nil {
		return err
	}
	h := deliveryDedupHashWithInterruptPolicy(agentID, 0, messageID, interruptPolicy)
	c.commitDeliveryDedup(h, agentID)
	c.sendAck(agentID, messageID, 0, interruptPolicy, true, "")
	return nil
}

func (c *Protocol) emitAgentStartFailure(agentID string, starts []agentStartPayload, err error) {
	c.emitAgentStartWakeFailures(agentID, starts, err)
	detail := "Start failed: " + strings.TrimSpace(err.Error())
	c.Emit(types.StatusEvent{
		AgentID: agentID,
		Status:  "inactive",
	})
	c.Emit(types.ActivityEvent{
		AgentID:  agentID,
		Activity: "offline",
		Detail:   detail,
	})
}

func parseUnreadSummary(raw json.RawMessage) map[string]int {
	if len(raw) == 0 || string(raw) == "null" {
		return nil
	}

	var decoded map[string]float64
	if err := json.Unmarshal(raw, &decoded); err != nil || len(decoded) == 0 {
		return nil
	}

	parsed := make(map[string]int, len(decoded))
	for ch, count := range decoded {
		if int(count) > 0 {
			parsed[ch] = int(count)
		}
	}
	if len(parsed) == 0 {
		return nil
	}
	return parsed
}

// handleAgentStop stops the agent process and cleans up its stored state.
func (c *Protocol) handleAgentStop(_ context.Context, raw []byte) {
	prepared, ok := c.prepareAgentStop(raw)
	if !ok {
		return
	}
	c.handlePreparedAgentStop(prepared)
}

type preparedAgentStop struct {
	message         agentStopMsg
	receiptSequence uint64
}

func (c *Protocol) prepareAgentStop(raw []byte) (preparedAgentStop, bool) {
	var msg agentStopMsg
	if err := json.Unmarshal(raw, &msg); err != nil {
		slog.Debug("AgentTeam: agent:stop parse error", "channel", c.id, "error", err)
		return preparedAgentStop{}, false
	}
	c.mu.Lock()
	if c.agents == nil {
		c.agents = make(map[string]*agentState)
	}
	receiptSequence := c.getOrCreateAgentState(msg.AgentID).nextReceiptSequence()
	c.mu.Unlock()
	return preparedAgentStop{message: msg, receiptSequence: receiptSequence}, true
}

func (c *Protocol) handlePreparedAgentStop(prepared preparedAgentStop) {
	msg := prepared.message
	c.metrics.agentStops.Add(1)

	slog.Info("AgentTeam: agent:stop received",
		"channel", c.id, "agent_id", msg.AgentID)

	c.disp.StopByServerAgentID(msg.AgentID)
	if msg.ClearSession {
		if clearer, ok := c.disp.(sessionResumeStateClearer); ok {
			if err := clearer.ClearSessionResumeStateByAgent(msg.AgentID); err != nil {
				slog.Warn("AgentTeam: clear session resume state failed", "channel", c.id, "agent_id", msg.AgentID, "error", err)
			}
		}
	}

	c.mu.Lock()
	state, ok := c.agents[msg.AgentID]
	if !ok || state == nil {
		delete(c.agents, msg.AgentID)
		c.mu.Unlock()
		return
	}
	next := &agentState{
		lifecycleTail:   state.lifecycleTail,
		receiptSequence: state.receiptSequence,
	}
	hasLaterAuthority := false
	if state.startConfigReceiptSequence > prepared.receiptSequence {
		next.Config = state.Config
		next.Sleeping = state.Sleeping
		next.startConfigReceiptSequence = state.startConfigReceiptSequence
		hasLaterAuthority = true
	}
	if state.identityReceiptSequence > prepared.receiptSequence {
		next.Config.Name = state.Config.Name
		next.Config.DisplayName = state.Config.DisplayName
		next.identityReceiptSequence = state.identityReceiptSequence
		hasLaterAuthority = true
	}
	if state.tokenReceiptSequence > prepared.receiptSequence {
		next.Token = state.Token
		next.tokenReceiptSequence = state.tokenReceiptSequence
		hasLaterAuthority = true
	}
	if next.lifecycleTail != nil || hasLaterAuthority {
		// Keep the receipt clock and predecessor chain. Only authority received
		// after this stop survives, so a later configured start cannot be erased
		// by this owner's delayed cleanup.
		c.agents[msg.AgentID] = next
	} else {
		delete(c.agents, msg.AgentID)
	}
	c.mu.Unlock()

	// Status/activity is reported by reportExitStatus when the process exits.
	// Do not report here to avoid duplicate or conflicting reports.
}

// handleAgentInterrupt interrupts the current runtime turn for an agent without
// clearing its daemon-local session metadata.
func (c *Protocol) handleAgentInterrupt(_ context.Context, raw []byte) {
	var msg agentInterruptMsg
	if err := json.Unmarshal(raw, &msg); err != nil {
		slog.Debug("AgentTeam: agent:interrupt parse error", "channel", c.id, "error", err)
		return
	}

	slog.Info("AgentTeam: agent:interrupt received",
		"channel", c.id, "agent_id", msg.AgentID)

	c.disp.InterruptActiveRunByServerAgentID(msg.AgentID)
}

// handleAgentSleep puts the agent process to sleep but retains its stored
// state so that a subsequent server agent:start can resume it.
func (c *Protocol) handleAgentSleep(_ context.Context, raw []byte) {
	c.metrics.agentSleeps.Add(1)
	var msg agentSleepMsg
	if err := json.Unmarshal(raw, &msg); err != nil {
		slog.Debug("AgentTeam: agent:sleep parse error", "channel", c.id, "error", err)
		return
	}

	slog.Info("AgentTeam: agent:sleep received",
		"channel", c.id, "agent_id", msg.AgentID)

	c.disp.SleepByServerAgentID(msg.AgentID)

	c.mu.Lock()
	if st, ok := c.agents[msg.AgentID]; ok {
		st.Sleeping = true
		st.Starting = false
	}
	c.mu.Unlock()

	// Status/activity is reported by reportExitStatus when the process exits.
	// Do not report here to avoid duplicate reports.
}

// handleUpdateIdentity processes an agent:update_identity command: forwards
// the name change to the dispatcher so it can update persisted state and
// save an identity notice for the running agent.
func (c *Protocol) handleUpdateIdentity(raw []byte) {
	msg, ok := c.prepareAgentIdentityUpdate(raw)
	if !ok {
		return
	}
	c.applyPreparedAgentIdentityUpdate(msg)
}

func (c *Protocol) prepareAgentIdentityUpdate(raw []byte) (agentUpdateIdentityMsg, bool) {
	var msg agentUpdateIdentityMsg
	if err := json.Unmarshal(raw, &msg); err != nil {
		slog.Debug("AgentTeam: agent:update_identity parse error", "channel", c.id, "error", err)
		return agentUpdateIdentityMsg{}, false
	}
	c.rememberAgentIdentity(msg.AgentID, msg.Name, msg.DisplayName)
	return msg, true
}

func (c *Protocol) rememberAgentIdentity(agentID, name, displayName string) {
	agentID = strings.TrimSpace(agentID)
	name = strings.TrimSpace(name)
	displayName = strings.TrimSpace(displayName)
	if displayName == "" {
		displayName = name
	}
	if name == "" {
		name = displayName
	}
	if agentID == "" || name == "" {
		return
	}
	c.mu.Lock()
	if c.agents == nil {
		c.agents = make(map[string]*agentState)
	}
	state := c.getOrCreateAgentState(agentID)
	receiptSequence := state.nextReceiptSequence()
	state.Config.Name = name
	state.Config.DisplayName = displayName
	state.identityReceiptSequence = receiptSequence
	c.mu.Unlock()
}

func (c *Protocol) applyPreparedAgentIdentityUpdate(msg agentUpdateIdentityMsg) {
	slog.Info("AgentTeam: agent:update_identity received",
		"channel", c.id, "agent_id", msg.AgentID, "name", msg.Name, "display_name", msg.DisplayName)
	c.disp.UpdateAgentIdentity(msg.AgentID, msg.Name, msg.DisplayName)
}

// handleMemoryAccessChanged re-evaluates memory watcher state for the listed
// agents after the server notifies that memory access has been granted or
// revoked.
func (c *Protocol) handleMemoryAccessChanged(ctx context.Context, msg memoryAccessChangedMsg) {
	if len(msg.AgentIDs) == 0 {
		return
	}
	slog.Info("AgentTeam: agent:memory:access-changed received",
		"channel", c.id, "agent_ids", msg.AgentIDs)
	if refresher, ok := c.disp.(memoryAccessRefresher); ok {
		refresher.RefreshMemoryAccess(ctx, msg.AgentIDs)
	}
}

// refreshMemoryAccessForSyncedAgents triggers memory pre-clone for all agents
// in the server catalog. Called after agent list sync to ensure memory repos
// are cloned and legacy MEMORY.md is migrated at startup.
func (c *Protocol) refreshMemoryAccessForSyncedAgents(agents []types.MachineAgentInfo, contexts ...context.Context) {
	ids := make([]string, 0, len(agents))
	for _, ag := range agents {
		if id := strings.TrimSpace(ag.ID); id != "" {
			ids = append(ids, id)
		}
	}
	if len(ids) == 0 {
		return
	}
	if refresher, ok := c.disp.(memoryAccessRefresher); ok {
		ctx := firstConnectionTaskContext(contexts)
		ctx, cancel := context.WithTimeout(ctx, 60*time.Second)
		defer cancel()
		slog.Info("AgentTeam: pre-cloning memory for synced agents",
			"channel", c.id, "agent_count", len(ids))
		refresher.RefreshMemoryAccess(ctx, ids)
	}
}

// refreshSkillAccessForSyncedAgents triggers skill repository sync for all
// agents in the server catalog. This stays off the machine:agent:list control
// plane path; fresh launches that still observe a cache miss perform their own
// bounded readiness refresh before starting.
func (c *Protocol) refreshSkillAccessForSyncedAgents(agents []types.MachineAgentInfo, contexts ...context.Context) {
	ids := make([]string, 0, len(agents))
	for _, ag := range agents {
		if id := strings.TrimSpace(ag.ID); id != "" {
			ids = append(ids, id)
		}
	}
	if len(ids) == 0 {
		return
	}
	if refresher, ok := c.disp.(skillAccessRefresher); ok {
		ctx := firstConnectionTaskContext(contexts)
		ctx, cancel := context.WithTimeout(ctx, 60*time.Second)
		defer cancel()
		slog.Info("AgentTeam: pre-cloning skills for synced agents",
			"channel", c.id, "agent_count", len(ids))
		refresher.RefreshSkillAccess(ctx, ids)
	}
}

func (c *Protocol) ensureManagedSkillsReadyForLaunch(ctx context.Context, agentID string) {
	if c == nil || c.disp == nil {
		return
	}
	agentID = strings.TrimSpace(agentID)
	if agentID == "" {
		return
	}
	if c.managedSkillsReadyForLaunch(agentID) {
		return
	}
	refresher, ok := c.disp.(skillAccessRefresher)
	if !ok || refresher == nil {
		return
	}
	if ctx == nil {
		ctx = context.Background()
	}
	ctx, cancel := context.WithTimeout(ctx, 60*time.Second)
	defer cancel()
	slog.Info("AgentTeam: synchronizing managed skills before fresh launch",
		"channel", c.id, "agent_id", agentID)
	refresher.RefreshSkillAccess(ctx, []string{agentID})
}

func (c *Protocol) managedSkillsReadyForLaunch(agentID string) bool {
	if c == nil || c.disp == nil {
		return true
	}
	agentID = strings.TrimSpace(agentID)
	if agentID == "" {
		return true
	}
	checker, ok := c.disp.(managedSkillLaunchReadinessChecker)
	if !ok || checker == nil {
		return true
	}
	return checker.ManagedSkillLaunchReady(agentID)
}

// handleAssetsAccessChanged is the unified asset (memory + skill) access
// change handler. The server emits this with asset_types listing which kinds
// changed; the daemon re-evaluates the relevant refreshers per kind.
//
// asset_types is filtered against the kinds we know how to refresh:
//   - "memory" → RefreshMemoryAccess
//   - "skill"  → RefreshSkillAccess
//   - "kb"     → RefreshKnowledgeBaseAccess
//
// An empty asset_types slice is treated as "both" for backward compatibility
// — older servers that did not populate the field still drive both refresh
// paths.
//
// Idempotency: the server includes event_id on every push and the durable
// outbox replays until acknowledged. Dedup is keyed by event_id (NOT by a
// channel-level version counter) because a single server push can cover
// multiple agents and asset kinds, and per-(agent, asset_type) versions are
// not monotonic across distinct server emissions. We keep a bounded ring of
// recently-seen event_ids so duplicates from reconnects or replays are
// skipped without being confused with unrelated events for other agents.
// Events with no event_id are always applied — the server only omits the
// field in legacy compatibility paths where dedup is impossible anyway.
func (c *Protocol) handleAssetsAccessChanged(ctx context.Context, msg assetsAccessChangedMsg) {
	if len(msg.AgentIDs) == 0 {
		return
	}
	if !c.shouldApplyAssetsEvent(msg.EventID) {
		slog.Debug("AgentTeam: agent:assets:access-changed skipping duplicate event",
			"channel", c.id, "event_id", msg.EventID,
			"agent_ids", msg.AgentIDs, "asset_types", msg.AssetTypes)
		return
	}

	kinds := normalizeAssetTypes(msg.AssetTypes)
	slog.Info("AgentTeam: agent:assets:access-changed received",
		"channel", c.id, "agent_ids", msg.AgentIDs,
		"asset_types", kinds, "version", msg.Version, "event_id", msg.EventID)

	dispatchAssetsAccessByKind(ctx, c.disp, kinds, msg.AgentIDs)
}

// dispatchAssetsAccessByKind fans an assets:access-changed event out to the
// memory and skill refreshers based on the kinds set. Extracted from the
// Protocol method so tests can drive it with a fake dispatcher that only
// implements the refresher interfaces.
func dispatchAssetsAccessByKind(ctx context.Context, disp any, kinds map[string]bool, agentIDs []string) {
	if kinds["memory"] {
		if refresher, ok := disp.(memoryAccessRefresher); ok {
			refresher.RefreshMemoryAccess(ctx, agentIDs)
		}
	}
	if kinds["skill"] {
		if refresher, ok := disp.(skillAccessRefresher); ok {
			refresher.RefreshSkillAccess(ctx, agentIDs)
		}
	}
	if kinds["kb"] {
		if refresher, ok := disp.(knowledgeBaseAccessRefresher); ok {
			refresher.RefreshKnowledgeBaseAccess(ctx, agentIDs)
		}
	}
}

// shouldApplyAssetsEvent returns false when eventID has already been applied
// by this channel, recording it as seen if not. An empty eventID means the
// server did not tag the event — always apply since we cannot dedup.
//
// The seen-set is a bounded ring (assetsEventDedupCapacity entries, FIFO
// eviction) so a long-running daemon does not accumulate IDs unboundedly.
// The bound is comfortably larger than the server's typical replay window
// so genuine replays after a daemon restart (which loses the in-memory set
// anyway) are still applied at least once.
func (c *Protocol) shouldApplyAssetsEvent(eventID string) bool {
	eventID = strings.TrimSpace(eventID)
	if eventID == "" {
		return true
	}
	c.assetsEventMu.Lock()
	defer c.assetsEventMu.Unlock()
	if c.seenAssetsEventIDs == nil {
		c.seenAssetsEventIDs = make(map[string]struct{}, assetsEventDedupCapacity)
		c.seenAssetsEventOrder = make([]string, 0, assetsEventDedupCapacity)
	}
	if _, seen := c.seenAssetsEventIDs[eventID]; seen {
		return false
	}
	if len(c.seenAssetsEventOrder) >= assetsEventDedupCapacity {
		// Evict oldest in FIFO order to keep the set bounded.
		evict := c.seenAssetsEventOrder[0]
		c.seenAssetsEventOrder = c.seenAssetsEventOrder[1:]
		delete(c.seenAssetsEventIDs, evict)
	}
	c.seenAssetsEventIDs[eventID] = struct{}{}
	c.seenAssetsEventOrder = append(c.seenAssetsEventOrder, eventID)
	return true
}

// normalizeAssetTypes turns the server's asset_types slice into a set lookup.
// An empty slice is treated as "all known kinds" so legacy pushes still drive
// both refreshers.
func normalizeAssetTypes(kinds []string) map[string]bool {
	if len(kinds) == 0 {
		return map[string]bool{"memory": true, "skill": true}
	}
	out := make(map[string]bool, len(kinds))
	for _, k := range kinds {
		switch strings.ToLower(strings.TrimSpace(k)) {
		case "memory", "skill", "kb":
			out[strings.ToLower(strings.TrimSpace(k))] = true
		}
	}
	return out
}

// handleResetWorkspace resets the agent's workspace directory and reports status.
func (c *Protocol) handleResetWorkspace(_ context.Context, raw []byte) {
	var msg agentResetWorkspaceMsg
	if err := json.Unmarshal(raw, &msg); err != nil {
		slog.Debug("AgentTeam: agent:reset-workspace parse error", "channel", c.id, "error", err)
		return
	}

	slog.Info("AgentTeam: agent:reset-workspace received",
		"channel", c.id, "agent_id", msg.AgentID)

	reset := false
	var err error
	if disp, ok := c.disp.(resettableDispatcher); ok {
		reset, err = disp.ResetByAgent(msg.AgentID)
	}
	if err == nil && !reset {
		_, err = workspacefs.DeleteDir(msg.AgentID)
	}
	if err != nil {
		slog.Warn("AgentTeam: reset workspace failed", "channel", c.id, "agent_id", msg.AgentID, "error", err)
	}

	c.Emit(types.StatusEvent{
		AgentID: msg.AgentID,
		Status:  "inactive",
	})
}

// extensionCatalogKey computes a deterministic key from the agents' Extension
// Catalog revisions. It fires a conditional catalog refresh only when the
// server reports a newer per-agent revision, not on every machine sync.
func extensionCatalogKey(agents []types.MachineAgentInfo) string {
	var sb strings.Builder
	for _, agent := range agents {
		if agent.ExtensionCatalog != nil {
			fmt.Fprintf(&sb, "%s:%d|", agent.ID, agent.ExtensionCatalog.Version)
		}
	}
	return sb.String()
}

// handleAgentListResponse processes the server's agent list and syncs locally.
func (c *Protocol) handleAgentListResponse(raw []byte, contexts ...context.Context) {
	var msg agentListResponseMsg
	if err := json.Unmarshal(raw, &msg); err != nil {
		slog.Debug("AgentTeam: machine:agent:list:response parse error", "channel", c.id, "error", err)
		return
	}
	ctx := firstConnectionTaskContext(contexts)
	c.syncAgents(msg.Agents, ctx)
	if reconciler, ok := c.disp.(agentExtensionCatalogReconciler); ok {
		reconciler.ObserveAgentExtensionCatalogHints(c.id, msg.Agents)
	}
	// Fire the catalog-change notification only when the Extension Catalog
	// actually changed (revision comparison), not on every sync (stage 6).
	catalogKey := extensionCatalogKey(msg.Agents)
	if catalogKey != c.lastExtensionCatalogKey {
		c.lastExtensionCatalogKey = catalogKey
		if notifier, ok := c.disp.(agentCatalogChangeNotifier); ok {
			notifier.PublishAgentCatalogChanged(c.id)
		}
	}

	c.completeAgentListSync(true)

	if reconciler, ok := c.disp.(agentExtensionCatalogReconciler); ok {
		for _, agent := range msg.Agents {
			if agent.ExtensionCatalog == nil {
				continue
			}
			agentID := strings.TrimSpace(agent.ID)
			if agentID == "" {
				continue
			}
			api := c.APIClient(agentID)
			api.LocalAgentID = agentID
			c.startConnectionRunTask(ctx, func(ctx context.Context) {
				if err := reconciler.RefreshAgentExtensionCatalog(ctx, c.id, agentID, api); err != nil {
					slog.Warn(
						"AgentTeam: refresh Agent Extension Catalog failed",
						"channel", c.id,
						"agent_id", agentID,
						"error", err,
					)
				}
			})
		}
	}

	// Keep asset prefetch off the machine:agent:list control-plane path. Agent
	// list sync waiters gate agent:start / agent:deliver recovery and must only
	// wait for local catalog sync, not best-effort MeshX asset refresh.
	c.startConnectionRunTask(ctx, func(ctx context.Context) { c.refreshSkillAccessForSyncedAgents(msg.Agents, ctx) })
	c.startConnectionRunTask(ctx, func(ctx context.Context) { c.refreshKnowledgeBaseAccessForSyncedAgents(msg.Agents, ctx) })

	// After catalog sync, proactively pre-clone memory repos for all agents
	// owned by this channel. This ensures memory/ is ready before any session
	// starts and migrates legacy MEMORY.md files.
	c.startConnectionRunTask(ctx, func(ctx context.Context) { c.refreshMemoryAccessForSyncedAgents(msg.Agents, ctx) })
}

func firstConnectionTaskContext(contexts []context.Context) context.Context {
	if len(contexts) > 0 && contexts[0] != nil {
		return contexts[0]
	}
	return context.Background()
}

// handleWorkspaceListResponse resolves a pending AgentTeam workspace list request.
func (c *Protocol) handleWorkspaceListResponse(raw []byte) {
	var msg workspaceListResponseMsg
	if err := json.Unmarshal(raw, &msg); err != nil {
		slog.Debug("AgentTeam: machine:workspaces:list:response parse error", "channel", c.id, "error", err)
		c.failAllWorkspaceListRequests(err)
		return
	}
	requestID := strings.TrimSpace(msg.RequestID)
	if requestID == "" {
		err := errors.New("workspace list response requestId is required")
		slog.Debug("AgentTeam: machine:workspaces:list:response protocol error", "channel", c.id, "error", err)
		c.failAllWorkspaceListRequests(err)
		return
	}

	result := workspaceListResult{}
	if trimmedError := strings.TrimSpace(msg.Error); trimmedError != "" {
		result.err = workspaceListProtocolError(trimmedError)
	} else if !msg.Success {
		result.err = errors.New("workspace list request failed")
	} else {
		result.workspaces = normalizeWorkspaceList(msg.Workspaces)
	}
	c.deliverWorkspaceListResult(requestID, result)
}

func workspaceListProtocolError(message string) error {
	trimmed := strings.TrimSpace(message)
	if trimmed == "" {
		return errors.New("workspace list request failed")
	}
	lower := strings.ToLower(trimmed)
	switch {
	case strings.Contains(lower, "workspace list is unavailable"),
		strings.Contains(lower, "workspace list unsupported"),
		strings.Contains(lower, "does not support"),
		strings.Contains(lower, "unsupported"):
		return fmt.Errorf("%w: %s", types.ErrAgentTeamWorkspaceListUnsupported, trimmed)
	case strings.Contains(lower, "unauthorized"),
		strings.Contains(lower, "forbidden"),
		strings.Contains(lower, "permission denied"):
		return fmt.Errorf("%w: %s", types.ErrAgentTeamWorkspaceListUnauthorized, trimmed)
	default:
		return errors.New(trimmed)
	}
}

func normalizeWorkspaceList(items []types.AgentTeamWorkspaceSummary) []types.AgentTeamWorkspaceSummary {
	if items == nil {
		return []types.AgentTeamWorkspaceSummary{}
	}
	result := make([]types.AgentTeamWorkspaceSummary, 0, len(items))
	for _, item := range items {
		id := strings.TrimSpace(item.ID)
		if id == "" {
			continue
		}
		name := strings.TrimSpace(item.Name)
		if name == "" {
			name = strings.TrimSpace(item.Slug)
		}
		if name == "" {
			name = id
		}
		result = append(result, types.AgentTeamWorkspaceSummary{
			ID:   id,
			Name: name,
			Slug: strings.TrimSpace(item.Slug),
		})
	}
	return result
}

func (c *Protocol) deliverWorkspaceListResult(requestID string, result workspaceListResult) {
	c.mu.Lock()
	defer c.mu.Unlock()
	if ch := c.workspaceListRequests[requestID]; ch != nil {
		ch <- result
	}
}

func (c *Protocol) failAllWorkspaceListRequests(err error) {
	c.mu.Lock()
	defer c.mu.Unlock()
	for _, ch := range c.workspaceListRequests {
		if ch == nil {
			continue
		}
		select {
		case ch <- workspaceListResult{err: err}:
		default:
		}
	}
}

// handleWorkspaceList lists files in the agent workspace or explicit target directory.
func (c *Protocol) handleWorkspaceList(_ context.Context, raw []byte) {
	var msg workspaceListMsg
	if err := json.Unmarshal(raw, &msg); err != nil {
		slog.Debug("AgentTeam: workspace:list parse error", "channel", c.id, "error", err)
		return
	}

	if strings.TrimSpace(msg.AgentID) == "" && strings.TrimSpace(msg.TargetPath) == "" {
		slog.Warn("AgentTeam: workspace list missing agent_id and target_path", "channel", c.id)
		return
	}

	slog.Info("AgentTeam: agent:workspace:list received",
		"channel", c.id, "agent_id", msg.AgentID, "target_path", msg.TargetPath, "dir_path", msg.DirPath)

	files, page, err := workspacefs.ListTargetDirPage(msg.AgentID, msg.TargetPath, msg.DirPath, workspacefs.ListOptions{
		Limit:  msg.Limit,
		Cursor: msg.Cursor,
	})
	if err != nil {
		slog.Warn("AgentTeam: workspace list error", "channel", c.id, "agent_id", msg.AgentID, "error", err)
		files = []workspacefs.File{}
		page = workspacefs.PageInfo{Limit: workspacefs.DefaultListLimit}
	}

	c.sendJSON(map[string]any{
		"type":       "agent:workspace:file_tree",
		"agentId":    msg.AgentID,
		"dirPath":    msg.DirPath,
		"rootPath":   page.RootPath,
		"files":      files,
		"limit":      page.Limit,
		"cursor":     page.Cursor,
		"hasMore":    page.HasMore,
		"nextCursor": page.NextCursor,
	})
}

// handleWorkspaceRead reads a file from the agent workspace or explicit target directory.
func (c *Protocol) handleWorkspaceRead(_ context.Context, raw []byte) {
	var msg workspaceReadMsg
	if err := json.Unmarshal(raw, &msg); err != nil {
		slog.Debug("AgentTeam: workspace:read parse error", "channel", c.id, "error", err)
		return
	}

	if (strings.TrimSpace(msg.AgentID) == "" && strings.TrimSpace(msg.TargetPath) == "") || strings.TrimSpace(msg.Path) == "" {
		slog.Warn("AgentTeam: workspace read missing required fields", "channel", c.id)
		return
	}

	slog.Info("AgentTeam: agent:workspace:read received",
		"channel", c.id, "agent_id", msg.AgentID, "target_path", msg.TargetPath, "path", msg.Path)

	content, binary, err := workspacefs.ReadTargetFile(msg.AgentID, msg.TargetPath, msg.Path)
	if err != nil {
		slog.Warn("AgentTeam: workspace read error", "channel", c.id, "agent_id", msg.AgentID, "error", err)
		c.sendJSON(map[string]any{
			"type":      "agent:workspace:file_content",
			"agentId":   msg.AgentID,
			"requestId": msg.RequestID,
			"content":   nil,
			"binary":    false,
		})
		return
	}

	var contentPtr *string
	if !binary {
		contentPtr = &content
	}

	c.sendJSON(map[string]any{
		"type":      "agent:workspace:file_content",
		"agentId":   msg.AgentID,
		"requestId": msg.RequestID,
		"content":   contentPtr,
		"binary":    binary,
	})
}

// handleWorkspaceApplySeedFiles overwrites server-provided seed files inside
// the target agent workspace and reports the result with the same request id.
func (c *Protocol) handleWorkspaceApplySeedFiles(_ context.Context, raw []byte) {
	var msg workspaceApplySeedFilesMsg
	if err := json.Unmarshal(raw, &msg); err != nil {
		slog.Debug("AgentTeam: workspace:apply_seed_files parse error", "channel", c.id, "error", err)
		return
	}

	slog.Info("AgentTeam: agent:workspace:apply_seed_files received",
		"channel", c.id, "agent_id", msg.AgentID, "request_id", msg.RequestID, "seed_file_count", len(msg.SeedFiles))

	err := workspacefs.ApplySeedFiles(msg.AgentID, msg.SeedFiles)
	success := err == nil
	if err != nil {
		slog.Warn("AgentTeam: workspace apply seed files error", "channel", c.id, "agent_id", msg.AgentID, "error", err)
	}

	response := map[string]any{
		"type":      "agent:workspace:apply_seed_files_result",
		"agentId":   msg.AgentID,
		"requestId": msg.RequestID,
		"success":   success,
	}
	if err != nil {
		response["error"] = err.Error()
	}
	c.sendJSON(response)
}

// handleWorkspaceDelete removes a file or directory under the agent workspace.
func (c *Protocol) handleWorkspaceDelete(_ context.Context, raw []byte) {
	var msg workspaceDeleteMsg
	if err := json.Unmarshal(raw, &msg); err != nil {
		slog.Debug("AgentTeam: workspace:delete parse error", "channel", c.id, "error", err)
		return
	}

	slog.Info("AgentTeam: agent:workspace:delete received",
		"channel", c.id, "agent_id", msg.AgentID, "path", msg.Path, "request_id", msg.RequestID)

	err := workspacefs.DeletePath(msg.AgentID, msg.Path)
	success := err == nil
	if err != nil {
		slog.Warn("AgentTeam: workspace delete error", "channel", c.id, "agent_id", msg.AgentID, "path", msg.Path, "error", err)
	}

	response := map[string]any{
		"type":      "agent:workspace:delete_result",
		"agentId":   msg.AgentID,
		"requestId": msg.RequestID,
		"success":   success,
	}
	if err != nil {
		response["error"] = err.Error()
	}
	c.sendJSON(response)
}

// handleMachineScan scans one bounded page of agent workspace directories.
func (c *Protocol) handleMachineScan(_ context.Context, raw []byte) {
	var msg machineWorkspaceScanMsg
	if err := json.Unmarshal(raw, &msg); err != nil {
		slog.Debug("AgentTeam: machine:workspace:scan parse error", "channel", c.id, "error", err)
		return
	}

	slog.Info("AgentTeam: machine:workspace:scan received", "channel", c.id)

	dirs, page, err := workspacefs.ScanAllPage(workspacefs.ListOptions{
		Limit:  msg.Limit,
		Cursor: msg.Cursor,
	})
	if err != nil {
		slog.Warn("AgentTeam: machine scan error", "channel", c.id, "error", err)
		dirs = []workspacefs.DirInfo{}
		page = workspacefs.PageInfo{Limit: workspacefs.DefaultListLimit}
	}

	c.sendJSON(map[string]any{
		"type":        "machine:workspace:scan_result",
		"directories": dirs,
		"limit":       page.Limit,
		"cursor":      page.Cursor,
		"hasMore":     page.HasMore,
		"nextCursor":  page.NextCursor,
	})
}

// handleMachineDelete deletes an agent workspace directory.
func (c *Protocol) handleMachineDelete(_ context.Context, raw []byte) {
	var msg machineDeleteMsg
	if err := json.Unmarshal(raw, &msg); err != nil {
		slog.Debug("AgentTeam: machine:workspace:delete parse error", "channel", c.id, "error", err)
		return
	}

	slog.Info("AgentTeam: machine:workspace:delete received",
		"channel", c.id, "dir", msg.DirectoryName)

	success, err := workspacefs.DeleteDir(msg.DirectoryName)
	if err != nil {
		slog.Warn("AgentTeam: machine delete error", "channel", c.id, "error", err)
	}

	c.sendJSON(map[string]any{
		"type":          "machine:workspace:delete_result",
		"directoryName": msg.DirectoryName,
		"success":       success,
	})
}

func (c *Protocol) handleTrajectoryHistory(_ context.Context, raw []byte) {
	var msg agentTrajectoryHistoryMsg
	if err := json.Unmarshal(raw, &msg); err != nil {
		slog.Debug("AgentTeam: agent:trajectory:history parse error", "channel", c.id, "error", err)
		return
	}

	slog.Info("AgentTeam: agent:trajectory:history received",
		"channel", c.id, "agent_id", msg.AgentID, "request_id", msg.RequestID, "limit", msg.Limit, "before", msg.Before)

	result := agentTrajectoryHistoryResultMsg{
		Type:      "agent:trajectory:history_result",
		AgentID:   msg.AgentID,
		RequestID: msg.RequestID,
		Items:     []types.TrajectoryLogItem{},
	}
	if c.disp == nil {
		result.Error = "dispatcher unavailable"
	} else {
		result.Items = c.disp.TrajectoryHistory(msg.AgentID, msg.Limit, msg.Before)
	}
	c.sendJSON(result)
}

// handleAgentRestoreRecoveryPoint processes an agent:restore_recovery_point
// command by delegating to the dispatcher's recovery handler.
func (c *Protocol) handleAgentRestoreRecoveryPoint(_ context.Context, raw []byte) {
	var msg agentRestoreRecoveryPointMsg
	if err := json.Unmarshal(raw, &msg); err != nil {
		slog.Debug("AgentTeam: agent:restore_recovery_point parse error", "channel", c.id, "error", err)
		return
	}

	slog.Info("AgentTeam: agent:restore_recovery_point received",
		"channel", c.id, "agent_id", msg.AgentID, "request_id", msg.RequestID,
		"runtime", msg.Runtime, "strategy", msg.Strategy)

	result := agentRestoreRecoveryPointResultMsg{
		Type:      "agent:restore_recovery_point:result",
		AgentID:   msg.AgentID,
		RequestID: msg.RequestID,
		Success:   false,
	}

	if err := c.disp.RestoreAgentRecoveryPoint(c.id, msg.AgentID, msg.RequestID, msg.Runtime, msg.Model, msg.SessionID, msg.Strategy, msg.RecoveryRef); err != nil {
		result.Error = err.Error()
		slog.Warn("AgentTeam: agent:restore_recovery_point failed",
			"channel", c.id, "agent_id", msg.AgentID, "error", err)
	} else {
		result.Success = true
		slog.Info("AgentTeam: agent:restore_recovery_point succeeded",
			"channel", c.id, "agent_id", msg.AgentID)
	}

	c.sendJSON(result)
}

// handleAgentGCRecoveryPoints processes an agent:gc_recovery_points command
// by delegating to the dispatcher's recovery cleanup handler.
func (c *Protocol) handleAgentGCRecoveryPoints(_ context.Context, raw []byte) {
	var msg agentGCRecoveryPointsMsg
	if err := json.Unmarshal(raw, &msg); err != nil {
		slog.Debug("AgentTeam: agent:gc_recovery_points parse error", "channel", c.id, "error", err)
		return
	}

	slog.Info("AgentTeam: agent:gc_recovery_points received",
		"channel", c.id, "agent_id", msg.AgentID, "count", len(msg.Requests))

	requests := make([]types.RecoveryArtifactRequest, 0, len(msg.Requests))
	for _, r := range msg.Requests {
		requests = append(requests, types.RecoveryArtifactRequest{
			Strategy:    r.Strategy,
			RecoveryRef: r.RecoveryRef,
		})
	}

	if err := c.disp.GCRecoveryPoints(msg.AgentID, requests); err != nil {
		slog.Warn("AgentTeam: agent:gc_recovery_points failed",
			"channel", c.id, "agent_id", msg.AgentID, "error", err)
	}
}

// ---------------------------------------------------------------------------
// Review handlers
// ---------------------------------------------------------------------------

func (c *Protocol) handleReviewSummary(_ context.Context, msg agentReviewSummaryMsg) {
	slog.Info("AgentTeam: agent:review:summary received",
		"channel", c.id, "agent_id", msg.AgentID, "request_id", msg.RequestID)

	// TargetPath is the canonical repo root. Do NOT TrimSpace: a root that
	// differs only by trailing whitespace ("/repo " vs "/repo") is a distinct
	// filesystem location, and trimming would silently point the review at the
	// wrong worktree. Fall back to the agent workdir only on a literal empty
	// TargetPath.
	basePath := msg.TargetPath
	if basePath == "" {
		basePath = workspacefs.WorkDir(msg.AgentID)
	}

	summary, err := workspace.ReviewSummary(basePath, msg.BaseRef, msg.HeadRef, msg.ComparisonMode, msg.CompareRef)
	result := agentReviewSummaryResultMsg{
		Type:      "agent:review:summary_result",
		AgentID:   msg.AgentID,
		RequestID: msg.RequestID,
	}
	if err != nil {
		// Populate the stable ErrorCode for known product-limitation
		// conditions so the Hub can reconstruct a typed sentinel and the
		// HTTP layer can return a controlled response (422 unsupported).
		// The Error field carries fixed, path-free user-facing text; the
		// original workspace error already had any offending path stripped
		// by wrapReviewTreeError before reaching here, but err.Error() of
		// other errors could include paths, so for the control-char case
		// we use the workspace sentinel's own fixed text rather than
		// err.Error().
		if errors.Is(err, workspace.ErrReviewControlCharPath) {
			result.ErrorCode = reviewTargetUnsupportedLegacyCode
			result.Error = err.Error()
		} else {
			result.Error = err.Error()
		}
		slog.Warn("AgentTeam: agent:review:summary failed",
			"channel", c.id, "agent_id", msg.AgentID, "error_code", result.ErrorCode)
	} else {
		result.Summary = &summary
	}
	c.sendJSON(result)
}

func (c *Protocol) handleReviewFiles(_ context.Context, msg agentReviewFilesMsg) {
	slog.Info("AgentTeam: agent:review:files received",
		"channel", c.id, "agent_id", msg.AgentID, "request_id", msg.RequestID, "scope", msg.Scope)

	// TargetPath is the canonical repo root. Do NOT TrimSpace: a root that
	// differs only by trailing whitespace ("/repo " vs "/repo") is a distinct
	// filesystem location, and trimming would silently point the review at the
	// wrong worktree. Fall back to the agent workdir only on a literal empty
	// TargetPath.
	basePath := msg.TargetPath
	if basePath == "" {
		basePath = workspacefs.WorkDir(msg.AgentID)
	}

	files, err := workspace.ReviewFiles(basePath, msg.Scope, msg.BaseRef, msg.HeadRef, msg.ComparisonMode, msg.CompareRef)
	result := agentReviewFilesResultMsg{
		Type:      "agent:review:files_result",
		AgentID:   msg.AgentID,
		RequestID: msg.RequestID,
	}
	if err != nil {
		if errors.Is(err, workspace.ErrReviewControlCharPath) {
			result.ErrorCode = reviewTargetUnsupportedLegacyCode
		}
		result.Error = err.Error()
		slog.Warn("AgentTeam: agent:review:files failed",
			"channel", c.id, "agent_id", msg.AgentID, "error_code", result.ErrorCode)
	} else {
		result.Files = files
	}
	c.sendJSON(result)
}

func (c *Protocol) handleReviewDiff(_ context.Context, msg agentReviewDiffMsg) {
	// Path/TargetPath are raw filesystem paths from the user worktree and MUST
	// NOT appear in operator logs — codex6/reviewer3 path-leak audit. The
	// channel/agent_id/request_id/scope tuple is sufficient for correlation;
	// anything more specific (path, scope detail) leaks user data. The same
	// discipline applies to the failure slog.Warn below (it logs error_code
	// only) and to handleReviewSummary / handleReviewFiles (already clean).
	slog.Info("AgentTeam: agent:review:diff received",
		"channel", c.id, "agent_id", msg.AgentID, "request_id", msg.RequestID,
		"scope", msg.Scope)

	// TargetPath is the canonical repo root. Do NOT TrimSpace: a root that
	// differs only by trailing whitespace ("/repo " vs "/repo") is a distinct
	// filesystem location, and trimming would silently point the review at the
	// wrong worktree. Fall back to the agent workdir only on a literal empty
	// TargetPath.
	basePath := msg.TargetPath
	if basePath == "" {
		basePath = workspacefs.WorkDir(msg.AgentID)
	}

	diff, err := workspace.ReviewDiff(basePath, msg.Scope, msg.Path, msg.BaseRef, msg.HeadRef, msg.ComparisonMode, msg.CompareRef)
	result := agentReviewDiffResultMsg{
		Type:      "agent:review:diff_result",
		AgentID:   msg.AgentID,
		RequestID: msg.RequestID,
	}
	if err != nil {
		if errors.Is(err, workspace.ErrReviewControlCharPath) {
			result.ErrorCode = reviewTargetUnsupportedLegacyCode
		}
		result.Error = err.Error()
		slog.Warn("AgentTeam: agent:review:diff failed",
			"channel", c.id, "agent_id", msg.AgentID, "error_code", result.ErrorCode)
	} else {
		result.Diff = &diff
	}
	c.sendJSON(result)
}

func (c *Protocol) handleOutputTargetValidate(_ context.Context, msg agentOutputTargetValidateMsg) {
	slog.Info("AgentTeam: agent:output-target:validate received",
		"channel", c.id, "agent_id", msg.AgentID, "request_id", msg.RequestID, "kind", msg.Kind)

	err := workspace.ValidateOutputTarget(msg.TargetPath, msg.Kind)
	result := agentOutputTargetValidateResultMsg{
		Type:      "agent:output-target:validate_result",
		AgentID:   msg.AgentID,
		RequestID: msg.RequestID,
		Valid:     err == nil,
	}
	if err != nil {
		result.Error = err.Error()
	}
	c.sendJSON(result)
}