Files
kamidso-front/src/features/assistant/stream/useRunStream.ts
T
javadandCursor b17782eec2 Add Didar connections and a data hub for attaching sources.
The assistant can pick connectors and show subagent steps, and the databases page is replaced by connections, sources, and a dashboard.

Co-authored-by: Cursor <cursoragent@cursor.com>
2026-10-12 19:26:46 +03:30

654 lines
23 KiB
TypeScript

import { useCallback, useEffect, useReducer, useRef, useState } from 'react'
import { streamChat, cancelRun } from '../api/chat.api'
import { getConversationHistory } from '../api/conversations.api'
import { getRun, streamRunEvents } from '../api/runs.api'
import { isLiveRunStatus } from '../../../contracts/common'
import type { KamidsoArtifact } from '../../../contracts/run'
import { unwrapWorkspaceEvent } from '../lib/events'
import { cleanFinalContent, decodeFinalEvent } from '../lib/final'
import type { ChatMessage, ExecutionStep } from '../types/chat'
import { initialRunState } from '../state/runState'
import { historyToMessages } from '../adapters/historyToMessages'
import { runReducer } from './eventReducer'
import { seedTurnOutputs, downloadFilesFromArtifacts } from './downloadFiles'
const ACTIVE_RUN_KEY = 'kamidso.activeRunByConv'
const STREAM_TERMINAL_TYPES = new Set([
'final',
'answer.completed',
'run.completed',
'run.failed',
'run.cancelled',
'done',
])
function sleep(ms: number) {
return new Promise((resolve) => window.setTimeout(resolve, ms))
}
function readActiveRunMap(): Record<string, { runId: string; viewId: string }> {
try {
const raw = sessionStorage.getItem(ACTIVE_RUN_KEY)
if (!raw) return {}
const parsed = JSON.parse(raw) as Record<string, { runId: string; viewId: string }>
return parsed && typeof parsed === 'object' ? parsed : {}
} catch {
return {}
}
}
function rememberActiveRun(conversationId: string, runId: string, viewId: string) {
if (!conversationId || !runId) return
try {
const map = readActiveRunMap()
map[conversationId] = { runId, viewId }
sessionStorage.setItem(ACTIVE_RUN_KEY, JSON.stringify(map))
} catch {
/* ignore quota / private mode */
}
}
function forgetActiveRun(conversationId: string | null) {
if (!conversationId) return
try {
const map = readActiveRunMap()
delete map[conversationId]
sessionStorage.setItem(ACTIVE_RUN_KEY, JSON.stringify(map))
} catch {
/* ignore */
}
}
function peekActiveRun(conversationId: string) {
return readActiveRunMap()[conversationId] ?? null
}
type StreamSendParams = {
conversationId: string
query: string
model?: string
skillId?: string
reasoningMode?: 'fast' | 'medium' | 'deep'
reportMode?: boolean
resources?: Array<{ id: string; type: 'file' | 'datasource' | 'knowledge' | 'connector'; name?: string }>
humanMsg: ChatMessage
viewMsg: ChatMessage
}
export function useRunStream() {
const [state, dispatch] = useReducer(runReducer, undefined, initialRunState)
const [messages, setMessages] = useState<ChatMessage[]>([])
const [conversationId, setConversationId] = useState<string | null>(null)
const abortRef = useRef<AbortController | null>(null)
const activeRunId = useRef<string | null>(null)
const conversationIdRef = useRef<string | null>(null)
const messagesRef = useRef<ChatMessage[]>([])
const thinkingByViewIdRef = useRef<Record<string, string>>({})
const activeViewIdRef = useRef<string | null>(null)
const lastSendRef = useRef<StreamSendParams | null>(null)
const restoreGen = useRef(0)
const streamGen = useRef(0)
const lastSeqRef = useRef(0)
messagesRef.current = messages
thinkingByViewIdRef.current = state.thinkingByViewId
activeViewIdRef.current = state.activeViewId
const bindConversation = useCallback((id: string | null) => {
conversationIdRef.current = id
setConversationId(id)
}, [])
const stop = useCallback(() => {
streamGen.current += 1
abortRef.current?.abort()
abortRef.current = null
}, [])
useEffect(() => {
return () => {
abortRef.current?.abort()
abortRef.current = null
}
}, [])
const releaseLiveRun = useCallback(() => {
activeRunId.current = null
forgetActiveRun(conversationIdRef.current)
}, [])
const markViewStopped = useCallback((viewId: string | null) => {
if (!viewId) return
const thought = thinkingByViewIdRef.current[viewId] || ''
dispatch({
type: 'PROCESS_EVENT',
viewId,
payload: { type: 'run.cancelled' },
})
setMessages((prev) =>
prev.map((msg) =>
msg.id === viewId && msg.role === 'view'
? {
...msg,
thinking: false,
stopped: true,
canRetry: true,
thought: thought || msg.thought,
context: '',
}
: msg,
),
)
}, [])
const cancel = useCallback(async () => {
const runId = activeRunId.current
const viewId = activeViewIdRef.current
if (runId) {
await cancelRun(runId)
}
markViewStopped(viewId)
releaseLiveRun()
stop()
dispatch({ type: 'SET_LOADING', loading: false })
}, [markViewStopped, releaseLiveRun, stop])
const readStream = useCallback(async (response: Response, viewId: string) => {
const reader = response.body!.getReader()
const decoder = new TextDecoder('utf-8')
const thisController = abortRef.current
let buffer = ''
let reachedTerminal = false
try {
while (true) {
const { done, value } = await reader.read()
if (done) break
buffer += decoder.decode(value, { stream: true })
const parts = buffer.split('\n\n')
buffer = parts.pop() || ''
for (const part of parts) {
const line = part.trim()
if (!line.startsWith('data:')) continue
const data = line.slice(5).trim()
if (!data || data === '[DONE]') continue
try {
const payload = unwrapWorkspaceEvent(JSON.parse(data) as Record<string, unknown>)
if (payload.type === 'ping') continue
const seq = Number(payload.sequence ?? 0)
if (seq > lastSeqRef.current) lastSeqRef.current = seq
if (STREAM_TERMINAL_TYPES.has(String(payload.type ?? ''))) {
reachedTerminal = true
activeRunId.current = null
forgetActiveRun(conversationIdRef.current)
}
if (payload.run_id && !STREAM_TERMINAL_TYPES.has(String(payload.type ?? ''))) {
activeRunId.current = String(payload.run_id)
rememberActiveRun(conversationIdRef.current || '', String(payload.run_id), viewId)
setMessages((prev) =>
prev.map((msg) =>
msg.id === viewId && msg.role === 'view'
? { ...msg, runId: String(payload.run_id) }
: msg,
),
)
}
dispatch({ type: 'PROCESS_EVENT', viewId, payload })
if (payload.type === 'artifact.created') {
const artifact = payload.artifact as KamidsoArtifact | undefined
if (artifact?.type === 'file') {
const files = downloadFilesFromArtifacts([artifact])
if (files.length) {
setMessages((prev) =>
prev.map((msg) => {
if (msg.id !== viewId || msg.role !== 'view') return msg
const existing = msg.downloadFiles ?? []
const seen = new Set(existing.map((f) => f.url))
const merged = [...existing]
for (const file of files) {
if (!seen.has(file.url)) merged.push(file)
}
return { ...msg, downloadFiles: merged }
}),
)
}
}
}
if (payload.type === 'final' || payload.type === 'answer.completed') {
const finalAnswer = decodeFinalEvent(payload)
const summaryText = cleanFinalContent(finalAnswer.content)
setMessages((prev) =>
prev.map((msg) =>
msg.id === viewId && msg.role === 'view'
? { ...msg, context: summaryText, citations: finalAnswer.citations, thinking: false }
: msg,
),
)
}
if (payload.type === 'plan.update' || payload.type === 'plan.updated') {
const tasks = (payload.tasks ?? payload.plan) as ChatMessage['taskPlan']
if (Array.isArray(tasks)) {
setMessages((prev) =>
prev.map((msg) => (msg.id === viewId && msg.role === 'view' ? { ...msg, taskPlan: tasks } : msg)),
)
}
}
if (payload.type === 'run.cancelled') {
markViewStopped(viewId)
}
if (payload.type === 'run.failed') {
const err = (payload.error as Record<string, unknown> | undefined) ?? payload
const message = String(err.message ?? 'خطا در دریافت پاسخ. دوباره تلاش کنید.')
setMessages((prev) =>
prev.map((msg) =>
msg.id === viewId && msg.role === 'view' ? { ...msg, context: message, thinking: false } : msg,
),
)
}
} catch {
/* ignore malformed */
}
}
}
} finally {
if (abortRef.current === thisController) {
abortRef.current = null
}
}
return reachedTerminal
}, [markViewStopped])
const finishInactiveRun = useCallback((viewId: string, run: { status?: string; answer?: string | null; error?: unknown }) => {
if (run.status === 'cancelled') {
markViewStopped(viewId)
activeRunId.current = null
forgetActiveRun(conversationIdRef.current)
dispatch({ type: 'SET_LOADING', loading: false })
return
}
const failed = run.status === 'failed'
const message = failed
? String((run.error as { message?: string } | undefined)?.message ?? 'خطا در دریافت پاسخ. دوباره تلاش کنید.')
: run.answer
? String(run.answer)
: undefined
setMessages((prev) =>
prev.map((msg) =>
msg.id === viewId && msg.role === 'view'
? {
...msg,
context: message ?? msg.context,
thinking: false,
}
: msg,
),
)
activeRunId.current = null
forgetActiveRun(conversationIdRef.current)
dispatch({ type: 'SET_LOADING', loading: false })
}, [markViewStopped])
const recoverLiveStream = useCallback(
async (viewId: string, gen: number) => {
const runId = activeRunId.current
if (!runId) {
dispatch({ type: 'SET_LOADING', loading: false })
return
}
let delay = 400
while (gen === streamGen.current) {
await sleep(delay)
if (gen !== streamGen.current) return
try {
const run = await getRun(runId)
if (gen !== streamGen.current) return
if (!isLiveRunStatus(run.status)) {
finishInactiveRun(viewId, run)
return
}
dispatch({ type: 'SET_LOADING', loading: true })
const controller = new AbortController()
abortRef.current = controller
const response = await streamRunEvents(runId, lastSeqRef.current, controller.signal)
const terminal = await readStream(response, viewId)
if (terminal || controller.signal.aborted || gen !== streamGen.current) return
delay = 400
} catch (err) {
if ((err as Error).name === 'AbortError' || gen !== streamGen.current) return
delay = Math.min(Math.round(delay * 1.5), 5000)
}
}
},
[finishInactiveRun, readStream],
)
const hydrateRun = useCallback(
async (runId: string, viewId: string, options?: { live?: boolean; setActive?: boolean }) => {
const setActive = options?.setActive !== false
try {
const run = await getRun(runId)
if (setActive) activeRunId.current = runId
const live = options?.live ?? isLiveRunStatus(run.status)
const steps = (run.steps ?? []).map((step, index) => ({
id: step.id,
step: step.step ?? index + 1,
title: step.title,
status: (step.status === 'done' ? 'done' : step.status) as ExecutionStep['status'],
detail: step.detail,
phase: step.phase,
action: step.action,
}))
const running = [...steps].reverse().find((item) => item.status === 'running')
const artifacts = (run.artifacts ?? []) as KamidsoArtifact[]
const { outputs: seededOutputs, html: htmlReport } = seedTurnOutputs(artifacts)
const downloadFiles = downloadFilesFromArtifacts(artifacts)
dispatch({
type: 'HYDRATE_RUN',
run: {
id: setActive ? runId : undefined,
status: setActive ? run.status : undefined,
answer: setActive ? run.answer ?? '' : undefined,
plan: setActive ? [] : undefined,
steps: setActive ? steps : undefined,
artifacts: setActive ? artifacts : undefined,
activeViewId: setActive ? viewId : undefined,
loading: setActive ? live : undefined,
lastEventSequence: setActive ? 0 : undefined,
htmlByViewId: htmlReport ? { [viewId]: htmlReport } : undefined,
executionByViewId: {
[viewId]: {
steps,
outputs: seededOutputs,
activeStepId: running?.id ?? steps[steps.length - 1]?.id ?? null,
stepThoughts: {},
},
},
},
})
setMessages((prev) =>
prev.map((msg) => {
if (msg.id !== viewId || msg.role !== 'view') return msg
return {
...msg,
runId,
downloadFiles: downloadFiles.length
? downloadFiles
: msg.downloadFiles,
context:
setActive && run.answer
? run.answer
: msg.context,
thinking: setActive ? (live && !run.answer ? true : false) : msg.thinking,
}
}),
)
if (setActive) {
lastSeqRef.current = 0
const gen = streamGen.current
if (live) {
rememberActiveRun(conversationIdRef.current || '', runId, viewId)
dispatch({ type: 'SET_LOADING', loading: true })
const controller = new AbortController()
abortRef.current = controller
try {
const response = await streamRunEvents(runId, 0, controller.signal)
const terminal = await readStream(response, viewId)
if (!terminal && !controller.signal.aborted && gen === streamGen.current) {
await recoverLiveStream(viewId, gen)
} else if (gen === streamGen.current) {
dispatch({ type: 'SET_LOADING', loading: false })
}
} catch (err) {
if ((err as Error).name === 'AbortError' || gen !== streamGen.current) return
await recoverLiveStream(viewId, gen)
}
} else {
if (setActive) activeRunId.current = null
forgetActiveRun(conversationIdRef.current)
const controller = new AbortController()
abortRef.current = controller
try {
const response = await streamRunEvents(runId, 0, controller.signal)
await readStream(response, viewId)
} catch {
/* replay best-effort */
}
if (gen === streamGen.current) {
dispatch({ type: 'SET_LOADING', loading: false })
}
}
} else if (!live && (downloadFiles.length === 0 || !htmlReport)) {
try {
const response = await streamRunEvents(runId, 0)
await readStream(response, viewId)
} catch {
/* keep seeded/empty state */
}
}
} catch (err) {
if (setActive && (err as Error).name !== 'AbortError') {
dispatch({ type: 'SET_LOADING', loading: false })
}
}
},
[readStream, recoverLiveStream],
)
const restoreConversation = useCallback(
async (nextConversationId: string) => {
if (
conversationIdRef.current === nextConversationId &&
activeRunId.current &&
abortRef.current
) {
bindConversation(nextConversationId)
return
}
const gen = ++restoreGen.current
stop()
bindConversation(nextConversationId)
dispatch({ type: 'RESET' })
setMessages([])
activeRunId.current = null
try {
const data = await getConversationHistory(nextConversationId)
if (gen !== restoreGen.current) return
const stored = peekActiveRun(nextConversationId)
const latestRun = data.latest_run
? {
id: data.latest_run.id,
status: data.latest_run.status,
answer: data.latest_run.answer,
resources: data.latest_run.resources,
}
: stored
? { id: stored.runId, status: 'running' }
: null
const mapped = historyToMessages(data.messages, latestRun)
setMessages(mapped)
const runId = data.active_run_id || stored?.runId || data.latest_run?.id
const lastView = [...mapped].reverse().find((msg) => msg.role === 'view')
// Restore download links for every prior turn that has a run_id.
const priorViews = mapped.filter(
(msg) => msg.role === 'view' && msg.runId && msg.id !== lastView?.id,
)
for (const view of priorViews) {
if (gen !== restoreGen.current || !view.runId) break
await hydrateRun(view.runId, view.id, { live: false, setActive: false })
}
if (runId && lastView) {
await hydrateRun(runId, lastView.id, {
live: isLiveRunStatus(latestRun?.status || data.latest_run?.status || ''),
setActive: true,
})
}
} catch {
if (gen === restoreGen.current) {
dispatch({ type: 'SET_LOADING', loading: false })
}
}
},
[bindConversation, hydrateRun, stop],
)
const startStream = useCallback(
async (params: StreamSendParams) => {
lastSendRef.current = params
lastSeqRef.current = 0
const gen = streamGen.current
bindConversation(params.conversationId)
dispatch({ type: 'RUN_CREATED', runId: '', viewId: params.viewMsg.id })
dispatch({ type: 'SET_LOADING', loading: true })
const controller = new AbortController()
abortRef.current = controller
try {
const response = await streamChat(
{
conversation_id: params.conversationId,
message: params.query,
model_id: params.model,
skill_id: params.skillId,
reasoning_mode: params.reasoningMode || 'fast',
report_mode: Boolean(params.reportMode),
resources: params.resources,
mode: 'auto',
temperature: 0.2,
max_new_tokens: 65536,
},
controller.signal,
)
const terminal = await readStream(response, params.viewMsg.id)
if (!terminal && !controller.signal.aborted && gen === streamGen.current) {
await recoverLiveStream(params.viewMsg.id, gen)
} else if (gen === streamGen.current) {
dispatch({ type: 'SET_LOADING', loading: false })
}
} catch (err) {
if ((err as Error).name === 'AbortError' || gen !== streamGen.current) return
if (activeRunId.current) {
await recoverLiveStream(params.viewMsg.id, gen)
return
}
setMessages((prev) =>
prev.map((msg) =>
msg.id === params.viewMsg.id
? { ...msg, context: 'خطا در دریافت پاسخ. دوباره تلاش کنید.', thinking: false }
: msg,
),
)
dispatch({ type: 'SET_LOADING', loading: false })
}
},
[bindConversation, readStream, recoverLiveStream],
)
const send = useCallback(
async (params: StreamSendParams) => {
if (state.loading || abortRef.current) {
return
}
stop()
setMessages((prev) => [
...prev.map((msg) =>
msg.role === 'view' && msg.stopped ? { ...msg, canRetry: false } : msg,
),
params.humanMsg,
params.viewMsg,
])
await startStream(params)
},
[startStream, state.loading, stop],
)
const retryStopped = useCallback(async () => {
if (state.loading || abortRef.current) return
stop()
const msgs = messagesRef.current
let viewIndex = -1
for (let i = msgs.length - 1; i >= 0; i -= 1) {
const msg = msgs[i]
if (msg.role === 'view' && msg.stopped && msg.canRetry) {
viewIndex = i
break
}
}
if (viewIndex < 1) return
const viewMsg = msgs[viewIndex]
const humanMsg = msgs[viewIndex - 1]
if (humanMsg.role !== 'human') return
const previous = lastSendRef.current
const conversationId = previous?.conversationId || conversationIdRef.current
if (!conversationId) return
const nextView: ChatMessage = {
...viewMsg,
context: '',
thinking: true,
stopped: false,
canRetry: false,
thought: undefined,
}
setMessages((prev) => prev.map((msg) => (msg.id === viewMsg.id ? nextView : msg)))
await startStream({
conversationId,
query: previous?.query || humanMsg.context,
model: previous?.model,
skillId: previous?.skillId || humanMsg.attachedSkill?.id,
reasoningMode: previous?.reasoningMode,
reportMode: previous?.reportMode,
resources: previous?.resources,
humanMsg,
viewMsg: nextView,
})
}, [startStream, state.loading, stop])
const resetConversation = useCallback(() => {
restoreGen.current += 1
stop()
forgetActiveRun(conversationIdRef.current)
dispatch({ type: 'RESET' })
setMessages([])
activeRunId.current = null
bindConversation(null)
}, [bindConversation, stop])
const setPendingQuestion = useCallback((q: typeof state.question) => {
if (!q) dispatch({ type: 'CLEAR_QUESTION' })
}, [])
return {
messages,
setMessages,
conversationId,
executionMap: state.executionByViewId,
loading: state.loading,
taskPlan: state.plan,
pendingQuestion: state.question,
setPendingQuestion,
activeViewId: state.activeViewId,
setActiveViewId: (id: string | null) => dispatch({ type: 'SET_ACTIVE_VIEW', viewId: id }),
runStatus: state.status,
runId: state.id ?? activeRunId.current,
artifacts: state.artifacts,
htmlByViewId: state.htmlByViewId,
thinkingByViewId: state.thinkingByViewId,
answer: state.answer,
error: state.error,
send,
retryStopped,
stop,
cancel,
hydrateRun,
restoreConversation,
resetConversation,
}
}