oikos/web/src/lib/stores/chat.ts

import { writable, get, type Writable } from 'svelte/store'
import {
  streamChat,
  fetchSessions,
  fetchMessages,
  fetchMessagesOrNotFound,
  deleteSession as apiDeleteSession
} from '$lib/api'
import type { ChatEvent, Session, Message } from '$lib/api'
import type { ToolCallResult } from '$lib/types'
import { liveEvents, subscribeEvents } from './events'

export type { ToolCallResult }

export interface PendingApproval {
  executionId: string
  action: string
  target: string
  destructive: boolean
  command?: string
  purpose?: string
}

export interface ChatMessage {
  id: string
  role: 'user' | 'assistant'
  text: string
  thinking?: string
  tools: ToolCallResult[]
  pendingApprovals: PendingApproval[]
  created_at?: string
}

const APPROVAL_RE = /execution\s+([0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12})/i

// Deliberately NOT filtered by tool name. There is no fixed set of gated
// tools — `run` can execute anything, and any future tool that queues an
// approval should surface a card the same way. A prior version hardcoded
// `t.name === 'request_execution'`, so approvals raised by the newer `run`
// tool were silently invisible in chat: no card, no feedback, nothing to
// self-heal, forcing the operator to the Ops page with zero acknowledgement
// back in the conversation. Matching on the response shape (not the tool
// name) is what makes this robust to new gated tools without another
// silent breakage.
function extractApprovals(tools: ToolCallResult[]): PendingApproval[] {
  const out: PendingApproval[] = []
  for (const t of tools) {
    if (t.type !== 'tool_result') continue
    const text = typeof t.result === 'string' ? t.result : JSON.stringify(t.result ?? '')
    if (!text.includes('requires approval')) continue
    const m = text.match(APPROVAL_RE)
    if (m) {
      const args = t.args ?? {}
      const purpose = typeof args.purpose === 'string' ? args.purpose : undefined
      out.push({
        executionId: m[1],
        action: purpose
          ? purpose.slice(0, 60)
          : typeof args.action === 'string'
            ? args.action
            : t.name,
        target: typeof args.target === 'string' ? args.target : 'unknown',
        destructive: /\bDESTRUCTIVE\b/.test(text),
        command: typeof args.command === 'string' ? args.command : undefined,
        purpose
      })
    }
  }
  return out
}

function mid(): string {
  return crypto.randomUUID()
}

// dropOptimisticAssistantBubble removes the trailing empty assistant message
// that sendSessionMessage/startTask optimistically append — used when a turn is
// QUEUED behind an in-flight one (plan 2026-08-03 F2): no live assistant stream
// is attached, so the empty placeholder must go (otherwise it lingers as a
// blank bubble). Shared so the guard can't drift between the two call sites.
function dropOptimisticAssistantBubble(messages: Writable<ChatMessage[]>): void {
  messages.update((ms) => {
    const last = ms[ms.length - 1]
    if (last && last.role === 'assistant' && last.text === '' && last.tools.length === 0) {
      return ms.slice(0, -1)
    }
    return ms
  })
}

export const messages = writable<ChatMessage[]>([])
export const streaming = writable(false)
export const connectionState = writable<'connected' | 'disconnected' | 'reconnecting'>('connected')
export const currentSession = writable<string | null>(null)
export const sessions = writable<Session[]>([])
export const sessionMessages = writable<Message[]>([])
export const error = writable<string | null>(null)
export const chatErrors = writable<{ id: string; message: string; action?: string; tag?: string }[]>([])

export function dismissError(id: string) {
  chatErrors.update((e) => e.filter((x) => x.id !== id))
}

export function addChatError(message: string, action?: string, tag?: string) {
  chatErrors.update((e) => [...e, { id: crypto.randomUUID(), message, action, tag }])
}

// Session statuses where no turn is running — the agent reached a terminal
// state (done/failed/abandoned) or paused for operator input (awaiting_input).
// A task.status event landing in one of these is an authoritative "the turn
// ended" signal, used by clearTurnState (F3) to unstick a chat view that lost
// its SSE stream mid-turn.
const TURN_ENDED_STATUS = new Set(['done', 'failed', 'abandoned', 'awaiting_input'])

