dtoro
b414722fc7
- New ActivityTimeline: unified timeline in sidebar showing all agent actions (goal, plan steps, tool calls, knowledge, completion) in reverse chron order - activityLog derived store merges messages + planSteps + currentTask - AgentIndicator stays in chat (thinking/working indicator), simplified props - ToolCallGroup removed from chat — tools visible only in sidebar timeline - SessionDigest replaced by ActivityTimeline - PlanProgress restored in sidebar (conceptual steps, separate from timeline)
631 lines
24 KiB
TypeScript
631 lines
24 KiB
TypeScript
import { writable, derived, get } from 'svelte/store'
|
|
import { streamChat, fetchSessions, fetchMessages, deleteSession as apiDeleteSession } from '$lib/api'
|
|
import { planSteps, currentTask } from '$lib/stores/workspace'
|
|
import type { ChatEvent, Session, Message } from '$lib/api'
|
|
|
|
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
|
|
tools: ToolCallResult[]
|
|
pendingApprovals: PendingApproval[]
|
|
}
|
|
|
|
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) {
|
|
out.push({
|
|
executionId: m[1],
|
|
action: t.args?.purpose?.slice(0, 60) ?? t.args?.action ?? t.name ?? 'unknown',
|
|
target: t.args?.target ?? 'unknown',
|
|
destructive: /\bDESTRUCTIVE\b/.test(text),
|
|
command: t.args?.command,
|
|
purpose: t.args?.purpose
|
|
})
|
|
}
|
|
}
|
|
return out
|
|
}
|
|
|
|
export interface ToolCallResult {
|
|
type: 'tool_use' | 'tool_result'
|
|
name: string
|
|
id?: string
|
|
args?: any
|
|
result?: any
|
|
error?: string
|
|
}
|
|
|
|
function mid(): string {
|
|
return crypto.randomUUID()
|
|
}
|
|
|
|
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 }[]>([])
|
|
|
|
export function dismissError(id: string) {
|
|
chatErrors.update((e) => e.filter((x) => x.id !== id))
|
|
}
|
|
|
|
export function addChatError(message: string, action?: string) {
|
|
chatErrors.update((e) => [...e, { id: crypto.randomUUID(), message, action }])
|
|
}
|
|
|
|
// ── Activity timeline ────────────────────────────────────────────────
|
|
|
|
export interface ActivityEntry {
|
|
id: string
|
|
type: 'goal' | 'plan' | 'step_running' | 'step_done' | 'step_failed' |
|
|
'tool_running' | 'tool_done' | 'tool_error' |
|
|
'knowledge' | 'complete' | 'question' | 'error'
|
|
description: string
|
|
detail?: string
|
|
timestamp: number
|
|
toolName?: string
|
|
status: 'running' | 'done' | 'failed'
|
|
}
|
|
|
|
export const activityLog = derived([messages, planSteps, currentTask], ([$msgs, $steps, $task]) => {
|
|
const entries: ActivityEntry[] = []
|
|
const now = Date.now()
|
|
|
|
// Goal
|
|
if ($task?.goal) {
|
|
entries.push({ id: 'goal', type: 'goal', description: $task.goal, timestamp: 0, status: 'done' })
|
|
}
|
|
|
|
// Plan steps
|
|
for (const s of $steps) {
|
|
if (s.status === 'pending') continue
|
|
const stepLabel = s.title || `Step ${s.seq}`
|
|
entries.push({
|
|
id: s.id,
|
|
type: s.status === 'running' ? 'step_running' : s.status === 'done' ? 'step_done' : 'step_failed',
|
|
description: `Step ${s.seq}: ${stepLabel}`,
|
|
detail: s.detail || undefined,
|
|
timestamp: s.started_at ? new Date(s.started_at).getTime() : now,
|
|
status: s.status === 'running' ? 'running' : s.status === 'done' ? 'done' : 'failed'
|
|
})
|
|
}
|
|
|
|
// Tool calls (from messages)
|
|
let entryIdx = 0
|
|
for (let mi = 0; mi < $msgs.length; mi++) {
|
|
for (const t of $msgs[mi].tools) {
|
|
const label = toolActivityLabel(t)
|
|
if (t.type === 'tool_use') {
|
|
entries.push({
|
|
id: t.id ?? `tool_${mi}_${entryIdx++}`,
|
|
type: 'tool_running',
|
|
description: label,
|
|
timestamp: now - ($msgs.length - mi) * 1000,
|
|
toolName: t.name,
|
|
status: 'running'
|
|
})
|
|
} else if (t.type === 'tool_result') {
|
|
// Find and update matching tool_use entry
|
|
const running = entries.find((e) =>
|
|
e.type === 'tool_running' && e.id === t.id && e.status === 'running'
|
|
)
|
|
if (running && t.error) {
|
|
running.type = 'tool_error'
|
|
running.status = 'failed'
|
|
running.description = `${t.name}: ${t.error.slice(0, 80)}`
|
|
} else if (running) {
|
|
running.type = 'tool_done'
|
|
running.status = 'done'
|
|
running.detail = typeof t.result === 'string'
|
|
? t.result.slice(0, 200)
|
|
: JSON.stringify(t.result ?? '').slice(0, 200)
|
|
} else {
|
|
entries.push({
|
|
id: t.id ?? `tool_${mi}_${entryIdx++}`,
|
|
type: t.error ? 'tool_error' : 'tool_done',
|
|
description: t.error ? `${t.name}: ${t.error.slice(0, 80)}` : t.name,
|
|
detail: !t.error ? (typeof t.result === 'string' ? t.result.slice(0, 200) : '') : undefined,
|
|
timestamp: now - ($msgs.length - mi) * 1000,
|
|
toolName: t.name,
|
|
status: t.error ? 'failed' : 'done'
|
|
})
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
// Knowledge recorded — detect from upsert_knowledge tool results
|
|
for (let mi = 0; mi < $msgs.length; mi++) {
|
|
for (const t of $msgs[mi].tools) {
|
|
if (t.type === 'tool_result' && t.name === 'upsert_knowledge' && !t.error) {
|
|
const title = t.args?.title ?? ''
|
|
entries.push({
|
|
id: `knowledge_${mi}`,
|
|
type: 'knowledge',
|
|
description: title ? `Recorded: ${title.slice(0, 60)}` : 'Recorded knowledge',
|
|
timestamp: now - ($msgs.length - mi) * 1000,
|
|
status: 'done'
|
|
})
|
|
}
|
|
}
|
|
}
|
|
|
|
// Task completion
|
|
if ($task?.outcome) {
|
|
entries.push({
|
|
id: 'complete',
|
|
type: 'complete',
|
|
description: $task.summary || `Task ${$task.outcome}`,
|
|
timestamp: now,
|
|
status: $task.outcome === 'failure' ? 'failed' : 'done'
|
|
})
|
|
}
|
|
|
|
// Sort newest first
|
|
entries.sort((a, b) => b.timestamp - a.timestamp)
|
|
|
|
return entries
|
|
})
|
|
|
|
function toolActivityLabel(t: ToolCallResult): string {
|
|
const args = t.args ?? {}
|
|
switch (t.name) {
|
|
case 'set_goal': return 'Set goal'
|
|
case 'propose_plan': return 'Proposed plan'
|
|
case 'search_knowledge': return `Research: ${args.query || ''}`
|
|
case 'get_entity': return `Lookup: ${args.slug_or_id || ''}`
|
|
case 'get_entity_knowledge': return 'Check prior knowledge'
|
|
case 'get_relations': return 'Check relationships'
|
|
case 'list_lxcs': return 'List containers'
|
|
case 'list_entities': return 'List entities'
|
|
case 'get_health_summary': return 'Fleet health'
|
|
case 'get_state_snapshot': return 'State snapshot'
|
|
case 'run': {
|
|
const purpose = args.purpose as string || ''
|
|
const target = (args.target as string) || ''
|
|
if (purpose) return purpose
|
|
if (target) return `Run on ${target}`
|
|
return 'Run command'
|
|
}
|
|
case 'get_execution_status': return 'Check execution'
|
|
case 'update_plan_step': return 'Update plan'
|
|
case 'upsert_knowledge': return 'Record knowledge'
|
|
case 'complete_task': return 'Complete task'
|
|
case 'ping_service': return 'Check service'
|
|
case 'ask_operator': return 'Ask operator'
|
|
default: return t.name
|
|
}
|
|
}
|
|
|
|
// 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 tools = mergeToolCalls(m.content?.tool_calls)
|
|
return {
|
|
id: m.id,
|
|
role: m.role as 'user' | 'assistant',
|
|
text: m.content?.text ?? (typeof m.content === 'string' ? m.content : ''),
|
|
tools,
|
|
pendingApprovals: extractApprovals(tools)
|
|
}
|
|
})
|
|
}
|
|
|
|
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
|
|
// No cheap "anything new?" check: the auto-continuation worker updates a
|
|
// placeholder message IN PLACE as each tool call lands (see
|
|
// cmd/nomos/continue.go), so the message COUNT stays the same while the
|
|
// content changes — a length-only diff (the previous version of this
|
|
// code) never detected those updates and progress looked frozen even
|
|
// though the backend was actively working. Just re-set every tick;
|
|
// Svelte's own diffing keeps the actual re-render cheap.
|
|
sessionMessages.set(msgs)
|
|
messages.set(toChatMessages(msgs))
|
|
}, 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])
|
|
|
|
let 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') {
|
|
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') {
|
|
last.tools = last.tools.map((t) =>
|
|
t.id === ev.data.id ? updated : t
|
|
)
|
|
last.pendingApprovals = extractApprovals(last.tools)
|
|
}
|
|
return [...ms]
|
|
})
|
|
}
|
|
} else if (ev.type === 'text_delta') {
|
|
messages.update((ms) => {
|
|
const last = ms[ms.length - 1]
|
|
if (last && last.role === 'assistant') {
|
|
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') {
|
|
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') {
|
|
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 SSE stream drops mid-turn without
|
|
// receiving a 'done' event. Falls back to polling and attempts reconnection.
|
|
function handleDisconnect(sessionId: string) {
|
|
const MAX_RECONNECT = 3
|
|
connectionState.set('disconnected')
|
|
startPolling(sessionId)
|
|
addChatError('Agent connection lost. The task is still running — retrying…', 'Dismiss')
|
|
|
|
let attempts = 0
|
|
let delay = 1000
|
|
|
|
const attemptReconnect = () => {
|
|
if (get(currentSession) !== sessionId || attempts >= MAX_RECONNECT) {
|
|
connectionState.set('disconnected')
|
|
streaming.set(false)
|
|
return
|
|
}
|
|
if (attempts > 0) {
|
|
connectionState.set('reconnecting')
|
|
addChatError(`Reconnecting to agent (attempt ${attempts + 1}/${MAX_RECONNECT})…`, 'Dismiss')
|
|
}
|
|
attempts++
|
|
const controller = streamChat(
|
|
'',
|
|
sessionId,
|
|
(_ev: ChatEvent) => {},
|
|
(_err: string) => {
|
|
delay = Math.min(delay * 2, 8000)
|
|
setTimeout(attemptReconnect, delay)
|
|
},
|
|
() => {
|
|
if (get(currentSession) === sessionId) {
|
|
connectionState.set('connected')
|
|
streaming.set(false)
|
|
loadSessionMessages(sessionId)
|
|
}
|
|
}
|
|
)
|
|
if (activeControllers.get(sessionId)) {
|
|
activeControllers.get(sessionId)?.abort()
|
|
}
|
|
activeControllers.set(sessionId, controller)
|
|
}
|
|
|
|
setTimeout(attemptReconnect, delay)
|
|
}
|
|
|
|
export function reconnect() {
|
|
const sid = get(currentSession)
|
|
if (!sid) return
|
|
connectionState.set('reconnecting')
|
|
const controller = streamChat(
|
|
'',
|
|
sid,
|
|
(_ev: ChatEvent) => {},
|
|
(_err: string) => {
|
|
connectionState.set('disconnected')
|
|
addChatError('Reconnect failed. The task may still be running — try sending a message to wake the agent.', 'Dismiss')
|
|
},
|
|
() => {
|
|
if (get(currentSession) === sid) {
|
|
connectionState.set('connected')
|
|
streaming.set(false)
|
|
loadSessionMessages(sid)
|
|
}
|
|
}
|
|
)
|
|
if (activeControllers.get(sid)) {
|
|
activeControllers.get(sid)?.abort()
|
|
}
|
|
activeControllers.set(sid, controller)
|
|
}
|
|
|
|
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()
|
|
}
|