import { store } from '../state/store';
import { warnIfNotCanonicalSessionId } from './sessionIdShape';
import { unstable_batchedUpdates } from 'react-dom';
import {
  updateSession,
  updateSessionName,
  updateGroupMeta,
  addMessage,
  addApprovalRequest,
  removeApprovalRequest,
  updateSessionStatus,
  setSessionTestState,
  updateSessionCost,
  updateSessionContext,
  setContextOverflow,
  setRateLimited,
  setProviderRetrying,
  setContextRecovered,
  setAppDepsChanged,
  setMcpSuggestions,
  addBranch,
  setActiveBranch,
  closeSessionFromWs,
  trackAgentNotification,
  setSessionConnState,
  fetchSession,
  recordCompaction,
  setTurnLabel,
  setQueued,
  clearTurnLabel,
} from '../state/agentsSlice';
import { streamStart, streamDelta, streamEnd, clearStreamingForSession } from '../state/streamingSlice';
import { addBrowserCardFromBackend, setBrowserDocked, markBrowserCardEnding, keepBrowserCardOpen, placeBesideCard, placeBelowCard, placeBrowserBesideChat, setBrowserCardPosition, setGlowingBrowserCards, fadeGlowingBrowserCards, clearGlowingBrowserCards, removeBrowserCard, GRID_GAP, WORKFLOW_CARD_GAP, openWorkflowsApp, openWorkflowMonitor } from '../state/dashboardLayoutSlice';
import { upsertOutput } from '../state/outputsSlice';
import { setCardPosition } from '../state/dashboardLayoutSlice';
import { fetchSettings } from '../state/settingsSlice';
import { displaySessionName } from '../state/sessionDisplay';
import { upsertRun, ackRun, runWorkflowNow, openWorkflowCard, upsertWorkflow, removeWorkflow } from '../state/workflowsSlice';
import { stepsSignature } from '@/app/pages/Workflows/scheduleUtils';
import { getAuthToken } from '../config';
import { notifyAgentCompletion, notifyWorkflowRun } from '../notifications';

// Phase 0 boot instrumentation: one-shot flag so we report the first streamed agent token to Electron main exactly once per app launch. Module scope (not instance) because multiple WebSocketManagers exist (one per session WS).
let firstAgentResponseMarked = false;

// Thin wrapper around getAuthToken so the connect() call site stays synchronous. If the token isn't cached yet, returns '' and the WS handshake will 4401, onclose catches that and refreshes the token before the next reconnect.
const _getAuthTokenSafe = (): string => {
  try { return getAuthToken() || ''; } catch { return ''; }
};


const _genUuid = (): string => {
  // Avoid pulling in `crypto.randomUUID` for compat, this is a disambiguator, not a security boundary, so a 96-bit hex string is plenty.
  const a = Math.floor(Math.random() * 2 ** 32).toString(16).padStart(8, '0');
  const b = Math.floor(Math.random() * 2 ** 32).toString(16).padStart(8, '0');
  const c = Math.floor(Math.random() * 2 ** 32).toString(16).padStart(8, '0');
  return `${a}${b}${c}`;
};

type WSEvent = {
  event: string;
  session_id?: string;
  data: Record<string, any>;
  seq?: number;
};

interface WSManagerOptions {
  skipStreamEvents?: boolean;
  // Session-scoped WSes opt into resume + connection-state dispatches by passing this. Dashboard WS doesn't.
  sessionId?: string;
}

// Heartbeat tuning. 25s is below typical aggressive NAT idle timeouts (some enterprise firewalls drop after 30s of silence), and well below browser-tab background throttling thresholds. 10s pong timeout is a balance: long enough to tolerate flaky cellular RTT spikes, short enough that a real dead socket reconnects fast.
const HEARTBEAT_INTERVAL_MS = 25_000;
const HEARTBEAT_TIMEOUT_MS = 10_000;

// Manual runs whose monitor card we've already popped open, so the repeated "workflow:run" updates that stream during a run don't re-pin or re-stack the card.
const autoOpenedRunIds = new Set<string>();

interface QueuedFrame {
  event: string;
  data: Record<string, any>;
  // Lets the future server-side dedup index match retries to originals. Today the server treats most events idempotently anyway (stop on stopped is a no-op), but the client sends this forward-compatibly so a future server upgrade is safe without a protocol bump.
  client_msg_id: string;
}

class WebSocketManager {
  private ws: WebSocket | null = null;
  private url: string;
  private skipStreamEvents: boolean;
  private sessionId: string | null;

  // Resume state. lastSeq is the highest server-assigned seq this client has applied; it's sent on every (re)connect so the server can replay missed events. Persists for the lifetime of this WebSocketManager instance, when the user navigates away and a new createSessionWs() is constructed, lastSeq starts at 0 and we get a full replay.
  private connectionUuid: string;
  private lastSeq: number = 0;
  private resumeAcked: boolean = false;

  private reconnectTimer: ReturnType<typeof setTimeout> | null = null;
  private reconnectDelay = 1000;
  private maxReconnectDelay = 30000;
  // Set to true by `disconnect()` so we don't reconnect after an explicit close (component unmount / user clicks Close).
  private explicitlyClosed: boolean = false;
  private hasConnectedOnce: boolean = false;

  // Heartbeat. We send a ping on a fixed cadence and arm a timeout for the pong; if the timeout fires, we force-close the socket so `onclose` triggers reconnect. Detects laptop-sleep / NAT-drop silent failures that wouldn't otherwise surface until the next outbound send.
  private heartbeatTimer: ReturnType<typeof setInterval> | null = null;
  private pongTimeoutTimer: ReturnType<typeof setTimeout> | null = null;

  // Outbound queue. Frames the user enqueues while the WS isn't OPEN, or while OPEN but pre-resume-ack, wait here and flush after the resume handshake completes. Queue is in-memory only: surviving a full app restart isn't worth the localStorage complexity given how rare that case is for a transient drop.
  private outboundQueue: QueuedFrame[] = [];