// humanizeChatError turns raw transport/SDK error strings into operator-
// readable, non-alarming copy. The raw forms ("llm: error in input stream:
// …", "Failed to fetch", "HTTP 502") read as catastrophic and unactionable;
// most are transient model-connection drops where the task itself is fine.
// Used for both the inline error box (LLM error events) and the connection
// toast (F2).
function humanizeChatError(raw: string): string {
  const s = raw.toLowerCase()
  if (
    s.includes('input stream') ||
    s.includes('llm:') ||
    s.includes('failed to fetch') ||
    s.includes('network') ||
    s.includes('econnreset') ||
    s.includes('timeout') ||
    /http 5\d\d/.test(s)
  ) {
    return 'The model connection dropped. The task keeps running in the background — it will catch up here automatically.'
  }
  if (/http 401|http 403|unauthor|forbidden/.test(s)) {
    return 'Your session expired. Reconnect to continue.'
  }
  return raw
}

// clearTurnState resets a session's chat view to a clean "connected, idle"
// state — the recovery action when a dropped SSE left it stuck showing
// streaming/disconnected after the turn had already ended. Clears the window
// bundle, the global bundle (if that's the viewed session), and any
// connection-lost toasts tagged 'connection' (F2/F3).
function clearTurnState(sessionId: string) {
  const win = sessionChats.get(sessionId)
  if (win) {
    win.streaming.set(false)
    win.connectionState.set('connected')
  }
  if (get(currentSession) === sessionId) {
    streaming.set(false)
    connectionState.set('connected')
  }
  chatErrors.update((errs) => errs.filter((e) => e.tag !== 'connection'))
}

// One app-lifetime subscription to the always-on event stream: a terminal
// task.status for a session we have open is the authoritative end-of-turn
// signal, and recovers a chat view whose SSE dropped without a 'done' event
// (the "task never ended" symptom). Ref-counted by subscribeEvents, so this
// shares the single connection the rest of the app already keeps open.
//
// Lazily armed from chatFor() (P2.2) rather than at module import, so
// importing this module — e.g. in a test — doesn't open an SSE connection as
// an import side-effect.
let chatEventSyncArmed = false
function ensureChatEventSync() {
  if (chatEventSyncArmed) return
  chatEventSyncArmed = true
  subscribeEvents()
  liveEvents.subscribe((events) => {
    const ev = events[0]
    if (!ev || ev.type !== 'task.status') return
    const sid = ev.correlation_id
    if (!sid) return
    const status = (ev.data as { status?: string } | null)?.status
    if (typeof status === 'string' && TURN_ENDED_STATUS.has(status)) {
      clearTurnState(sid)
    }
  })
}

// Per-session controller tracking. Multiple tasks can stream concurrently
// (see sendMessage's session guard above this used to be a single global
// `activeController`, which meant cancelStream()/newChat() always aborted
// whichever stream happened to be the MOST RECENTLY started one, regardless
// of what the operator was currently viewing — starting Task A, switching to
// (already-loaded) Task B, then clicking "New task" would silently abort
// Task A's still-running turn even though the operator was never looking at
// it and never asked to cancel it. Keyed by session id once known;
// pendingController covers the brief window for a brand-new task between
// streamChat() starting and its 'session' event assigning a real id.
const activeControllers = new Map<string, AbortController>()
let pendingController: AbortController | null = null

export async function loadSessions() {
  const list = await fetchSessions()
  sessions.set(list)
}

// mergeToolCalls collapses a persisted tool_calls array into one entry per
// call id. Nomos persists the tool_use and tool_result as two separate
// entries sharing the same id (matching the SSE event pair); Chat.svelte
// renders tools in a keyed {#each ... (tool.id)}, which throws on duplicate
// keys and silently aborts the whole message list. Live-streamed messages
// never hit this because sendMessage() merges tool_result into the existing
// tool_use entry in place rather than appending a second one.
function mergeToolCalls(raw: ToolCallResult[] | undefined): ToolCallResult[] {
  const byId = new Map<string, ToolCallResult>()
  for (const tc of raw ?? []) {
    const key = tc.id ?? crypto.randomUUID()
    const existing = byId.get(key)
    byId.set(key, existing ? { ...existing, ...tc, id: key } : { ...tc, id: key })
  }
  return Array.from(byId.values())
}

function toChatMessages(msgs: Message[]): ChatMessage[] {
  return msgs.map((m) => {
    const content = typeof m.content === 'string' ? { text: m.content } : m.content
    const tools = mergeToolCalls(content?.tool_calls)
    return {
      id: m.id,
      role: m.role as 'user' | 'assistant',
      text: content?.text ?? '',
      thinking: content?.thinking ?? undefined,
      tools,
      pendingApprovals: extractApprovals(tools),
      created_at: m.created_at
    }
  })
}

