package im

import (
	"context"
	"encoding/json"
	"log/slog"
	"strings"
	"sync"
	"time"

	"github.com/tidbcloud/AgentTeam/internal/api/dto"
	"github.com/tidbcloud/AgentTeam/internal/api/dto/converter"
	"github.com/tidbcloud/AgentTeam/internal/store/app/runtime"
	storeim "github.com/tidbcloud/AgentTeam/internal/store/im"
	"github.com/tidbcloud/AgentTeam/internal/store/taskboard"
	"github.com/tidbcloud/AgentTeam/internal/unread"
)

const (
	liveEmitterWakeStatusTTL           = 15 * time.Minute
	liveEmitterWakeStatusPruneInterval = time.Minute
	liveEmitterLookupTimeout           = 5 * time.Second
)

// appChannelLookup resolves the channel ID behind a message ID. It is satisfied
// by the app store so the live emitter can route agent:wake_status without
// keeping a socket.io room index.
type appChannelLookup interface {
	LookupMessageChannel(ctx context.Context, workspaceID string, messageID string) (string, string, error)
}

// LiveEventEmitter publishes runtime/control-plane live-only events through
// the Workspace IM streaming hub. It is the streaming-native replacement for
// realtime.Service: events are not persisted to im_stream_events and eventSeq
// stays 0; clients recover state through the bootstrap snapshot sent on every
// fresh /events/stream connection.
// liveEventHubBackend is the in-process hub contract the emitter depends on:
// publishing live-only events, per-user scope membership sync so channel
// join/leave handlers can keep existing connections in sync, and per-user
// forced disconnect so workspace admin actions can revoke access promptly.
type liveEventHubBackend interface {
	LiveOnlyEventPublisher
	UserScopeMembershipSync
	UserConnectionController
}

type LiveEventEmitter struct {
	hub    liveEventHubBackend
	app    appChannelLookup
	logger *slog.Logger

	// Task unread snapshots are re-read and published in order for each
	// (workspace, member, Task). Independent owners and Tasks must not share a
	// publish lock because a slow live subscriber for one key must not stall
	// unrelated unread updates.
	taskUnreadPublishSequencer liveTaskUnreadPublishSequencer
	taskUnreadReaderMu         sync.RWMutex
	unreadCurrentReader        unreadCurrentDeltaReader

	wakeStatusMu          sync.Mutex
	wakeStatusBySpace     map[string]map[string]map[string]liveWakeStatusSnapshot
	nextWakeStatusPruneAt time.Time
}

type unreadCurrentDeltaReader interface {
	CurrentUnreadDelta(workspaceID, userID, memberID, scopeID string, fallbackScopeType unread.ScopeType) (unread.UnreadDelta, bool)
}

// NewLiveEventEmitter wires a streaming-native runtime/control-plane emitter.
// app may be nil when the caller does not use EmitAgentWakeStatus.
func NewLiveEventEmitter(hub liveEventHubBackend, app appChannelLookup, logger *slog.Logger) *LiveEventEmitter {
	if logger == nil {
		logger = slog.Default()
	}
	return &LiveEventEmitter{
		hub:               hub,
		app:               app,
		logger:            logger,
		wakeStatusBySpace: make(map[string]map[string]map[string]liveWakeStatusSnapshot),
	}
}

// SetUnreadCurrentDeltaReader binds the same owner-local runtime that emits
// unread deltas. It must be called during startup before the emitter is exposed
// as that runtime's DeltaSink.
func (emitter *LiveEventEmitter) SetUnreadCurrentDeltaReader(reader unreadCurrentDeltaReader) {
	if emitter == nil {
		return
	}
	emitter.taskUnreadReaderMu.Lock()
	emitter.unreadCurrentReader = reader
	emitter.taskUnreadReaderMu.Unlock()
}

// Runtime event type namespace. The runtime.* prefix keeps live-only events
// distinct from durable message.* / conversation.* / read.* stream events.
const (
	liveEventTypeRoomsJoined             = "runtime.rooms_joined"
	liveEventTypeChannelUpdated          = "runtime.channel_updated"
	liveEventTypeChannelMembersUpdated   = "runtime.channel_members_updated"
	liveEventTypeWorkspaceAppsUpdated    = "runtime.workspace_apps_updated"
	liveEventTypeWorkspaceMembersUpdated = "runtime.workspace_members_updated"
	liveEventTypeUnreadReset             = "runtime.unread_reset"
	liveEventTypeUnreadUpdated           = "runtime.unread_updated"
	liveEventTypeServiceAccountDeleted   = "runtime.service_account_deleted"
	liveEventTypeAgentCreated            = "runtime.agent_created"
	liveEventTypeAgentDeleted            = "runtime.agent_deleted"
	liveEventTypeAgentStatus             = "runtime.agent_status"
	liveEventTypeAgentActivity           = "runtime.agent_activity"
	liveEventTypeAgentSession            = "runtime.agent_session"
	liveEventTypeAgentTrajectory         = "runtime.agent_trajectory"
	liveEventTypeAgentWakeStatus         = "runtime.agent_wake_status"
	liveEventTypeMachineStatus           = "runtime.machine_status"
	liveEventTypeMachineCapabilities     = "runtime.machine_capabilities"
	liveEventTypeDaemonStatus            = "runtime.daemon_status"
	liveEventTypeTaskCreated             = "runtime.task_created"
	liveEventTypeTaskUpdated             = "runtime.task_updated"
	liveEventTypeTaskDeleted             = "runtime.task_deleted"
)