  private listeners: Map<string, Set<(data: any) => void>> = new Map();
  // Frame-aligned message coalescer. Buffers incoming WS messages from all WebSocketManager instances and flushes them in ONE batched React render per animation frame. Without this, N concurrent agents each cause their own renders on every WS message, dozens of full app re-renders per second, fanning out to every useSelector. With it: max one render per frame regardless of message volume.
  private static _messageQueue: Array<{ mgr: WebSocketManager; msg: WSEvent }> = [];
  private static _flushScheduled = false;
  // rAF never fires when the window paints no frames (minimized, on another Space, occluded). Without a fallback, WS-delivered dashboard mutations (a spawned browser card, an evict) buffer forever and the UI silently desyncs, agent browser cards just never mount. A timer flushes the queue when rAF won't; whichever fires first wins and cancels the other, so a visible window is unchanged (rAF beats a 250ms timer every frame).
  private static _flushTimer: ReturnType<typeof setTimeout> | null = null;

  private static _enqueueMessage(mgr: WebSocketManager, msg: WSEvent) {
    WebSocketManager._messageQueue.push({ mgr, msg });
    if (WebSocketManager._flushScheduled) return;
    WebSocketManager._flushScheduled = true;
    requestAnimationFrame(WebSocketManager._flushMessages);
    WebSocketManager._flushTimer = setTimeout(WebSocketManager._flushMessages, 250);
  }

  // First moment a flush was deferred for a live drag; bounds the damming so a stuck drag class can't buffer forever.
  private static _dragDeferredAt = 0;

  private static _flushMessages = () => {
    if (!WebSocketManager._flushScheduled) return; // the other trigger already drained this batch
    WebSocketManager._flushScheduled = false;
    if (WebSocketManager._flushTimer !== null) {
      clearTimeout(WebSocketManager._flushTimer);
      WebSocketManager._flushTimer = null;
    }
    if (WebSocketManager._messageQueue.length === 0) return;
    // A live card drag owns the main thread: WS-driven renders mid-drag are what made dragging a
    // working agent feel laggy, and nobody reads streaming tokens while holding a card. Buffer until
    // the pointer settles, hard-capped by time and queue depth.
    if (document.body.classList.contains('dashboard-marquee-active')) {
      if (!WebSocketManager._dragDeferredAt) WebSocketManager._dragDeferredAt = Date.now();
      if (Date.now() - WebSocketManager._dragDeferredAt < 2000 && WebSocketManager._messageQueue.length < 500) {
        WebSocketManager._flushScheduled = true;
        WebSocketManager._flushTimer = setTimeout(WebSocketManager._flushMessages, 100);
        return;
      }
    }
    WebSocketManager._dragDeferredAt = 0;
    const batch = WebSocketManager._messageQueue;
    WebSocketManager._messageQueue = [];
    // unstable_batchedUpdates collapses all dispatches inside the callback into a single React render. Available in React 17; React 18's automatic batching covers this too, but explicit wrap remains correct in both and protects against future batching-context changes.
    unstable_batchedUpdates(() => {
      for (const { mgr, msg } of batch) {
        try {
          mgr.handleMessage(msg);
        } catch (e) {
          console.warn('[ws] message handler threw', e);
        }
      }
    });
  };

  constructor(url: string, options?: WSManagerOptions) {
    this.url = url;
    this.skipStreamEvents = options?.skipStreamEvents ?? false;
    this.sessionId = options?.sessionId ?? null;
    this.connectionUuid = _genUuid();
    // Seed lastSeq from the cross-mount persistent map so a fresh manager (created on every AgentChat remount via key={session.id}) doesn't ask the server to replay events the previous manager already saw. This is the architectural fix for "completed chats re-type themselves on reopen": the server's resume protocol now sees a real high-water mark and has nothing to replay.
    if (this.sessionId) {
      this.lastSeq = _sessionLastSeq.get(this.sessionId) ?? 0;
    }
  }

  // Tokens render as they arrive (claude.ai feel). Per-frame WS batching in _enqueueMessage still coalesces N concurrent agents' messages into ONE React render per animation frame, so removing the pacing layer doesn't reintroduce the parallel-agent re-render storm.
  private dispatchDelta(sessionId: string, messageId: string, delta: string) {
    // Phase 0 boot instrumentation: the first streamed agent token is the "app is actually useful" milestone. Report it once to the Electron main process, which owns the timing log. Guarded by a module-level flag so this is a single no-op branch on every subsequent token.
    if (!firstAgentResponseMarked) {
      firstAgentResponseMarked = true;
      try { (window as any).openswarm?.markFirstAgentResponse?.(); } catch { /* not in Electron */ }
    }
    store.dispatch(streamDelta({ sessionId, messageId, delta }));
  }