function chatMessagesChanged(a: ChatMessage[], b: ChatMessage[]): boolean {
  if (a.length !== b.length) return true
  for (let i = 0; i < a.length; i++) {
    if (a[i].id !== b[i].id || a[i].role !== b[i].role || a[i].text !== b[i].text || a[i].tools.length !== b[i].tools.length) {
      return true
    }
  }
  return false
}

export async function loadSessionMessages(sessionId: string) {
  currentSession.set(sessionId)
  // This is a fresh view of sessionId's current (REST-loaded) state — reset
  // streaming regardless of whether some OTHER task's stream happens to still
  // be in flight in the background. Without this, switching to a task while
  // a different one is mid-turn could leave `streaming` stuck true here (that
  // other stream's completion callback now correctly skips touching it, per
  // sendMessage's session guard) — which would disable the input AND silently
  // stop startPolling's loop from ever applying updates (it bails while
  // $streaming is true), making the newly-opened task look frozen.
  streaming.set(false)
  const msgs = await fetchMessages(sessionId)
  sessionMessages.set(msgs)
  messages.set(toChatMessages(msgs))
  startPolling(sessionId)
}

// Live visibility for autonomous work: the auto-continuation worker (see
// cmd/nomos/continue.go) runs entirely server-side and has no live push —
// previously the only way to see its result was to manually reload the
// session, so approving a plan and then waiting felt like nothing was
// happening even while the agent was actively working. This polls the
// session's persisted messages every few seconds and merges in anything new
// (an auto-continuation's result, a fresh pending approval it queued, etc.)
// so the transcript updates on its own. Only runs between turns — never
// while a live streaming turn owns the message list, to avoid clobbering the
// in-progress optimistic UI.
let pollTimer: ReturnType<typeof setInterval> | null = null
let pollingSessionId: string | null = null

function startPolling(sessionId: string) {
  stopPolling()
  pollingSessionId = sessionId
  pollTimer = setInterval(async () => {
    // Allow polling while disconnected — the agent is still working
    // server-side and the poller is the only way to see it.
    if (get(streaming) && get(connectionState) === 'connected') return
    if (pollingSessionId !== sessionId || get(currentSession) !== sessionId) return
    const msgs = await fetchMessages(sessionId)
    if (get(streaming) || pollingSessionId !== sessionId) return // re-check: the fetch itself takes time
      const incoming = toChatMessages(msgs)
      if (chatMessagesChanged(get(messages), incoming)) {
        sessionMessages.set(msgs)
        messages.set(incoming)
      }
  }, 3000)
}

export function stopPolling() {
  if (pollTimer) {
    clearInterval(pollTimer)
    pollTimer = null
  }
  pollingSessionId = null
}