// EmitRoomsJoined signals the caller's connection that room joins completed.
// Routed per (workspace, user) so the notification only reaches the targeted
// sessions even when several connections share a workspace.
func (emitter *LiveEventEmitter) EmitRoomsJoined(workspaceID string, userID string) {
	if emitter == nil || emitter.hub == nil {
		return
	}
	workspaceID = strings.TrimSpace(workspaceID)
	userID = strings.TrimSpace(userID)
	if workspaceID == "" || userID == "" {
		return
	}
	emitter.hub.PublishUserLiveOnlyEvent(workspaceID, userID, storeim.StreamEvent{
		WorkspaceID: workspaceID,
		Type:        liveEventTypeRoomsJoined,
	})
}

// EmitChannelUpdated broadcasts a workspace-wide channel refresh signal.
func (emitter *LiveEventEmitter) EmitChannelUpdated(workspaceID string) {
	if emitter == nil || emitter.hub == nil {
		return
	}
	workspaceID = strings.TrimSpace(workspaceID)
	if workspaceID == "" {
		return
	}
	emitter.hub.PublishWorkspaceLiveOnlyEvent(storeim.StreamEvent{
		WorkspaceID: workspaceID,
		Type:        liveEventTypeChannelUpdated,
	})
}

// SubscribeScopeForUser mirrors the old Socket.IO SocketsJoin-by-user-room
// call: every active connection owned by (workspaceID, userID) starts
// receiving channel-scoped live-only events for scopeID without needing to
// reconnect. Used when a user is added to a channel or a DM is opened for
// them after they have already connected.
func (emitter *LiveEventEmitter) SubscribeScopeForUser(workspaceID string, userID string, scopeID string) {
	if emitter == nil || emitter.hub == nil {
		return
	}
	emitter.hub.SubscribeScopeForUser(workspaceID, userID, scopeID)
}

// UnsubscribeScopeForUser mirrors the old Socket.IO SocketsLeave-by-user-room
// call: every active connection owned by (workspaceID, userID) stops
// receiving channel-scoped live-only events for scopeID immediately. Used
// when a user leaves a channel or is removed by an admin.
func (emitter *LiveEventEmitter) UnsubscribeScopeForUser(workspaceID string, userID string, scopeID string) {
	if emitter == nil || emitter.hub == nil {
		return
	}
	emitter.hub.UnsubscribeScopeForUser(workspaceID, userID, scopeID)
}

// DisconnectWorkspaceUser mirrors the old Socket.IO SocketsDisconnect-by-user-room
// call: every active /events/stream connection owned by (workspaceID, userID)
// is forcibly closed so the browser reconnects and the workspace auth
// middleware rejects the new handshake. Used when a workspace admin removes a
// member.
func (emitter *LiveEventEmitter) DisconnectWorkspaceUser(workspaceID string, userID string) {
	if emitter == nil || emitter.hub == nil {
		return
	}
	emitter.hub.DisconnectUser(workspaceID, userID)
}

// EmitChannelUpdatedForUser targets a single user's connections.
func (emitter *LiveEventEmitter) EmitChannelUpdatedForUser(workspaceID string, userID string) {
	if emitter == nil || emitter.hub == nil {
		return
	}
	workspaceID = strings.TrimSpace(workspaceID)
	userID = strings.TrimSpace(userID)
	if workspaceID == "" || userID == "" {
		return
	}
	emitter.hub.PublishUserLiveOnlyEvent(workspaceID, userID, storeim.StreamEvent{
		WorkspaceID: workspaceID,
		Type:        liveEventTypeChannelUpdated,
	})
}

// EmitChannelMembersUpdated broadcasts a channel membership change.
func (emitter *LiveEventEmitter) EmitChannelMembersUpdated(workspaceID string, channelID string) {
	if emitter == nil || emitter.hub == nil {
		return
	}
	workspaceID = strings.TrimSpace(workspaceID)
	channelID = strings.TrimSpace(channelID)
	if workspaceID == "" || channelID == "" {
		return
	}
	emitter.hub.PublishWorkspaceLiveOnlyEvent(storeim.StreamEvent{
		WorkspaceID: workspaceID,
		Type:        liveEventTypeChannelMembersUpdated,
		Payload:     map[string]any{"channelId": channelID},
	})
}