  connect() {
    if (this.ws?.readyState === WebSocket.OPEN) return;
    this.explicitlyClosed = false;

    // Append our per-install auth token to the URL. The backend's WS handshake validates this before accepting; without it, any webpage loaded on the same machine could open a WS and read agent traffic. See backend/auth.py + main.py:_ws_auth_ok.
    const token = _getAuthTokenSafe();
    const sep = this.url.includes('?') ? '&' : '?';
    const urlWithToken = token ? `${this.url}${sep}token=${encodeURIComponent(token)}` : this.url;
    this.ws = new WebSocket(urlWithToken);

    this.ws.onopen = () => {
      this.reconnectDelay = 1000;
      this.resumeAcked = false;
      this.startHeartbeat();
      // Send hello immediately so the server can replay anything the server sent that we never applied. On a fresh session, last_seq=0 → server replays from buffer start (empty) and we proceed normally.
      if (this.sessionId) {
        this.sendRaw('client:hello', {
          session_id: this.sessionId,
          connection_uuid: this.connectionUuid,
          last_seq: this.lastSeq,
        });
      } else {
        // Dashboard / global WS: no resume, queue can flush right away.
        this.resumeAcked = true;
        this.flushQueue();
        // Global broadcasts skip the replay log, so anything missed during a socket gap only reappears if subscribers refetch on reconnect.
        if (this.hasConnectedOnce) {
          this.listeners.get('dashboard:reconnected')?.forEach((fn) => fn({}));
        }
      }
      this.hasConnectedOnce = true;
    };

    this.ws.onmessage = (event) => {
      try {
        const msg: WSEvent = JSON.parse(event.data);
        // Pong bypasses the rAF coalescer: rAF stalls when the window gets no frames (minimized, display asleep) while ping timers keep firing, so a buffered pong read as silence and a healthy socket got killed.
        if (msg.event === 'server:pong') {
          this.clearPongTimeout();
          return;
        }
        // Buffer incoming messages and flush them per animation frame in a single React batch. With N concurrent agents/browsers streaming, each WS instance used to trigger its own React render, dozens per frame, fanning out to every useSelector subscriber, starving the main thread. Coalescing flips that to ONE batched render per frame regardless of how many messages arrived. Stream deltas dispatch directly into Redux (no client-side pacing), so the typed-text rate matches what the server sends, the same way claude.ai feels.
        WebSocketManager._enqueueMessage(this, msg);
      } catch {
        // ignore malformed messages
      }
    };

    this.ws.onclose = (ev) => {
      this.stopHeartbeat();
      // 4401 = our backend's auth-failure code. Happens on stale token after backend restart (dev hot-reload). Re-fetch from Electron IPC before retrying.
      if (ev && ev.code === 4401) {
        import('@/shared/config').then(mod => mod.refreshAuthToken().catch(() => {}));
      }
      // Mark UI as reconnecting so the run card shows a clear "trying to reconnect" state rather than implying the run died. Skipped on an explicit disconnect (user navigated away) since there's no run to surface state for.
      if (this.sessionId && !this.explicitlyClosed) {
        store.dispatch(setSessionConnState({
          sessionId: this.sessionId,
          state: 'reconnecting',
        }));
      }
      if (!this.explicitlyClosed) this.scheduleReconnect();
    };

    this.ws.onerror = () => {
      // Force the close path to run, onclose will mark state reconnecting and schedule a retry.
      this.ws?.close();
    };
  }

  disconnect() {
    this.explicitlyClosed = true;
    if (this.reconnectTimer) {
      clearTimeout(this.reconnectTimer);
      this.reconnectTimer = null;
    }
    this.stopHeartbeat();
    this.ws?.close();
    this.ws = null;
  }

  private scheduleReconnect() {
    if (this.reconnectTimer) return;
    // No retry cap. Long-horizon agent runs may outlast a multi-hour network outage (overnight laptop sleep, captive portal limbo); giving up would silently desync the UI. Backoff is bounded at 30s so the user-visible "Reconnecting…" loop never hammers the network, and a small jitter prevents thundering-herd if many session WSes reconnect at once after a backend restart.
    const jitter = 0.8 + Math.random() * 0.4; // ±20%
    const delay = Math.min(this.reconnectDelay, this.maxReconnectDelay) * jitter;
    this.reconnectTimer = setTimeout(() => {
      this.reconnectTimer = null;
      this.reconnectDelay = Math.min(this.reconnectDelay * 2, this.maxReconnectDelay);
      this.connect();
    }, delay);
  }

  private startHeartbeat() {
    this.stopHeartbeat();
    this.heartbeatTimer = setInterval(() => {
      this.sendPing();
    }, HEARTBEAT_INTERVAL_MS);
  }

  private stopHeartbeat() {
    if (this.heartbeatTimer != null) {
      clearInterval(this.heartbeatTimer);
      this.heartbeatTimer = null;
    }
    if (this.pongTimeoutTimer != null) {
      clearTimeout(this.pongTimeoutTimer);
      this.pongTimeoutTimer = null;
    }
  }

  private sendPing() {
    if (this.ws?.readyState !== WebSocket.OPEN) return;
    const nonce = _genUuid();
    try {
      this.ws.send(JSON.stringify({ event: 'client:ping', data: { nonce } }));
    } catch {
      // socket dying, let the close handler take over
      return;
    }
    if (this.pongTimeoutTimer != null) clearTimeout(this.pongTimeoutTimer);
    this.pongTimeoutTimer = setTimeout(() => {
      // Silent death: no pong arrived in time. Force a close so the browser's onclose path (and our reconnect) runs immediately instead of waiting for the OS TCP keepalive (~75s).
      try { this.ws?.close(); } catch { /* nothing */ }
    }, HEARTBEAT_TIMEOUT_MS);
  }

  private clearPongTimeout() {
    if (this.pongTimeoutTimer != null) {
      clearTimeout(this.pongTimeoutTimer);
      this.pongTimeoutTimer = null;
    }
  }

  private flushQueue() {
    if (this.ws?.readyState !== WebSocket.OPEN) return;
    if (!this.resumeAcked) return;
    const queue = this.outboundQueue;
    this.outboundQueue = [];
    for (const frame of queue) {
      try {
        this.ws.send(JSON.stringify({ event: frame.event, data: frame.data }));
      } catch {
        // Re-queue and bail; reconnect will retry.
        this.outboundQueue.unshift(frame);
        break;
      }
    }
  }

  // Direct send that bypasses the queue. Used for hello/ping which must NOT be queued (they're connection-scoped, not session-data).
  private sendRaw(event: string, data: Record<string, any>) {
    if (this.ws?.readyState !== WebSocket.OPEN) return;
    try { this.ws.send(JSON.stringify({ event, data })); } catch { /* nothing */ }
  }