export function sendMessage(text: string) {
  error.set(null)
  streaming.set(true)

  const userMsg: ChatMessage = {
    id: mid(),
    role: 'user',
    text,
    tools: [],
    pendingApprovals: []
  }
  messages.update((ms) => [...ms, userMsg])

  const assistantMsg: ChatMessage = {
    id: mid(),
    role: 'assistant',
    text: '',
    tools: [],
    pendingApprovals: []
  }
  messages.update((ms) => [...ms, assistantMsg])

  const activeTools: Map<string, ToolCallResult> = new Map()

  // Multiple tasks can stream concurrently (the backend runs each turn as its
  // own goroutine — nothing serializes them), but `messages`/`currentSession`
  // are a single global view. Without this guard, switching to a different
  // task while this stream is still open lets its later events (tool_use,
  // text_delta, ..., and worst of all the 'done' handler's
  // currentSession.set) get applied to whatever the operator is NOW looking
  // at — silently corrupting another task's transcript, or yanking the view
  // back to this one. openedFor is the session this call started for (null
  // for a brand-new task, until the 'session' event assigns the real id);
  // every branch below checks the CURRENT $currentSession still matches
  // before touching `messages`. The task itself keeps running server-side
  // regardless — dropped events just mean the live view isn't watching it;
  // navigating back re-hydrates via REST/poll same as it already does for
  // auto-continuation.
  const openedFor = get(currentSession)
  let streamSessionID = openedFor
  let receivedDone = false

  const controller = streamChat(
    text,
    get(currentSession), // continue the active session so the agent keeps context
    (ev: ChatEvent) => {
      if (ev.type === 'session') {
        streamSessionID = ev.data
        // Move this stream's controller into the per-session map now that its
        // real id is known, so a later cancelStream()/newChat() from THIS
        // session's view can find and abort it — and, just as importantly,
        // so cancelling/leaving a DIFFERENT session never reaches this one.
        // For a continued (non-new) session, openedFor already equals ev.data
        // and the controller was stored under that key at creation below;
        // this only does real work for a brand-new task's first assignment.
        if (pendingController === controller) pendingController = null
        activeControllers.set(ev.data, controller)
        // Only claim currentSession if the operator hasn't already navigated
        // to something else since this call started (openedFor covers both
        // "still on the task I was on" and "still hadn't opened one yet").
        if (get(currentSession) === openedFor) currentSession.set(ev.data)
        return
      }
      if (get(currentSession) !== streamSessionID) return // stream's task isn't the one on screen — drop
      if (ev.type === 'tool_use') {
        const tr: ToolCallResult = {
          type: 'tool_use',
          name: ev.data.name,
          id: ev.data.id,
          args: ev.data.args
        }
        activeTools.set(ev.data.id, tr)
        messages.update((ms) => {
          const last = ms[ms.length - 1]
          if (last && last.role === 'assistant') {
            ms[ms.length - 1] = { ...last, tools: [...last.tools, tr] }
          }
          return [...ms]
        })
      } else if (ev.type === 'tool_result') {
        const existing = activeTools.get(ev.data.id)
        if (existing) {
          const updated: ToolCallResult = {
            ...existing,
            type: 'tool_result',
            result: ev.data.result,
            error: ev.data.error
          }
          activeTools.set(ev.data.id, updated)
          messages.update((ms) => {
            const last = ms[ms.length - 1]
            if (last && last.role === 'assistant') {
              const tools = last.tools.map((t) => (t.id === ev.data.id ? updated : t))
              ms[ms.length - 1] = { ...last, tools, pendingApprovals: extractApprovals(tools) }
            }
            return [...ms]
          })
        }
      } else if (ev.type === 'text_delta') {
        messages.update((ms) => {
          const last = ms[ms.length - 1]
          if (last && last.role === 'assistant') {
            ms[ms.length - 1] = { ...last, text: last.text + ev.data }
          }
          return [...ms]
        })
      } else if (ev.type === 'text') {
        // Final authoritative content for the turn; replaces accumulated deltas.
        messages.update((ms) => {
          const last = ms[ms.length - 1]
          if (last && last.role === 'assistant') {
            if (ev.is_thinking) {
              ms[ms.length - 1] = {
                ...last,
                thinking: (last.thinking || '') + ev.data,
                text: ''
              }
            } else {
              ms[ms.length - 1] = { ...last, text: ev.data }
            }
          }
          return [...ms]
        })
      } else if (ev.type === 'done') {
        receivedDone = true
        connectionState.set('connected')
        messages.update((ms) => {
          const last = ms[ms.length - 1]
          if (last && last.role === 'assistant') {
            ms[ms.length - 1] = { ...last, pendingApprovals: extractApprovals(last.tools) }
          }
          return [...ms]
        })
        const sid = ev.data?.session_id ?? ev.session_id
        // Start polling for auto-continuation results now that the live turn
        // is over — this is what makes an approved plan's later steps show up
        // on their own instead of requiring a manual reload. (startPolling's
        // own loop already re-checks $currentSession before applying results,
        // so this is safe to call even if the operator has since navigated
        // elsewhere — it just won't visibly do anything until/unless they
        // come back.)
        if (sid) startPolling(sid)
      } else if (ev.type === 'error') {
        error.set(ev.data)
      }
    },
    (err: string) => {
      // Distinguish user abort from network drop.
      if (err === 'AbortError' || err.includes('aborted')) {
        if (get(currentSession) === streamSessionID) streaming.set(false)
        return
      }
      // Network blip / server restart — initiate reconnect.
      if (get(currentSession) === streamSessionID) {
        error.set(err)
        if (!receivedDone && streamSessionID) {
          handleDisconnect(streamSessionID)
        } else {
          streaming.set(false)
        }
      }
    },
    () => {
      // SSE stream completed without error. If we never received 'done',
      // the connection was severed mid-turn — treat as disconnect.
      if (!receivedDone && streamSessionID && get(currentSession) === streamSessionID) {
        handleDisconnect(streamSessionID)
      } else if (get(currentSession) === streamSessionID) {
        streaming.set(false)
      }
      if (streamSessionID && activeControllers.get(streamSessionID) === controller) {
        activeControllers.delete(streamSessionID)
      }
      if (pendingController === controller) pendingController = null
      loadSessions()
    }
  )

  // Register immediately (not just inside the 'session' handler above) so a
  // cancelStream() during the brief pre-'session' window for a CONTINUED
  // session (openedFor already known) can find it right away.
  if (openedFor) {
    activeControllers.set(openedFor, controller)
  } else {
    pendingController = controller
  }
}