// EmitWorkspaceMembersUpdated broadcasts a workspace membership change.
func (emitter *LiveEventEmitter) EmitWorkspaceMembersUpdated(workspaceID string) {
	if emitter == nil || emitter.hub == nil {
		return
	}
	workspaceID = strings.TrimSpace(workspaceID)
	if workspaceID == "" {
		return
	}
	emitter.hub.PublishWorkspaceLiveOnlyEvent(storeim.StreamEvent{
		WorkspaceID: workspaceID,
		Type:        liveEventTypeWorkspaceMembersUpdated,
		Payload:     map[string]any{"workspaceId": workspaceID},
	})
}

// EmitWorkspaceMemberRoleUpdated broadcasts a workspace membership role change.
func (emitter *LiveEventEmitter) EmitWorkspaceMemberRoleUpdated(workspaceID string, userID string, role string) {
	if emitter == nil || emitter.hub == nil {
		return
	}
	workspaceID = strings.TrimSpace(workspaceID)
	userID = strings.TrimSpace(userID)
	role = strings.TrimSpace(role)
	if workspaceID == "" || userID == "" || role == "" {
		return
	}
	emitter.hub.PublishWorkspaceLiveOnlyEvent(storeim.StreamEvent{
		WorkspaceID: workspaceID,
		Type:        liveEventTypeWorkspaceMembersUpdated,
		Payload: map[string]any{
			"workspaceId":  workspaceID,
			"changeKind":   "role_updated",
			"targetUserId": userID,
			"role":         role,
		},
	})
}

// EmitUnreadUpdated implements unreadruntime.DeltaSink. The delta is routed to
// the recipient's (workspace, user) live subscription, matching the legacy
// userRoom-targeted Socket.IO emit.
func (emitter *LiveEventEmitter) EmitUnreadUpdated(workspaceID string, delta unread.UnreadDelta) {
	if emitter == nil || emitter.hub == nil {
		return
	}
	trimmedWorkspaceID := strings.TrimSpace(delta.WorkspaceID)
	if trimmedWorkspaceID == "" {
		trimmedWorkspaceID = strings.TrimSpace(workspaceID)
	}
	trimmedUserID := strings.TrimSpace(delta.UserID)
	trimmedMemberID := strings.TrimSpace(delta.MemberID)
	trimmedScopeID := strings.TrimSpace(delta.ScopeID)
	if trimmedWorkspaceID == "" || trimmedUserID == "" || trimmedMemberID == "" || trimmedScopeID == "" {
		return
	}
	// Task mark-read and newly-accounted attention can race between a callback's
	// runtime snapshot and this live-only publish. Serialize publishes for this
	// owner and Task, then re-read the owner runtime while holding that keyed
	// publish lock. A delayed older callback can no longer overwrite newer
	// truth, while a slow subscriber cannot block unrelated owners or Tasks.
	if delta.ScopeType == unread.ScopeTypeTask {
		release := emitter.taskUnreadPublishSequencer.acquire(liveTaskUnreadPublishKey{
			workspaceID: trimmedWorkspaceID,
			memberID:    trimmedMemberID,
			scopeID:     trimmedScopeID,
		})
		defer release()

		emitter.taskUnreadReaderMu.RLock()
		currentReader := emitter.unreadCurrentReader
		emitter.taskUnreadReaderMu.RUnlock()
		if currentReader != nil {
			current, found := currentReader.CurrentUnreadDelta(
				trimmedWorkspaceID,
				trimmedUserID,
				trimmedMemberID,
				trimmedScopeID,
				unread.ScopeTypeTask,
			)
			if found {
				// Always publish the re-read value. The callback that advanced
				// the runtime may have been coalesced or may not itself emit;
				// this invocation is still the repair opportunity.
				delta = current
			} else if delta.ServingState == unread.ServingStateReady ||
				delta.ServingState == unread.ServingStateSnapshotServing {
				// The callback captured externally-servable state that has since
				// been hidden, invalidated, or deleted. Its newer suppression
				// projection (or durable lifecycle event) must win.
				return
			}
		}
	}
	payload := map[string]any{
		"workspaceId":        trimmedWorkspaceID,
		"memberId":           trimmedMemberID,
		"scopeId":            trimmedScopeID,
		"scopeType":          string(delta.ScopeType),
		"count":              delta.Count,
		"manualUnread":       delta.ManualUnread,
		"hasUnread":          delta.HasUnread,
		"workspaceHasUnread": delta.WorkspaceHasUnread,
		"servingState":       string(delta.ServingState),
	}
	// No scope-local callback can order a member-wide aggregate against changes
	// in another scope. Let the browser apply the per-scope hasUnread
	// transition; reconnect bootstrap remains the only live transport boundary
	// that supplies one coherent authoritative aggregate.
	emitter.hub.PublishUserLiveOnlyEvent(trimmedWorkspaceID, trimmedUserID, storeim.StreamEvent{
		WorkspaceID: trimmedWorkspaceID,
		MemberID:    trimmedMemberID,
		ScopeID:     trimmedScopeID,
		Type:        liveEventTypeUnreadUpdated,
		Payload:     payload,
	})
}