  private handleMessage(msg: WSEvent) {
    const { event, session_id, data } = msg;

    // Update lastSeq for events that carry one. seq is monotonic per session, so this is the high-water mark we send back on resume.
    if (typeof msg.seq === 'number' && msg.seq > this.lastSeq) {
      this.lastSeq = msg.seq;
      // Mirror to the module-scope persistent map so the next fresh manager (next AgentChat remount) starts here, not at zero.
      if (this.sessionId) {
        _sessionLastSeq.set(this.sessionId, this.lastSeq);
      }
    }

    // Cross-socket dedupe: the backend fans every session frame out to BOTH the dashboard socket and the chat's own socket (same stamped seq), so an expanded chat parsed and reduced everything twice, and whichever copy landed second could be a replayed stale one. Time-windowed rather than a high-water mark so a deliberate later replay (gap recovery resets lastSeq to 0) is never starved.
    if (typeof msg.seq === 'number' && session_id) {
      const key = `${session_id}:${msg.seq}`;
      const now = Date.now();
      const seen = _recentFrameTimes.get(key);
      if (seen !== undefined && now - seen < FRAME_DEDUPE_WINDOW_MS) return;
      _recentFrameTimes.set(key, now);
      if (_recentFrameTimes.size > 4000) {
        for (const [k, t] of _recentFrameTimes) {
          if (now - t >= FRAME_DEDUPE_WINDOW_MS) _recentFrameTimes.delete(k);
        }
      }
    }

    // ----- Connection-scoped frames (no business-logic side effects) -----

    if (event === 'server:pong') {
      this.clearPongTimeout();
      return;
    }

    if (event === 'server:hello') {
      // Resume handshake completed. The server has either replayed missed events (which arrived as separate frames before this ack), surfaced a gap, or signalled "you're caught up." Mark ourselves live and flush any queued outbound frames.
      this.resumeAcked = true;
      if (this.sessionId) {
        store.dispatch(setSessionConnState({
          sessionId: this.sessionId,
          state: 'live',
        }));
      }
      this.flushQueue();
      return;
    }

    if (event === 'agent:gap_detected') {
      // We were offline long enough that the server's ring buffer rolled past our lastSeq. Re-fetch authoritative state via REST so the slice's view doesn't have a silent gap.
      if (session_id) {
        store.dispatch(fetchSession(session_id));
        // Reset lastSeq, the REST refetch is the new authoritative baseline; subsequent server events with seq numbers will re-establish the high-water mark. Also wipe the cross-mount persistent map so a remount during this gap window doesn't resurrect the stale value.
        this.lastSeq = 0;
        _sessionLastSeq.delete(session_id);
        // The recovery replay re-delivers seqs possibly seen moments ago; drop them from the dedupe window so it's never starved.
        for (const k of _recentFrameTimes.keys()) {
          if (k.startsWith(`${session_id}:`)) _recentFrameTimes.delete(k);
        }
      }
      return;
    }

    if (this.skipStreamEvents) {
      if (event === 'agent:stream_start' || event === 'agent:stream_delta' || event === 'agent:stream_end') {
        return;
      }
    }

    switch (event) {
      case 'agent:test_state':
        // broadcast_global puts everything under data (no top-level session_id).
        if (data.session_id && data.state) {
          store.dispatch(setSessionTestState({ sessionId: data.session_id, state: data.state }));
        }
        break;

      case 'agent:status':
        // Capture pre-transition status so we only fire a system notification on a real running→terminal transition. Otherwise a session that was already 'completed' on disk and got refetched would re-toast.
        {
          const prevSession = session_id ? store.getState().agents.sessions[session_id] : undefined;
          const prevStatus = prevSession?.status;

          if (data.session) {
            store.dispatch(updateSession(data.session));
          } else if (session_id) {
            store.dispatch(updateSessionStatus({ sessionId: session_id, status: data.status }));
          }
          if (data.status === 'running' && session_id) {
            store.dispatch(trackAgentNotification(session_id));
          }
          // Native OS notification when an agent finishes while the user is elsewhere: workflows already had this; long chat tasks deserve the same "it's done" tap on both platforms. Sub-agents stay silent (their parent's finish is the story).
          if (data.status === 'completed' && session_id && document.hidden) {
            const p_sess2 = data.session ?? store.getState().agents.sessions[session_id];
            if (p_sess2 && !p_sess2.parent_session_id && p_sess2.mode !== 'browser-agent') {
              void (window as any).openswarm?.notify?.({
                title: 'Agent finished',
                body: (p_sess2.name && p_sess2.name !== 'Untitled' ? p_sess2.name : 'Your task is done.').slice(0, 200),
                deepLink: `openswarm://session/${session_id}`,
              });
            }
          }

          // An AppAgent driving an app card announces itself only via this status event (no card_added like browsers), so light the app card here. Keyed by the parent chat like browser glows, so the same terminal fade below clears it.
          const p_sess = data.session;
          if (p_sess && p_sess.mode === 'browser-agent' && typeof p_sess.browser_id === 'string' && p_sess.browser_id.startsWith('app:')
              && (p_sess.status === 'running' || p_sess.status === 'waiting_approval')) {
            store.dispatch(setGlowingBrowserCards({
              browserIds: [p_sess.browser_id],
              sessionId: p_sess.parent_session_id || p_sess.id,
              label: 'Use App',
            }));
          }

          // Fade this session's browser glows on the terminal transition HERE, not only in AgentChat's effect: a collapsed chat is unmounted at finish, and a never-faded glow pins the browser's renderer (exempt from suspend + the webview cap) forever.
          const newStatus = data.status ?? data.session?.status;
          const wasWorking = prevStatus === 'running' || prevStatus === 'waiting_approval';
          if (session_id && wasWorking && (newStatus === 'completed' || newStatus === 'stopped' || newStatus === 'error')) {
            store.dispatch(fadeGlowingBrowserCards(session_id));
            setTimeout(() => store.dispatch(clearGlowingBrowserCards(session_id)), 600);
          }

          // Fire a native notification when an agent terminates while the window is hidden. Skips sub-agents and browser-agents (the parent's own completion is what the user cares about) and only fires on a real transition from a non-terminal state.
          const TERMINAL = new Set(['completed', 'error']);
          const NON_TERMINAL = new Set(['running', 'waiting_approval', undefined, null, '']);
          if (
            session_id &&
            TERMINAL.has(data.status) &&
            NON_TERMINAL.has(prevStatus as any) &&
            data.session?.mode !== 'browser-agent' &&
            data.session?.mode !== 'sub-agent' &&
            data.session?.mode !== 'invoked-agent'
          ) {
            const sess = data.session ?? prevSession;
            if (sess) {
              const lastAssistant = [...(sess.messages || [])]
                .reverse()
                .find((m: any) => m.role === 'assistant' && typeof m.content === 'string');
              notifyAgentCompletion({
                sessionId: session_id,
                sessionName: displaySessionName(sess.name),
                dashboardId: sess.dashboard_id,
                status: data.status as 'completed' | 'error',
                bodyExcerpt: lastAssistant ? String(lastAssistant.content) : undefined,
              });
            }
          }

          // Clear any leftover turn label when the agent reaches a terminal state so the next turn doesn't show a stale label before its own aux call lands.
          if (session_id && (data.status === 'completed' || data.status === 'error' || data.status === 'stopped')) {
            store.dispatch(clearTurnLabel(session_id));
            // Mid-stream stop: the backend keeps the partial reply, but cancelling the SDK can lag several seconds, so its authoritative agent:message lands late. Promote the in-flight assistant text to a real message NOW so it doesn't blink out the instant we clear the overlay; the late agent:message (same id) just refreshes it.
            if (data.status === 'stopped') {
              const entry = store.getState().streaming.bySession[session_id];
              if (entry && entry.role === 'assistant' && entry.content) {
                const sess = store.getState().agents.sessions[session_id];
                if (!sess?.messages?.some((m) => m.id === entry.id)) {
                  store.dispatch(addMessage({
                    sessionId: session_id,
                    message: {
                      id: entry.id,
                      role: 'assistant',
                      content: entry.content,
                      timestamp: new Date().toISOString(),
                      branch_id: sess?.active_branch_id ?? 'main',
                      parent_id: null,
                    },
                  }));
                }
              }
            }
            store.dispatch(clearStreamingForSession(session_id));
          }
        }
        // Clean spawned browser cards when the PARENT finishes, never per sub-agent: the parent reuses the same browser_id for its next step, so deleting on sub-agent completion strands BrowserAgent(browser_id) on a dead card. 'stopped' skipped to allow inspect-after-manual-stop. Mark the card as ending instead of removing immediately so the card shows a fade + Keep pill; BrowserCard owns the 3s timer to remove.
        if (
          session_id &&
          (data.status === 'completed' || data.status === 'error') &&
          data.session?.mode !== 'browser-agent'
        ) {
          const browserCards = store.getState().dashboardLayout.browserCards;
          for (const card of Object.values(browserCards)) {
            if (card.spawned_by === session_id && !card.keep_open) {
              store.dispatch(markBrowserCardEnding({
                browserId: card.browser_id, status: data.status,
              }));
            }
          }
        }
        break;

      case 'agent:message':
        if (session_id && data.message) {
          store.dispatch(addMessage({ sessionId: session_id, message: data.message }));
        }
        break;

      case 'agent:output_upserted':
        // Emitted by the backend when an Output row is created (canvas-launched App Builder seed) or updated (post-session meta.json sync). The upsert reducer merges over an existing row so a UI that already loaded the row doesn't lose locally-applied fields.
        if (data.output && data.output.id) {
          store.dispatch(upsertOutput(data.output));
        }
        break;

      case 'apps_sdk:tool_grant_request':
        // An app asked to use a connected MCP tool; AppToolGrantHost (global mount) owns the dialog.
        window.dispatchEvent(new CustomEvent('openswarm:app-tool-grant', { detail: data }));
        break;

      case 'apps_sdk:place_agent_card': {
        // An app asked for its spawned agent at a specific canvas spot; the card is created async
        // by the session lifecycle, so nudge it into place with a short bounded retry.
        const sid = data.session_id as string;
        const px = Number(data.x);
        const py = Number(data.y);
        if (!sid || !Number.isFinite(px) || !Number.isFinite(py)) break;
        let tries = 0;
        const place = () => {
          const exists = !!store.getState().dashboardLayout.cards[sid];
          if (exists) {
            store.dispatch(setCardPosition({ sessionId: sid, x: px, y: py }));
            return;
          }
          if (++tries < 20) setTimeout(place, 300);
        };
        place();
        break;
      }

      case 'agent:stream_start':
      case 'agent:stream_delta':
      case 'agent:stream_end':
        // Replay-skip guard. The WS resume protocol replays buffered events from the ring buffer with seq > last_seq. When this manager is freshly constructed (every AgentChat mount, because of `key={session.id}`), last_seq is 0, so the server replays EVERY buffered stream_* event for the session. Without this guard, opening any chat with prior streaming turns would replay every buffered delta as a live stream event, re-triggering the streaming UI on every reopen. The discriminator is `resumeAcked`: it flips to true when server:hello arrives, which the server sends AFTER the replay completes. Any stream_* event arriving while !resumeAcked is replay-from-buffer (historical) and can be dropped, the REST snapshot we awaited before connect is authoritative for any already-finalized message, and any genuinely live turn the server is pushing will continue emitting events after the ack.
        if (!this.resumeAcked) break;
        if (event === 'agent:stream_start') {
          if (session_id && data.message_id) {
            store.dispatch(streamStart({
              sessionId: session_id,
              messageId: data.message_id,
              role: data.role,
              toolName: data.tool_name,
            }));
          }
        } else if (event === 'agent:stream_delta') {
          if (session_id && data.message_id) {
            this.dispatchDelta(session_id, data.message_id, data.delta);
          }
        } else if (event === 'agent:stream_end') {
          if (session_id && data.message_id) {
            store.dispatch(streamEnd({
              sessionId: session_id,
              messageId: data.message_id,
            }));
          }
        }
        break;

      case 'agent:approval_request':
        if (session_id) {
          store.dispatch(addApprovalRequest({
            sessionId: session_id,
            request: {
              id: data.request_id,
              session_id: session_id,
              tool_name: data.tool_name,
              tool_input: data.tool_input,
              created_at: new Date().toISOString(),
              sensitive_pattern: data.sensitive_pattern ?? null,
              sensitive_label: data.sensitive_label ?? null,
              sensitive_why: data.sensitive_why ?? null,
            },
          }));
        }
        break;

      case 'agent:cost_update':
        if (session_id) {
          store.dispatch(updateSessionCost({
            sessionId: session_id,
            costUsd: data.cost_usd,
          }));
        }
        break;

      case 'agent:context_update':
        if (session_id) {
          store.dispatch(updateSessionContext({
            sessionId: session_id,
            inputTokens: data.input_tokens ?? 0,
            outputTokens: data.output_tokens ?? 0,
            cacheReadTokens: data.cache_read_tokens ?? 0,
            cacheReadPct: data.cache_read_pct ?? 0,
            ctxUsedPct: data.ctx_used_pct ?? 0,
            contextWindow: typeof data.context_window === 'number' ? data.context_window : undefined,
            frameworkOverheadTokens: typeof data.framework_overhead_tokens === 'number' ? data.framework_overhead_tokens : undefined,
            activeMcps: Array.isArray(data.active_mcps) ? data.active_mcps : [],
          }));
        }
        break;

      case 'agent:context_overflow':
        if (session_id) {
          store.dispatch(setContextOverflow({
            sessionId: session_id,
            reason: data.reason ?? 'long_context_required',
            message: data.message ?? 'Context full.',
          }));
        }
        break;

      case 'agent:provider_retrying':
        // Mid-turn CLI backoff: the provider 500/429'd and the CLI is silently waiting; show the muted pill so the card doesn't read as dead (ENG-178).
        if (session_id) {
          store.dispatch(setProviderRetrying({
            sessionId: session_id,
            attempt: typeof data.attempt === 'number' ? data.attempt : null,
            delayMs: typeof data.delay_ms === 'number' ? data.delay_ms : null,
          }));
        }
        break;

      case 'agent:rate_limited':
        // Provider throttle that outlasted the silent backoff. Transient muted pill, not a card; auto-clears frontend-side.
        if (session_id) {
          store.dispatch(setRateLimited({
            sessionId: session_id,
            retryAfterS: typeof data.retry_after_s === 'number' ? data.retry_after_s : null,
          }));
        }
        break;

      case 'agent:context_recovered':
        // The backend hit a context-overflow crash mid-turn, rebuilt from its local copy, and retried on its own. Transient muted pill so the recovery is visible without reading like an error.
        if (session_id) {
          store.dispatch(setContextRecovered({ sessionId: session_id }));
        }
        break;

      case 'agent:app_deps_changed':
        // A view-builder agent installed/changed deps this turn; the app card escalates its turn-finish reload from soft to a Vite restart so new deps actually load.
        if (session_id) {
          store.dispatch(setAppDepsChanged({ sessionId: session_id }));
        }
        break;

      case 'agent:context_status':
        // Auto-compaction collapsed older turns into a summary. Mirror compacted_through_msg_id locally so the renderer can drop a visible "N earlier turns summarized" chip into the transcript. Other reasons (cleared, etc.) flow through this same event but don't currently need a chip, ignore them for now.
        if (session_id && data.reason === 'compacted') {
          store.dispatch(recordCompaction({
            sessionId: session_id,
            throughMsgId: data.compacted_through_msg_id ?? null,
          }));
        }
        break;

      case 'agent:turn_label':
        // Aux-LLM-generated verb-phrase for the current turn. Replaces the static "Thinking…" label until the turn ends, then the ThinkingBubble freezes to "Thought for Ns · M tokens".
        if (session_id && data.label) {
          store.dispatch(setTurnLabel({
            sessionId: session_id,
            turnId: data.turn_id || '',
            label: data.label,
          }));
        }
        break;

      case 'agent:queued':
        // Admission gate: this turn is waiting for a concurrency slot; shows a "queued" chip so it doesn't read as a hung "working".
        if (session_id) store.dispatch(setQueued({ sessionId: session_id, queued: true }));
        break;

      case 'agent:admitted':
        // Slot acquired; the turn is about to stream. Clears the queued chip.
        if (session_id) store.dispatch(setQueued({ sessionId: session_id, queued: false }));
        break;

      case 'agent:auth_error':
        // Re-uses the context_overflow card slot, both are "this session is blocked, here's what to do" cards. Reason field disambiguates.
        if (session_id) {
          store.dispatch(setContextOverflow({
            sessionId: session_id,
            reason: data.reason ?? 'auth_error',
            message: data.message ?? 'Authentication failed.',
          }));
        }
        break;

      case 'agent:out_of_tokens':
        // Reuses the context_overflow card slot, same "this session is blocked, here's what to do" shape. Reason field disambiguates from auth/overflow.
        if (session_id) {
          store.dispatch(setContextOverflow({
            sessionId: session_id,
            reason: 'out_of_tokens',
            message: data.message ?? "You're out of tokens on this model.",
          }));
        }
        break;

      case 'agent:mcp_suggestions':
        if (session_id) {
          store.dispatch(setMcpSuggestions({
            sessionId: session_id,
            suggestions: Array.isArray(data.suggestions) ? data.suggestions : [],
            isVague: !!data.is_vague,
          }));
        }
        break;

      case 'agent:branch_created':
        if (session_id && data.branch) {
          store.dispatch(addBranch({ sessionId: session_id, branch: data.branch }));
          store.dispatch(setActiveBranch({ sessionId: session_id, branchId: data.active_branch_id }));
        }
        break;

      case 'agent:branch_switched':
        if (session_id) {
          store.dispatch(setActiveBranch({ sessionId: session_id, branchId: data.active_branch_id }));
        }
        break;

      case 'agent:name_updated':
        if (session_id && data.name) {
          store.dispatch(updateSessionName({ sessionId: session_id, name: data.name }));
        }
        break;

      case 'agent:group_meta_updated':
        if (session_id && data.group_id) {
          store.dispatch(updateGroupMeta({
            sessionId: session_id,
            groupId: data.group_id,
            name: data.name ?? '',
            svg: data.svg ?? '',
            isRefined: data.is_refined ?? false,
          }));
        }
        break;

      case 'agent:closed':
        if (session_id) {
          const closedStatus = data.status ?? 'stopped';
          // Don't evict a chat the user is actively watching from a workflow card; let it settle into a normal completed chat they can continue or close themselves.
          const watchedSidecar = Object.values(store.getState().workflows.openCards)
            .some((oc) => oc.sidecarSessionId === session_id);
          // The Run Monitor watches a run via its workflow id, not a sidecar: keep that run's session so the transcript survives completion.
          const monWf = store.getState().dashboardLayout.workflowsMonitorId;
          const watchedByMonitor = !!monWf
            && (store.getState().workflows.runs[monWf] || []).some((r) => r.session_id === session_id);
          store.dispatch(closeSessionFromWs({
            id: session_id,
            name: data.name ?? 'Untitled',
            status: closedStatus,
            model: data.model ?? '',
            mode: data.mode ?? '',
            created_at: data.created_at ?? new Date().toISOString(),
            closed_at: data.closed_at ?? new Date().toISOString(),
            cost_usd: data.cost_usd ?? 0,
            dashboard_id: data.dashboard_id,
            keepSession: watchedSidecar || watchedByMonitor,
          }));
          // Auto-delete browsers spawned by this agent when it finishes normally or errors out. We intentionally skip 'stopped', the user may want to inspect the browser after manually stopping. Mark for removal so BrowserCard renders a fade + Keep pill; the card itself runs the 3s timer to dispatch removeBrowserCard.
          if (closedStatus === 'completed' || closedStatus === 'error') {
            const browserCards = store.getState().dashboardLayout.browserCards;
            for (const card of Object.values(browserCards)) {
              if (card.spawned_by === session_id && !card.keep_open) {
                store.dispatch(markBrowserCardEnding({
                  browserId: card.browser_id, status: closedStatus,
                }));
              }
            }
          }
        }
        break;

      case 'workflow:run':
        if (data.run) {
          const run = data.run;
          store.dispatch(upsertRun(run));
          // A manual run (the Run button OR the edit agent's RunWorkflowNow) should surface its live card the moment it starts. Fire once per run so the run's later tool-call updates don't keep re-pinning the monitor; scheduled runs stay quiet so they never hijack the canvas.
          if (run.status === 'running' && run.triggered_by === 'manual' && run.id && !autoOpenedRunIds.has(run.id)) {
            autoOpenedRunIds.add(run.id);
            store.dispatch(openWorkflowMonitor({ workflowId: run.workflow_id, runId: run.id }));
          }
        }
        break;

      case 'workflow:updated':
        if (data.workflow) {
          store.dispatch(upsertWorkflow(data.workflow));
        }
        break;

      case 'workflow:deleted':
        if (data.workflow_id) {
          store.dispatch(removeWorkflow(data.workflow_id));
        }
        break;

      case 'workflow:notify':
        if (data.workflow_id) {
          notifyWorkflowRun({
            workflowId: data.workflow_id,
            workflowTitle: data.workflow_title || 'Workflow',
            runId: data.run_id,
            sessionId: data.session_id,
            status: data.status,
            tierKind: data.tier_kind,
            fallback: data.fallback,
          });
        }
        break;

      case 'dashboard:browser_card_keep':
        if (data.browser_id) {
          store.dispatch(keepBrowserCardOpen(data.browser_id));
        }
        break;

      case 'dashboard:browser_card_evict':
        // A wedged card the backend is tearing down BEFORE it spawns a recovery card. Remove it now (no fade, no Keep pill) so its <webview> unmounts and stops starving the renderer while the fresh card mounts.
        if (data.browser_id) {
          store.dispatch(removeBrowserCard(data.browser_id));
        }
        break;

      case 'dashboard:browser_card_added':
        if (data.browser_card) {
          // Tag with origin dashboard so the card renders only on the dashboard that spawned it, without this, a browser spawned by an agent on dashboard A leaks into whatever dashboard the user is currently viewing (the global browserCards dict + unfiltered render).
          store.dispatch(addBrowserCardFromBackend({
            ...data.browser_card,
            dashboard_id: data.dashboard_id,
          }));
          const parentId = data.parent_session_id;
          if (parentId) {
            const layoutState = store.getState().dashboardLayout;
            const browserCard = layoutState.browserCards[data.browser_card.browser_id];
            if (browserCard) {
              const exclude = { type: 'browser' as const, id: browserCard.browser_id };
              const parentCard = layoutState.cards[parentId];
              // Workflow chats have no standalone agent card: a run lives in the monitor (dock beside it), an edit/compose chat lives in the hub window (dock below it). Without this the browser keeps the backend's default spot, which overlaps the Workflows window.
              const sess = store.getState().agents.sessions[parentId];
              let pos: { x: number; y: number } | null = null;
              let glowLabel = 'Use Browser';
              if (parentCard) {
                pos = placeBrowserBesideChat(layoutState, parentCard, parentId, browserCard.width, browserCard.height, browserCard.browser_id);
                // A replacement browser buries its predecessor: the agent spins a fresh card when a tab dies, and the displaced spawned sibling must be torn down through the FULL path (a bare store delete gets resurrected by the backend layout sync as a free stacked window).
                for (const old of Object.values(layoutState.browserCards)) {
                  if (old.browser_id !== browserCard.browser_id && old.spawned_by === parentId && !old.keep_open) {
                    void import('@/shared/browserTeardown').then(({ removeBrowserCardCleanly }) => removeBrowserCardCleanly(old.browser_id, store.dispatch));
                  }
                }
                // Default home is INSIDE the chat: the card overlays the chat's dock slot while the
                // chat is expanded; the beside-chat spot stays the undock/collapse fallback.
                store.dispatch(setBrowserDocked({ browserId: browserCard.browser_id, dockedTo: parentId }));
              } else if (sess?.workflow_run_id && layoutState.workflowsMonitorCard) {
                pos = placeBesideCard(layoutState, layoutState.workflowsMonitorCard, browserCard.width, browserCard.height, undefined, exclude, WORKFLOW_CARD_GAP, true);
              } else if (sess?.workflow_edit_id && layoutState.workflowsHub) {
                pos = placeBelowCard(layoutState, layoutState.workflowsHub, browserCard.width, browserCard.height, undefined, exclude);
                glowLabel = 'Browser';
              }
              if (pos) {
                store.dispatch(setBrowserCardPosition({
                  browserId: data.browser_card.browser_id,
                  x: pos.x,
                  y: pos.y,
                }));
                store.dispatch(setGlowingBrowserCards({
                  browserIds: [data.browser_card.browser_id],
                  sessionId: parentId,
                  label: glowLabel,
                }));
              }
            }
          }
        }
        break;

      case 'settings:changed':
        // An agent wrote settings under us (not the user via the modal), so refetch now instead of waiting for the next window-focus. The slice's latestWriteId guard drops this if a newer user save is already in flight.
        store.dispatch(fetchSettings());
        break;
    }

    // Notify any custom listeners
    const handlers = this.listeners.get(event);
    if (handlers) {
      handlers.forEach((fn) => fn({ session_id, ...data }));
    }
  }