// handleDisconnect is called when the global SSE stream drops mid-turn
// without a 'done' event. NOTE: the global single-session chat action path
// (sendMessage → this) is not currently wired to any UI — only the
// per-session window path (sendSessionMessage/startTask) is live, which has
// its own inline equivalent. This is kept safe and turn-free in case the
// global path is re-wired: it falls back to polling and surfaces one
// connection toast; recovery is driven by the poller + the terminal
// task.status subscription (clearTurnState), NEVER by POSTing an empty
// message that would spawn a duplicate background turn (F1/F2).
function handleDisconnect(sessionId: string) {
  connectionState.set('disconnected')
  startPolling(sessionId)
  addChatError(
    'Connection to the agent dropped. The task keeps running — it will catch up here automatically.',
    'Dismiss',
    'connection'
  )
}

// Manual reconnect for the global view (currently unused — windows use
// loadSessionChat via their onReconnect). Re-fetches the transcript and
// resets state; does NOT start a new turn.
export function reconnect() {
  const sid = get(currentSession)
  if (!sid) return
  streaming.set(false)
  connectionState.set('connected')
  chatErrors.update((errs) => errs.filter((e) => e.tag !== 'connection'))
  loadSessionMessages(sid)
}

export function newChat() {
  cancelStream()
  stopPolling()
  connectionState.set('connected')
  currentSession.set(null)
  messages.set([])
  error.set(null)
  chatErrors.set([])
  streaming.set(false) // fresh view — see loadSessionMessages for why this must not depend on cancelStream's own reset
}

// Cancels the stream for whatever the operator is CURRENTLY VIEWING — never
// some other, unrelated task's background stream. Before per-session
// tracking, this aborted a single global `activeController`, which meant it
// always targeted the MOST RECENTLY STARTED stream regardless of what was on
// screen: start Task A, switch to already-loaded Task B, click "New task" —
// newChat()'s cancelStream() would silently abort Task A's still-running
// turn, even though the operator was never looking at it and never asked to
// cancel it. Now it looks up by $currentSession (or pendingController for
// the brief pre-'session'-event window of a just-started new task) so it can
// only ever touch the stream that belongs to the view being left.
export function cancelStream() {
  const sid = get(currentSession)
  const controller = sid ? activeControllers.get(sid) : pendingController
  if (!controller) return
  controller.abort()
  if (sid) activeControllers.delete(sid)
  if (pendingController === controller) pendingController = null
  streaming.set(false)
}

export async function deleteSession(sessionId: string) {
  const ok = await apiDeleteSession(sessionId)
  if (!ok) return
  if (get(currentSession) === sessionId) {
    newChat()
  }
  loadSessions()
}

// ─── per-session chat state, for floating task windows ─────────────────────
//
// Everything above this point is the single "whatever's on screen" view used
// by the main Chat page and the chat drawer — one global `currentSession`,
// one `messages` array, guarded so a background stream never clobbers the
// view. Floating task windows break that assumption: several sessions can be
// open and legitimately streaming at once, each wanting its own live
// transcript. Rather than retrofit the guard-heavy logic above (streamed
// events checking `get(currentSession) === streamSessionID` before applying),
// each window gets its own isolated store bundle keyed by session id, so
// there's nothing to guard — events for session X always land in X's own
// bundle regardless of what else is open or on screen.
export interface SessionChatState {
  messages: Writable<ChatMessage[]>
  streaming: Writable<boolean>
  connectionState: Writable<'connected' | 'disconnected' | 'reconnecting'>
  error: Writable<string | null>
  // Set by loadSessionChat when the backend 404s the session outright
  // (deleted, or an id that was never valid — a stale persisted window, a
  // bad deep link). Distinct from a merely-empty transcript, which is the
  // normal state for a session that exists but hasn't sent a message yet.
  notFound: Writable<boolean>
}

const sessionChats = new Map<string, SessionChatState>()
const sessionPollers = new Map<string, ReturnType<typeof setInterval>>()