// EmitServiceAccountDeleted tells every active workspace client to evict one
// deleted Service Account's Inbox, detail, message, and unread projections.
// The event is live-only: authoritative recovery comes from the Service
// Account directory, conversation index, and workspace snapshot on refetch.
func (emitter *LiveEventEmitter) EmitServiceAccountDeleted(workspaceID string, accountID string, principalID string, scopeID string) {
	if emitter == nil || emitter.hub == nil {
		return
	}
	workspaceID = strings.TrimSpace(workspaceID)
	accountID = strings.TrimSpace(accountID)
	principalID = strings.TrimSpace(principalID)
	scopeID = strings.TrimSpace(scopeID)
	if workspaceID == "" || accountID == "" || principalID == "" || scopeID == "" {
		return
	}
	emitter.hub.PublishWorkspaceLiveOnlyEvent(storeim.StreamEvent{
		WorkspaceID: workspaceID,
		ScopeID:     scopeID,
		Type:        liveEventTypeServiceAccountDeleted,
		Payload: map[string]any{
			"workspaceId": workspaceID,
			"accountId":   accountID,
			"principalId": principalID,
			"scopeId":     scopeID,
		},
	})
}

// EmitAgentCreated broadcasts that a new agent exists in the workspace.
func (emitter *LiveEventEmitter) EmitAgentCreated(workspaceID string) {
	if emitter == nil || emitter.hub == nil {
		return
	}
	workspaceID = strings.TrimSpace(workspaceID)
	if workspaceID == "" {
		return
	}
	emitter.hub.PublishWorkspaceLiveOnlyEvent(storeim.StreamEvent{
		WorkspaceID: workspaceID,
		Type:        liveEventTypeAgentCreated,
	})
}

// EmitAgentDeleted broadcasts that an agent was removed from the workspace.
func (emitter *LiveEventEmitter) EmitAgentDeleted(workspaceID string) {
	if emitter == nil || emitter.hub == nil {
		return
	}
	workspaceID = strings.TrimSpace(workspaceID)
	if workspaceID == "" {
		return
	}
	emitter.hub.PublishWorkspaceLiveOnlyEvent(storeim.StreamEvent{
		WorkspaceID: workspaceID,
		Type:        liveEventTypeAgentDeleted,
	})
}

// EmitTaskCreated fan-outs taskboard task creation to the channel scope.
func (emitter *LiveEventEmitter) EmitTaskCreated(workspaceID string, channelID string, tasks []taskboard.Task) {
	if emitter == nil || emitter.hub == nil {
		return
	}
	workspaceID = strings.TrimSpace(workspaceID)
	channelID = strings.TrimSpace(channelID)
	if workspaceID == "" || channelID == "" {
		return
	}
	payload := map[string]any{"tasks": converter.ToTasksEmpty(tasks)}
	emitter.publishChannelLiveOnlyEvent(workspaceID, channelID, liveEventTypeTaskCreated, payload)
}

// EmitTaskUpdated fan-outs taskboard task update to the channel scope.
func (emitter *LiveEventEmitter) EmitTaskUpdated(workspaceID string, channelID string, task taskboard.Task) {
	if emitter == nil || emitter.hub == nil {
		return
	}
	workspaceID = strings.TrimSpace(workspaceID)
	channelID = strings.TrimSpace(channelID)
	if workspaceID == "" || channelID == "" {
		return
	}
	payload := map[string]any{"task": converter.ToTask(task)}
	emitter.publishChannelLiveOnlyEvent(workspaceID, channelID, liveEventTypeTaskUpdated, payload)
}

// EmitTaskDeleted fan-outs taskboard task deletion to the channel scope.
func (emitter *LiveEventEmitter) EmitTaskDeleted(workspaceID string, channelID string, taskID string) {
	if emitter == nil || emitter.hub == nil {
		return
	}
	workspaceID = strings.TrimSpace(workspaceID)
	channelID = strings.TrimSpace(channelID)
	taskID = strings.TrimSpace(taskID)
	if workspaceID == "" || channelID == "" || taskID == "" {
		return
	}
	emitter.publishChannelLiveOnlyEvent(workspaceID, channelID, liveEventTypeTaskDeleted, map[string]any{"taskId": taskID})
}