  send(event: string, data: Record<string, any>) {
    // Queue if the socket isn't open OR resume hasn't been ack'd yet. The pre-ack gate prevents an outbound user message from racing the resume replay, the server might process the message before the replay finishes, leaving the slice's view of history incomplete.
    const open = this.ws?.readyState === WebSocket.OPEN;
    if (!open || !this.resumeAcked) {
      this.outboundQueue.push({ event, data, client_msg_id: _genUuid() });
      return;
    }
    try {
      this.ws!.send(JSON.stringify({ event, data }));
    } catch {
      this.outboundQueue.push({ event, data, client_msg_id: _genUuid() });
    }
  }

  sendMessage(
    sessionId: string,
    prompt: string,
    opts?: { mode?: string; model?: string; provider?: string; images?: Array<{ data: string; media_type: string }> },
  ) {
    this.send('agent:send_message', {
      session_id: sessionId,
      prompt,
      ...opts,
    });
  }

  sendApproval(requestId: string, behavior: 'allow' | 'deny', message?: string) {
    this.send('agent:approval_response', {
      request_id: requestId,
      behavior,
      message,
    });
  }

  stopAgent(sessionId: string) {
    this.send('agent:stop', { session_id: sessionId });
  }

  on(event: string, handler: (data: any) => void) {
    if (!this.listeners.has(event)) {
      this.listeners.set(event, new Set());
    }
    this.listeners.get(event)!.add(handler);
    return () => this.listeners.get(event)?.delete(handler);
  }