// Lazily creates (and memoizes) the store bundle for a session — call this to
// get the stores to subscribe to; it does not fetch anything.
export function chatFor(sessionId: string): SessionChatState {
  ensureChatEventSync() // arm the terminal task.status → clearTurnState recovery (P2.2)
  let c = sessionChats.get(sessionId)
  if (!c) {
    c = {
      messages: writable([]),
      streaming: writable(false),
      connectionState: writable('connected'),
      error: writable(null),
      notFound: writable(false)
    }
    sessionChats.set(sessionId, c)
  }
  return c
}

function startSessionPolling(sessionId: string) {
  const existing = sessionPollers.get(sessionId)
  if (existing) clearInterval(existing)
  const chat = chatFor(sessionId)
  sessionPollers.set(
    sessionId,
    setInterval(async () => {
      if (get(chat.streaming) && get(chat.connectionState) === 'connected') return
      const msgs = await fetchMessages(sessionId)
      if (get(chat.streaming)) return // re-check: the fetch itself takes time
      const incoming = toChatMessages(msgs)
      if (chatMessagesChanged(get(chat.messages), incoming)) {
        chat.messages.set(incoming)
      }
      // F3 safety net: if we're recovering from a dropped SSE but the
      // session's task has already reached a turn-ended status, clear the
      // stuck disconnected/streaming flags. Catches the edge where the
      // terminal task.status event fired during the brief disconnect window.
      if (get(chat.connectionState) !== 'connected') {
        const s = get(sessions).find((x) => x.id === sessionId)
        if (s?.status && TURN_ENDED_STATUS.has(s.status)) {
          clearTurnState(sessionId)
        }
      }
    }, 3000)
  )
}

export function stopSessionPolling(sessionId: string) {
  const t = sessionPollers.get(sessionId)
  if (t) {
    clearInterval(t)
    sessionPollers.delete(sessionId)
  }
}

// Fetches sessionId's current transcript into its own store bundle and
// starts polling it for auto-continuation updates — the per-session
// equivalent of loadSessionMessages, for a window rather than the main view.
export async function loadSessionChat(sessionId: string): Promise<void> {
  const chat = chatFor(sessionId)
  // A fresh (re)load is a clean view: not streaming, connected, no stale
  // error. This also serves the window's manual "Reconnect" button —
  // re-fetching the transcript and resetting state, never spawning a new
  // turn (the old reconnect path POSTed an empty message that started a
  // duplicate background turn; F1/F2 removed that).
  chat.streaming.set(false)
  chat.connectionState.set('connected')
  chat.error.set(null)
  chatErrors.update((errs) => errs.filter((e) => e.tag !== 'connection'))
  const msgs = await fetchMessagesOrNotFound(sessionId)
  if (msgs === null) {
    chat.notFound.set(true)
    return // nothing to poll — the session doesn't exist
  }
  chat.messages.set(toChatMessages(msgs))
  startSessionPolling(sessionId)
}