// EmitMachineStatus broadcasts a machine status change to the workspace.
func (emitter *LiveEventEmitter) EmitMachineStatus(workspaceID string, machineID string, status string) {
	if emitter == nil || emitter.hub == nil {
		return
	}
	workspaceID = strings.TrimSpace(workspaceID)
	machineID = strings.TrimSpace(machineID)
	status = strings.TrimSpace(status)
	if workspaceID == "" || machineID == "" || status == "" {
		return
	}
	emitter.hub.PublishWorkspaceLiveOnlyEvent(storeim.StreamEvent{
		WorkspaceID: workspaceID,
		Type:        liveEventTypeMachineStatus,
		Payload:     map[string]any{"machineId": machineID, "status": status},
	})
}

// EmitMachineCapabilities broadcasts a machine capabilities payload.
func (emitter *LiveEventEmitter) EmitMachineCapabilities(workspaceID string, machineID string, runtimes []string, runtimeDefaults map[string]runtime.RuntimeDefaults, hostname string, os string, daemonVersion string) {
	if emitter == nil || emitter.hub == nil {
		return
	}
	workspaceID = strings.TrimSpace(workspaceID)
	machineID = strings.TrimSpace(machineID)
	if workspaceID == "" || machineID == "" {
		return
	}
	normalizedRuntimes := runtimes
	if normalizedRuntimes == nil {
		normalizedRuntimes = []string{}
	}
	normalizedRuntimeDefaults := runtimeDefaults
	if normalizedRuntimeDefaults == nil {
		normalizedRuntimeDefaults = map[string]runtime.RuntimeDefaults{}
	}
	emitter.hub.PublishWorkspaceLiveOnlyEvent(storeim.StreamEvent{
		WorkspaceID: workspaceID,
		Type:        liveEventTypeMachineCapabilities,
		Payload: map[string]any{
			"machineId":       machineID,
			"runtimes":        normalizedRuntimes,
			"runtimeDefaults": normalizedRuntimeDefaults,
			"hostname":        strings.TrimSpace(hostname),
			"os":              strings.TrimSpace(os),
			"daemonVersion":   strings.TrimSpace(daemonVersion),
		},
	})
}

// EmitDaemonStatus broadcasts a daemon status change to the workspace.
func (emitter *LiveEventEmitter) EmitDaemonStatus(workspaceID string, daemonID string, status string) {
	if emitter == nil || emitter.hub == nil {
		return
	}
	workspaceID = strings.TrimSpace(workspaceID)
	daemonID = strings.TrimSpace(daemonID)
	status = strings.TrimSpace(status)
	if workspaceID == "" || daemonID == "" || status == "" {
		return
	}
	emitter.hub.PublishWorkspaceLiveOnlyEvent(storeim.StreamEvent{
		WorkspaceID: workspaceID,
		Type:        liveEventTypeDaemonStatus,
		Payload:     map[string]any{"daemonId": daemonID, "status": status},
	})
}

// EmitAgentActivity routes owner-only agent activity to the agent owner.
func (emitter *LiveEventEmitter) EmitAgentActivity(workspaceID string, ownerUserID string, agentID string, activity string, detail string) {
	if emitter == nil {
		return
	}
	emitter.EmitAgentActivityWithInstance(workspaceID, ownerUserID, agentID, activity, detail, nil)
}

// EmitAgentActivityWithInstance is the detailed variant that includes the
// owning agent instance blob.
func (emitter *LiveEventEmitter) EmitAgentActivityWithInstance(workspaceID string, ownerUserID string, agentID string, activity string, detail string, currentInstance *dto.AppAgentInstance) {
	if emitter == nil || emitter.hub == nil {
		return
	}
	workspaceID = strings.TrimSpace(workspaceID)
	ownerUserID = strings.TrimSpace(ownerUserID)
	agentID = strings.TrimSpace(agentID)
	activity = strings.TrimSpace(activity)
	if workspaceID == "" || ownerUserID == "" || agentID == "" || activity == "" {
		return
	}
	payload := map[string]any{
		"agentId":         agentID,
		"activity":        activity,
		"detail":          strings.TrimSpace(detail),
		"currentInstance": currentInstance,
	}
	emitter.hub.PublishUserLiveOnlyEvent(workspaceID, ownerUserID, storeim.StreamEvent{
		WorkspaceID: workspaceID,
		Type:        liveEventTypeAgentActivity,
		Payload:     payload,
	})
}

// EmitAgentStatus broadcasts agent status. The owner receives a detailed
// payload (with currentInstance); other workspace members receive the coarse
// variant. This mirrors the legacy Except(owner)+owner split.
func (emitter *LiveEventEmitter) EmitAgentStatus(workspaceID string, ownerUserID string, agentID string, status string) {
	if emitter == nil {
		return
	}
	emitter.EmitAgentStatusWithInstance(workspaceID, ownerUserID, agentID, status, nil)
}