  get connected(): boolean {
    return this.ws?.readyState === WebSocket.OPEN;
  }
}

import { WS_BASE } from '@/shared/config';

// Bridge native-notification button actions to workflow actions. Subscribe at module import time so we never miss an early callback fired before any component mounts.
(() => {
  try {
    const w: any = (typeof window !== 'undefined') ? (window as any).openswarm : null;
    if (!w?.onNotificationAction) return;
    w.onNotificationAction(({ outcome, runId, workflowId }: { outcome: string; runId?: string; workflowId?: string }) => {
      if (!workflowId) return;
      if (outcome === 'ack' && runId) {
        store.dispatch(ackRun(runId));
        return;
      }
      if (outcome === 'rerun') {
        const wf = store.getState().workflows.items[workflowId];
        store.dispatch(wf ? runWorkflowNow({ id: workflowId, signature: stepsSignature(wf.steps) }) : runWorkflowNow(workflowId));
        return;
      }
      if (outcome === 'edit' || outcome === 'open') {
        store.dispatch(openWorkflowsApp({ workflowId }));
        return;
      }
    });
  } catch { /* native notifications optional */ }
})();

export const dashboardWs = new WebSocketManager(`${WS_BASE}/ws/dashboard`, { skipStreamEvents: true });

// Per-session high-water mark for the resume protocol. Survives across AgentChat mounts/unmounts so reopening a chat doesn't re-trigger a full replay from the server's ring buffer. Why this exists: AgentChat uses `key={session.id}` on the embedded instance inside AgentCard, so every expand/collapse remounts the component, which constructs a fresh WebSocketManager. Without this persistent map, each fresh manager starts at last_seq=0 and asks the server for the entire buffered history. The server faithfully replays it, the client renders the typewriter animation again, and the user sees their completed chat "type itself out" on every reopen. Lifetime: tied to the JS module load, which means the page tab. Lost on full app reload (intentional, that should re-hydrate from REST). On backend restart the buffers are wiped anyway, so a stale lastSeq pointing past the buffer top falls into the "fresh client" path on the server (last_seq>0 but no buffer) which short-circuits to a no-op replay. Safe.
const _sessionLastSeq: Map<string, number> = new Map();