// Per-session equivalent of sendMessage — writes into sessionId's own store
// bundle unconditionally (no "is this still on screen" guard needed, since
// the bundle IS the screen for this session's window) and shares
// `activeControllers` with the singleton path above so cancelStream() from
// either a window or the main view (if the same session happens to be open
// in both) finds the same in-flight call.
export function sendSessionMessage(sessionId: string, text: string) {
  const chat = chatFor(sessionId)
  chat.error.set(null)
  chat.streaming.set(true)

  const userMsg: ChatMessage = { id: mid(), role: 'user', text, tools: [], pendingApprovals: [] }
  chat.messages.update((ms) => [...ms, userMsg])
  const assistantMsg: ChatMessage = {
    id: mid(),
    role: 'assistant',
    text: '',
    tools: [],
    pendingApprovals: []
  }
  chat.messages.update((ms) => [...ms, assistantMsg])

  const activeTools: Map<string, ToolCallResult> = new Map()
  let receivedDone = false

  const controller = streamChat(
    text,
    sessionId,
    (ev: ChatEvent) => {
      if (ev.type === 'session') return // sessionId is already known for a window
      if (ev.type === 'queued') {
        // This message was queued behind an in-flight turn (plan 2026-08-03
        // F2): no assistant stream is attached to this response. Drop the
        // optimistic empty assistant bubble so the user message is the last
        // thing on screen — the thread then shows a "Queued" hint while the
        // session is working, and the poller surfaces the queued turn's
        // result once it runs server-side.
        dropOptimisticAssistantBubble(chat.messages)
        return
      }
      if (ev.type === 'tool_use') {
        const tr: ToolCallResult = {
          type: 'tool_use',
          name: ev.data.name,
          id: ev.data.id,
          args: ev.data.args
        }
        activeTools.set(ev.data.id, tr)
        chat.messages.update((ms) => {
          const last = ms[ms.length - 1]
          if (last && last.role === 'assistant') {
            ms[ms.length - 1] = { ...last, tools: [...last.tools, tr] }
          }
          return [...ms]
        })
      } else if (ev.type === 'tool_result') {
        const existing = activeTools.get(ev.data.id)
        if (existing) {
          const updated: ToolCallResult = {
            ...existing,
            type: 'tool_result',
            result: ev.data.result,
            error: ev.data.error
          }
          activeTools.set(ev.data.id, updated)
          chat.messages.update((ms) => {
            const last = ms[ms.length - 1]
            if (last && last.role === 'assistant') {
              const tools = last.tools.map((t) => (t.id === ev.data.id ? updated : t))
              ms[ms.length - 1] = { ...last, tools, pendingApprovals: extractApprovals(tools) }
            }
            return [...ms]
          })
        }
      } else if (ev.type === 'text_delta') {
        chat.messages.update((ms) => {
          const last = ms[ms.length - 1]
          if (last && last.role === 'assistant') {
            ms[ms.length - 1] = { ...last, text: last.text + ev.data }
          }
          return [...ms]
        })
      } else if (ev.type === 'text') {
        chat.messages.update((ms) => {
          const last = ms[ms.length - 1]
          if (last && last.role === 'assistant') {
            if (ev.is_thinking) {
              ms[ms.length - 1] = {
                ...last,
                thinking: (last.thinking || '') + ev.data,
                text: ''
              }
            } else {
              ms[ms.length - 1] = { ...last, text: ev.data }
            }
          }
          return [...ms]
        })
      } else if (ev.type === 'done') {
        receivedDone = true
        chat.connectionState.set('connected')
        chat.messages.update((ms) => {
          const last = ms[ms.length - 1]
          if (last && last.role === 'assistant') {
            ms[ms.length - 1] = { ...last, pendingApprovals: extractApprovals(last.tools) }
          }
          return [...ms]
        })
        startSessionPolling(sessionId)
      } else if (ev.type === 'error') {
        chat.error.set(humanizeChatError(ev.data))
      }
    },
    (err: string) => {
      if (err === 'AbortError' || err.includes('aborted')) {
        chat.streaming.set(false)
        return
      }
      // Network drop (no 'done' received): show ONE connection-lost surface
      // and recover via the poller + terminal task.status event (F2/F3).
      // Don't also set chat.error — the banner+toast convey it, and a raw
      // "Failed to fetch" alongside would just be noise.
      if (!receivedDone) {
        chat.connectionState.set('disconnected')
        startSessionPolling(sessionId)
        addChatError(
          'Connection to the agent dropped. The task keeps running — it will catch up here automatically.',
          'Dismiss',
          'connection'
        )
      } else {
        // Stream ended cleanly but fetch reported an error tail — surface it.
        chat.error.set(humanizeChatError(err))
        chat.streaming.set(false)
      }
    },
    () => {
      chat.streaming.set(false)
      if (activeControllers.get(sessionId) === controller) activeControllers.delete(sessionId)
      loadSessions()
    }
  )
  activeControllers.set(sessionId, controller)
}

export function cancelSessionStream(sessionId: string) {
  const controller = activeControllers.get(sessionId)
  if (!controller) return
  controller.abort()
  activeControllers.delete(sessionId)
  chatFor(sessionId).streaming.set(false)
}