// EmitAgentStatusWithInstance is the detailed variant of EmitAgentStatus.
func (emitter *LiveEventEmitter) EmitAgentStatusWithInstance(workspaceID string, ownerUserID string, agentID string, status string, currentInstance *dto.AppAgentInstance) {
	if emitter == nil || emitter.hub == nil {
		return
	}
	workspaceID = strings.TrimSpace(workspaceID)
	agentID = strings.TrimSpace(agentID)
	status = strings.TrimSpace(status)
	if workspaceID == "" || agentID == "" || status == "" {
		return
	}
	coarsePayload := map[string]any{"agentId": agentID, "status": status}
	trimmedOwner := strings.TrimSpace(ownerUserID)
	if trimmedOwner != "" {
		workspaceEvent := storeim.StreamEvent{
			WorkspaceID: workspaceID,
			Type:        liveEventTypeAgentStatus,
			Payload:     coarsePayload,
		}
		emitter.hub.PublishWorkspaceLiveOnlyEventExceptUser(workspaceEvent, trimmedOwner)
		emitter.hub.PublishUserLiveOnlyEvent(workspaceID, trimmedOwner, storeim.StreamEvent{
			WorkspaceID: workspaceID,
			Type:        liveEventTypeAgentStatus,
			Payload: map[string]any{
				"agentId":         agentID,
				"status":          status,
				"currentInstance": currentInstance,
			},
		})
		return
	}
	emitter.hub.PublishWorkspaceLiveOnlyEvent(storeim.StreamEvent{
		WorkspaceID: workspaceID,
		Type:        liveEventTypeAgentStatus,
		Payload:     coarsePayload,
	})
}

// EmitAgentSession routes owner-only agent session updates.
func (emitter *LiveEventEmitter) EmitAgentSession(workspaceID string, ownerUserID string, agentID string, sessionID string) {
	if emitter == nil {
		return
	}
	emitter.EmitAgentSessionWithInstance(workspaceID, ownerUserID, agentID, sessionID, nil)
}

// EmitAgentSessionWithInstance is the detailed variant of EmitAgentSession.
func (emitter *LiveEventEmitter) EmitAgentSessionWithInstance(workspaceID string, ownerUserID string, agentID string, sessionID string, currentInstance *dto.AppAgentInstance) {
	if emitter == nil || emitter.hub == nil {
		return
	}
	workspaceID = strings.TrimSpace(workspaceID)
	ownerUserID = strings.TrimSpace(ownerUserID)
	agentID = strings.TrimSpace(agentID)
	if workspaceID == "" || ownerUserID == "" || agentID == "" {
		return
	}
	emitter.hub.PublishUserLiveOnlyEvent(workspaceID, ownerUserID, storeim.StreamEvent{
		WorkspaceID: workspaceID,
		Type:        liveEventTypeAgentSession,
		Payload: map[string]any{
			"agentId":         agentID,
			"sessionId":       strings.TrimSpace(sessionID),
			"currentInstance": currentInstance,
		},
	})
}

// EmitAgentTrajectory routes owner-only agent trajectory deltas.
func (emitter *LiveEventEmitter) EmitAgentTrajectory(workspaceID string, ownerUserID string, agentID string, sessionID string, timestamp int64, entries json.RawMessage) {
	if emitter == nil || emitter.hub == nil {
		return
	}
	workspaceID = strings.TrimSpace(workspaceID)
	ownerUserID = strings.TrimSpace(ownerUserID)
	agentID = strings.TrimSpace(agentID)
	if workspaceID == "" || ownerUserID == "" || agentID == "" || len(entries) == 0 {
		return
	}
	emitter.hub.PublishUserLiveOnlyEvent(workspaceID, ownerUserID, storeim.StreamEvent{
		WorkspaceID: workspaceID,
		Type:        liveEventTypeAgentTrajectory,
		Payload: map[string]any{
			"agentId":   agentID,
			"sessionId": strings.TrimSpace(sessionID),
			"timestamp": timestamp,
			"entries":   json.RawMessage(entries),
		},
	})
}