// (session_id:seq) -> arrival time; entries older than the window are prunable. Bounded by event rate x window, not session count.
const FRAME_DEDUPE_WINDOW_MS = 5_000;
const _recentFrameTimes: Map<string, number> = new Map();

/** Seed the resume cursor from a REST hydrate (GET /sessions returns event_seq), so the follow-up WS connect replays only what happened AFTER the snapshot instead of the whole ring buffer the client just received as JSON. Never lowers an existing high-water mark. */
export function seedSessionSeq(sessionId: string, seq: number): void {
  if (typeof seq !== 'number' || seq <= 0) return;
  const cur = _sessionLastSeq.get(sessionId) ?? 0;
  if (seq > cur) _sessionLastSeq.set(sessionId, seq);
}

export function createSessionWs(sessionId: string): WebSocketManager {
  warnIfNotCanonicalSessionId(sessionId, 'createSessionWs');
  return new WebSocketManager(`${WS_BASE}/ws/agents/${sessionId}`, { sessionId });
}

// One backgrounded session socket, kept alive across a hop so an active agent's stream doesn't pay a reconnect+resume handshake on reopen (the "Locking-in" lag). Bounded to ONE: opening any other chat tears the previous one down, so at most a single detached socket lingers, still pumping events into Redux.
let _backgroundedSessionWs: { sessionId: string; ws: WebSocketManager } | null = null;

export function acquireSessionWs(sessionId: string): WebSocketManager {
  if (_backgroundedSessionWs?.sessionId === sessionId) {
    const ws = _backgroundedSessionWs.ws;
    _backgroundedSessionWs = null;
    return ws;
  }
  warnIfNotCanonicalSessionId(sessionId, 'acquireSessionWs');
  return new WebSocketManager(`${WS_BASE}/ws/agents/${sessionId}`, { sessionId });
}

export function releaseSessionWs(sessionId: string, ws: WebSocketManager, keepAlive: boolean): void {
  // Never keep more than one detached socket around.
  if (_backgroundedSessionWs && _backgroundedSessionWs.sessionId !== sessionId) {
    _backgroundedSessionWs.ws.disconnect();
    _backgroundedSessionWs = null;
  }
  if (keepAlive) {
    _backgroundedSessionWs = { sessionId, ws };
  } else {
    ws.disconnect();
    if (_backgroundedSessionWs?.sessionId === sessionId) _backgroundedSessionWs = null;
  }
}

export default WebSocketManager;