// ─── new-task launcher (desktop center input / Tasks app) ───────────────────
//
// Starting a brand-new task has no session id to hang a window off of until
// the stream's own 'session' event assigns one (see the 'session' branch in
// sendMessage above) — the desktop launcher needs to open that task's window
// the moment an id exists, not before. startTask begins the stream
// immediately, buffers any events that arrive before 'session' (defensive:
// in practice 'session' always arrives first), then seeds that session's own
// chatFor() bundle exactly like sendSessionMessage does and hands the id back
// via onSession so the caller can open its window. From that point on the
// window behaves exactly like any other task window.
export function startTask(text: string, onSession: (sessionId: string) => void): void {
  const userMsg: ChatMessage = { id: mid(), role: 'user', text, tools: [], pendingApprovals: [] }
  const assistantMsg: ChatMessage = {
    id: mid(),
    role: 'assistant',
    text: '',
    tools: [],
    pendingApprovals: []
  }
  const activeTools: Map<string, ToolCallResult> = new Map()
  let receivedDone = false
  let sessionId: string | null = null
  let chat: SessionChatState | null = null
  const buffered: ChatEvent[] = []

  function apply(ev: ChatEvent) {
    const c = chat
    if (!c || !sessionId) return
    if (ev.type === 'queued') {
      // Defensive: a brand-new task won't normally queue (its session has no
      // in-flight turn), but handle it symmetrically with sendSessionMessage —
      // drop the optimistic empty assistant bubble. See plan 2026-08-03 F2.
      dropOptimisticAssistantBubble(c.messages)
      return
    }
    if (ev.type === 'tool_use') {
      const tr: ToolCallResult = {
        type: 'tool_use',
        name: ev.data.name,
        id: ev.data.id,
        args: ev.data.args
      }
      activeTools.set(ev.data.id, tr)
      c.messages.update((ms) => {
        const last = ms[ms.length - 1]
        if (last && last.role === 'assistant') {
          ms[ms.length - 1] = { ...last, tools: [...last.tools, tr] }
        }
        return [...ms]
      })
    } else if (ev.type === 'tool_result') {
      const existing = activeTools.get(ev.data.id)
      if (existing) {
        const updated: ToolCallResult = {
          ...existing,
          type: 'tool_result',
          result: ev.data.result,
          error: ev.data.error
        }
        activeTools.set(ev.data.id, updated)
        c.messages.update((ms) => {
          const last = ms[ms.length - 1]
          if (last && last.role === 'assistant') {
            const tools = last.tools.map((t) => (t.id === ev.data.id ? updated : t))
            ms[ms.length - 1] = { ...last, tools, pendingApprovals: extractApprovals(tools) }
          }
          return [...ms]
        })
      }
    } else if (ev.type === 'text_delta') {
      c.messages.update((ms) => {
        const last = ms[ms.length - 1]
        if (last && last.role === 'assistant') {
          ms[ms.length - 1] = { ...last, text: last.text + ev.data }
        }
        return [...ms]
      })
    } else if (ev.type === 'text') {
      c.messages.update((ms) => {
        const last = ms[ms.length - 1]
        if (last && last.role === 'assistant') {
          if (ev.is_thinking) {
            ms[ms.length - 1] = {
              ...last,
              thinking: (last.thinking || '') + ev.data,
              text: ''
            }
          } else {
            ms[ms.length - 1] = { ...last, text: ev.data }
          }
        }
        return [...ms]
      })
    } else if (ev.type === 'done') {
      receivedDone = true
      c.connectionState.set('connected')
      c.messages.update((ms) => {
        const last = ms[ms.length - 1]
        if (last && last.role === 'assistant') {
          ms[ms.length - 1] = { ...last, pendingApprovals: extractApprovals(last.tools) }
        }
        return [...ms]
      })
      startSessionPolling(sessionId)
    } else if (ev.type === 'error') {
      c.error.set(humanizeChatError(ev.data))
    }
  }

  const controller = streamChat(
    text,
    null,
    (ev: ChatEvent) => {
      if (ev.type === 'session') {
        sessionId = ev.data
        activeControllers.set(sessionId, controller)
        chat = chatFor(sessionId)
        chat.streaming.set(true)
        chat.messages.update((ms) => [...ms, userMsg, assistantMsg])
        onSession(sessionId)
        for (const b of buffered.splice(0)) apply(b)
        return
      }
      if (!chat) {
        buffered.push(ev)
        return
      }
      apply(ev)
    },
    (err: string) => {
      if (!chat) return // never got a session id — nothing to show the error in
      if (err === 'AbortError' || err.includes('aborted')) {
        chat.streaming.set(false)
        return
      }
      if (!receivedDone && sessionId) {
        chat.connectionState.set('disconnected')
        startSessionPolling(sessionId)
        addChatError(
          'Connection to the agent dropped. The task keeps running — it will catch up here automatically.',
          'Dismiss',
          'connection'
        )
      } else {
        chat.error.set(humanizeChatError(err))
        chat.streaming.set(false)
      }
    },
    () => {
      if (chat) chat.streaming.set(false)
      if (sessionId && activeControllers.get(sessionId) === controller)
        activeControllers.delete(sessionId)
      loadSessions()
    }
  )
}