// EmitAgentWakeStatus stores a TTL-bound wake status snapshot and broadcasts
// it to the channel scope. Recoverability on reconnect comes from
// ReplayWakeStatusesForChannel during the streaming bootstrap.
func (emitter *LiveEventEmitter) EmitAgentWakeStatus(workspaceID string, agentID string, agentName string, agentDisplayName string, messageID string, wakeToken string, target string, status string, replyMessageID string, detail string, timestamp string) {
	if emitter == nil || emitter.hub == nil {
		return
	}
	agentID = strings.TrimSpace(agentID)
	messageID = strings.TrimSpace(messageID)
	wakeToken = strings.TrimSpace(wakeToken)
	status = strings.TrimSpace(status)
	timestamp = strings.TrimSpace(timestamp)
	if agentID == "" || messageID == "" || wakeToken == "" || status == "" || timestamp == "" {
		return
	}
	if emitter.app == nil {
		return
	}
	ctx, cancel := context.WithTimeout(context.Background(), liveEmitterLookupTimeout)
	defer cancel()
	channelID, _, err := emitter.app.LookupMessageChannel(ctx, strings.TrimSpace(workspaceID), messageID)
	if err != nil {
		emitter.logger.Warn("live emitter lookup wake status message channel",
			slog.Any("error", err),
			slog.String("workspace_id", workspaceID),
			slog.String("message_id", messageID),
		)
		return
	}
	channelID = strings.TrimSpace(channelID)
	if channelID == "" {
		emitter.logger.Warn("live emitter skip wake status for unknown message channel",
			slog.String("workspace_id", strings.TrimSpace(workspaceID)),
			slog.String("message_id", messageID),
		)
		return
	}
	emitter.emitAgentWakeStatusForScope(
		workspaceID, channelID, agentID, agentName, agentDisplayName, messageID, wakeToken, target,
		status, replyMessageID, detail, timestamp,
	)
}

// EmitAgentWakeStatusForScope publishes a wake transition when the caller
// already owns the exact durable Scope identity.
func (emitter *LiveEventEmitter) EmitAgentWakeStatusForScope(workspaceID string, channelID string, agentID string, messageID string, wakeToken string, target string, status string, replyMessageID string, detail string, timestamp string) {
	emitter.emitAgentWakeStatusForScope(
		workspaceID, channelID, agentID, "", "", messageID, wakeToken, target,
		status, replyMessageID, detail, timestamp,
	)
}

func (emitter *LiveEventEmitter) emitAgentWakeStatusForScope(workspaceID string, channelID string, agentID string, agentName string, agentDisplayName string, messageID string, wakeToken string, target string, status string, replyMessageID string, detail string, timestamp string) {
	if emitter == nil || emitter.hub == nil {
		return
	}
	workspaceID = strings.TrimSpace(workspaceID)
	channelID = strings.TrimSpace(channelID)
	agentID = strings.TrimSpace(agentID)
	messageID = strings.TrimSpace(messageID)
	wakeToken = strings.TrimSpace(wakeToken)
	status = strings.TrimSpace(status)
	timestamp = strings.TrimSpace(timestamp)
	if workspaceID == "" || channelID == "" || agentID == "" || messageID == "" || wakeToken == "" || status == "" || timestamp == "" {
		return
	}
	snapshot := liveWakeStatusSnapshot{
		agentID:        agentID,
		agentName:      strings.TrimSpace(agentName),
		displayName:    strings.TrimSpace(agentDisplayName),
		channelID:      channelID,
		messageID:      messageID,
		wakeToken:      wakeToken,
		target:         strings.TrimSpace(target),
		status:         status,
		replyMessageID: strings.TrimSpace(replyMessageID),
		detail:         strings.TrimSpace(detail),
		timestamp:      timestamp,
	}
	emitter.storeWakeStatusSnapshot(workspaceID, snapshot)
	emitter.publishChannelLiveOnlyEvent(workspaceID, channelID, liveEventTypeAgentWakeStatus, snapshot.payloadMap())
}

// ClearAgentWakeStatusesForChannel terminates non-expired wake status
// snapshots for one channel+agent and emits the terminal transition.
func (emitter *LiveEventEmitter) ClearAgentWakeStatusesForChannel(workspaceID string, channelID string, agentID string) {
	if emitter == nil {
		return
	}
	terminal := emitter.clearAgentWakeStatusSnapshotsForChannel(workspaceID, channelID, agentID, time.Now())
	if emitter.hub == nil {
		return
	}
	for _, snapshot := range terminal {
		emitter.publishChannelLiveOnlyEvent(workspaceID, snapshot.channelID, liveEventTypeAgentWakeStatus, snapshot.payloadMap())
	}
}

// ReplayWorkspaceRuntimeBootstrap emits the runtime snapshot (machines,
// daemons, agents) onto the supplied writer-style sink. It is invoked by the
// streaming WS handler right after the hello frame, before bootstrap_complete.
func (emitter *LiveEventEmitter) ReplayWorkspaceRuntimeBootstrap(workspaceID string, userID string, snapshot WorkspaceRuntimeSnapshot, emit func(storeim.StreamEvent)) {
	if emitter == nil || emit == nil {
		return
	}
	workspaceID = strings.TrimSpace(workspaceID)
	if workspaceID == "" {
		return
	}
	for _, machine := range snapshot.Machines {
		machineID := strings.TrimSpace(machine.MachineID)
		if machineID == "" {
			continue
		}
		emit(storeim.StreamEvent{
			WorkspaceID: workspaceID,
			Type:        liveEventTypeMachineStatus,
			Payload:     map[string]any{"machineId": machineID, "status": "online"},
		})
		emit(storeim.StreamEvent{
			WorkspaceID: workspaceID,
			Type:        liveEventTypeDaemonStatus,
			Payload:     map[string]any{"daemonId": machineID, "status": "online"},
		})
		for _, agent := range machine.Agents {
			agentID := strings.TrimSpace(agent.AgentID)
			if agentID == "" {
				continue
			}
			if status := strings.TrimSpace(agent.Status); status != "" {
				emit(storeim.StreamEvent{
					WorkspaceID: workspaceID,
					Type:        liveEventTypeAgentStatus,
					Payload:     map[string]any{"agentId": agentID, "status": status},
				})
			}
			if sessionID := strings.TrimSpace(agent.SessionID); sessionID != "" && strings.TrimSpace(machine.OwnerUserID) == strings.TrimSpace(userID) {
				emit(storeim.StreamEvent{
					WorkspaceID: workspaceID,
					Type:        liveEventTypeAgentSession,
					Payload:     map[string]any{"agentId": agentID, "sessionId": sessionID},
				})
			}
		}
	}
}

// ReplayUnreadProjectionState emits the unread projection snapshot for a
// member onto the supplied writer-style sink.
func (emitter *LiveEventEmitter) ReplayUnreadProjectionState(workspaceID string, memberID string, reader UnreadProjectionReader, emit func(storeim.StreamEvent)) {
	if emitter == nil || emit == nil {
		return
	}
	workspaceID = strings.TrimSpace(workspaceID)
	memberID = strings.TrimSpace(memberID)
	if workspaceID == "" || memberID == "" || reader == nil {
		return
	}
	// A bootstrap snapshot is authoritative by omission as well as by value.
	// Clear the browser's previous owner/reconnect epoch before replaying the
	// captured per-scope states; otherwise a hidden/deleted scope, or a member
	// with zero tracked scopes, can retain a stale unread badge forever.
	emit(storeim.StreamEvent{
		WorkspaceID: workspaceID,
		MemberID:    memberID,
		Type:        liveEventTypeUnreadReset,
		Payload: map[string]any{
			"workspaceId": workspaceID,
			"memberId":    memberID,
		},
	})
	for _, payload := range UnreadProjectionReplayPayloads(workspaceID, memberID, reader) {
		emit(storeim.StreamEvent{
			WorkspaceID: payload["workspaceId"].(string),
			MemberID:    payload["memberId"].(string),
			ScopeID:     payload["scopeId"].(string),
			Type:        liveEventTypeUnreadUpdated,
			Payload:     payload,
		})
	}
}

// ReplayWakeStatusesForChannel emits non-expired wake status snapshots
// attached to the channel onto the supplied writer-style sink.
func (emitter *LiveEventEmitter) ReplayWakeStatusesForChannel(workspaceID string, channelID string, emit func(storeim.StreamEvent)) {
	if emitter == nil || emit == nil {
		return
	}
	now := time.Now()
	workspaceID = strings.TrimSpace(workspaceID)
	channelID = strings.TrimSpace(channelID)
	if workspaceID == "" || channelID == "" {
		return
	}
	emitter.wakeStatusMu.Lock()
	defer emitter.wakeStatusMu.Unlock()
	snapshotsByChannel := emitter.wakeStatusBySpace[workspaceID]
	if len(snapshotsByChannel) == 0 {
		return
	}
	snapshots := snapshotsByChannel[channelID]
	if len(snapshots) == 0 {
		return
	}
	for key, snapshot := range snapshots {
		if !snapshot.expiresAt.After(now) {
			delete(snapshots, key)
			continue
		}
		emit(storeim.StreamEvent{
			WorkspaceID: workspaceID,
			ScopeID:     channelID,
			Type:        liveEventTypeAgentWakeStatus,
			Payload:     snapshot.payloadMap(),
		})
	}
	if len(snapshots) == 0 {
		delete(snapshotsByChannel, channelID)
	}
	if len(snapshotsByChannel) == 0 {
		delete(emitter.wakeStatusBySpace, workspaceID)
	}
}

// publishChannelLiveOnlyEvent routes a single live-only event to every
// connection currently subscribed to the channel's visible scope. This
// preserves the Socket.IO channel-room semantics without leaking task or
// wake-status payloads to workspace members who are not viewing the channel;
// non-viewers recover state through REST bootstrap when they open the scope.
func (emitter *LiveEventEmitter) publishChannelLiveOnlyEvent(workspaceID string, channelID string, eventType string, payload map[string]any) {
	if emitter == nil || emitter.hub == nil {
		return
	}
	emitter.hub.PublishScopeLiveOnlyEvent(workspaceID, channelID, storeim.StreamEvent{
		Type:    eventType,
		Payload: payload,
	})
}