diff --git a/gui/views/chat/ChatWorkflow.vue b/gui/views/chat/ChatWorkflow.vue index bc738a5..5760afc 100644 --- a/gui/views/chat/ChatWorkflow.vue +++ b/gui/views/chat/ChatWorkflow.vue @@ -2,7 +2,11 @@ Top-level chat-style page for a single MCP workflow. Owns the message transcript and a single in-flight run; each user prompt triggers one /plugin/mcp/execute call and appends the resulting assistant message to - the transcript. The composer is disabled while a run is in progress. + the transcript. + + A run belongs to the server, so leaving the page only stops the poller. The + run_id rides on its assistant message, and a remount re-attaches. Only the + composer is gated during a run, because this view polls one at a time. This component intentionally does not maintain a server-side session: the backend is still single-shot. The chat UI gives users a familiar Claude.ai @@ -25,15 +29,16 @@ class="back-button" @click="$emit('back')" type="button" - :disabled="run.isRunning.value" - :title="run.isRunning.value - ? 'Cannot leave while a run is in progress' - : 'Back to workflow list'" + title="Back to workflow list" > Back - + Working @@ -48,7 +53,6 @@ class="header-action history-toggle" :class="{ 'is-on': historyEnabled }" @click="historyEnabled = !historyEnabled" - :disabled="run.isRunning.value" type="button" :title="historyEnabled ? 'Chat history is being threaded into each prompt. Click to disable for the rest of this session.' @@ -61,9 +65,10 @@ v-if="messages.length" class="header-action" @click="clearTranscript" - :disabled="run.isRunning.value" type="button" - title="Start a new chat" + :title="run.isRunning.value + ? 'Start a new chat. The run in progress keeps going on the server and stays in the History tab.' + : 'Start a new chat'" > New chat @@ -183,6 +188,8 @@ onMounted(() => { // before the first render pass. Has to happen before the height sync so // the transcript scroll position settles correctly on a hydrated view. session.hydrate() + // hydrate() leaves a still-running bubble RUNNING; give it a fresh poller. + _resumeRunInFlight() prevBodyOverflow = document.body.style.overflow prevHtmlOverflow = document.documentElement.style.overflow @@ -207,10 +214,12 @@ const { thoughts, adversary, abilityNames, splitSentences, isInjectedSentence } // Sync the in-flight run into its assistant message as state changes. The // assistant message is created at submit time with a known id; we update -// that message in place rather than re-pushing. +// that message in place rather than re-pushing. runId is watched too so the +// handle reaches the transcript without waiting for the first poll. let pendingAssistantId = null watch( () => ({ + runId: run.runId.value, status: run.status.value, stage: run.stage.value, finalResult: run.finalResult.value, @@ -222,6 +231,7 @@ watch( if (!pendingAssistantId) return const msg = messages.value.find(m => m.id === pendingAssistantId) if (!msg) return + msg.runId = run.runId.value msg.status = run.status.value msg.stage = run.stage.value msg.finalResult = run.finalResult.value @@ -237,10 +247,26 @@ watch( { deep: true } ) +// Only the newest bubble can still own a run: hydrate() has already failed +// every older RUNNING message. +function _resumeRunInFlight() { + const pending = [...messages.value] + .reverse() + .find(m => m.role === 'assistant' && m.status === 'RUNNING' && m.runId) + if (!pending) return + pendingAssistantId = pending.id + run.attach(pending.runId) +} + +// Without this the 1s interval outlives the view. The run itself keeps going. +onBeforeUnmount(run.stop) + // --- Workflow-derived UI bits ---------------------------------------------- const examplePrompts = computed(() => props.workflow?.example_prompts || []) const composerPlaceholder = computed(() => { - if (run.isRunning.value) return 'Working on the previous request…' + // The box is closed, but the page is not: say so. + if (run.isRunning.value) + return 'Working on the previous request. You can leave this page, it keeps running…' if (messages.value.length) { if (historyActive.value) return 'Send a follow-up prompt in this session…' if (supportsChatHistory.value) @@ -269,6 +295,8 @@ function handleSubmit() { messages.value.push({ id: assistantId, role: 'assistant', + // Set by _recordRunHandle; the handle a later mount re-attaches with. + runId: null, status: 'RUNNING', stage: '', finalResult: '', @@ -280,10 +308,20 @@ function handleSubmit() { }) composerText.value = '' - _startRun(text) + _startRun(text, assistantId) } -async function _startRun(text) { +// Written imperatively because Vue stops the watchers on unmount: clicking +// Back mid-POST would otherwise store a live run with runId null, and the +// next mount would call it failed. +function _recordRunHandle(assistantId, resp) { + if (resp.session_id && !sessionId.value) sessionId.value = resp.session_id + const msg = messages.value.find(m => m.id === assistantId) + if (msg) msg.runId = resp.run_id || null + session.persist() +} + +async function _startRun(text, assistantId) { const context = workflowContext.value || {} const selectedCtiFiles = Array.isArray(context.selected_stix_files) ? context.selected_stix_files @@ -367,16 +405,18 @@ async function _startRun(text) { try { const resp = await run.start(payload) - if (resp?.session_id && !sessionId.value) { - sessionId.value = resp.session_id - } + // null means "New chat" wiped this transcript mid-POST; adopting the ids + // would thread the fresh chat onto the run the user walked away from. + if (resp) _recordRunHandle(assistantId, resp) } catch { // run.errorMessage is already populated by useMcpRun. } } function clearTranscript() { - if (run.isRunning.value) return + // Allowed mid-run: only this view's link to the run is lost, which is the + // point. The run carries on and stays reachable in the History tab. + // // "New chat" is the only path that wipes durable session state. The // composable resets messages, sessionId, historyEnabled, and // selectedRag in memory AND removes this workflow's slice from diff --git a/gui/views/chat/composables/useChatSession.js b/gui/views/chat/composables/useChatSession.js index 07cbf2e..c0ced3d 100644 --- a/gui/views/chat/composables/useChatSession.js +++ b/gui/views/chat/composables/useChatSession.js @@ -23,9 +23,10 @@ // What it does not own: // composerText, transient run-state (status/stage/finalResult on the // in-flight assistant message). Composer text is intentionally not -// persisted — half-typed prompts surviving a remount is a worse UX -// than a clean text box. In-flight run state belongs to useMcpRun -// and is reconstructed from the polling endpoint, not localStorage. +// persisted: half-typed prompts surviving a remount is a worse UX +// than a clean text box. In-flight run state belongs to useMcpRun and +// is reconstructed from the polling endpoint; localStorage only keeps +// the run_id needed to ask for it again. // // Storage shape: // localStorage key 'mcp_chat_sessions' holds: @@ -36,11 +37,10 @@ // The schema version lets future changes drop incompatible blobs // without confusing rehydration. // -// On hydrate the composable also marks any message whose status is -// still 'RUNNING' as FAILED with a "polling stopped" note. Polling -// timers do not survive a component unmount; leaving the bubble -// frozen on "Thinking" forever is misleading. The MLflow run is still -// findable via the History tab if the user needs the result. +// Assistant messages carry their run_id, so a bubble left mid-run survives +// a remount and ChatWorkflow re-attaches a poller to it. Older RUNNING +// messages, and any with no run_id, are failed on hydrate: nothing is going +// to move them again. // // Limits: // Per-workflow caps at MAX_MESSAGES messages and MAX_BYTES total @@ -101,24 +101,24 @@ function _trimMessages(messages) { return trimmed } -function _markStaleRunningAsFailed(messages) { - // A polling timer cannot survive a component unmount. Any message - // that was RUNNING when we last saved is no longer being updated; - // surfacing the stale state would lie to the user. Mark it FAILED - // with a hint so the user knows where to look (History tab). - return messages.map(m => { - if (m.status === 'RUNNING') { - return { - ...m, - status: 'FAILED', - stage: '', - errorMessage: - m.errorMessage - || 'Polling stopped when you navigated away. ' - + 'Check the History tab for the final result.', - } +function _markUnresumableRunning(messages) { + // Only the newest RUNNING message can be re-attached; the view polls one + // run at a time. A bubble left frozen on "Thinking" would lie to the user. + const resumableIndex = messages.reduce( + (last, m, i) => (m.status === 'RUNNING' && m.runId ? i : last), + -1 + ) + return messages.map((m, i) => { + if (m.status !== 'RUNNING' || i === resumableIndex) return m + return { + ...m, + status: 'FAILED', + stage: '', + errorMessage: + m.errorMessage + || 'The page stopped tracking this run. ' + + 'Check the History tab for the final result.', } - return m }) } @@ -131,7 +131,7 @@ export function useChatSession(workflowId) { function hydrate() { const slice = _readWorkflow(workflowId) if (!slice) return - messages.value = _markStaleRunningAsFailed(slice.messages || []) + messages.value = _markUnresumableRunning(slice.messages || []) sessionId.value = slice.sessionId ?? null historyEnabled.value = typeof slice.historyEnabled === 'boolean' ? slice.historyEnabled : true @@ -172,6 +172,9 @@ export function useChatSession(workflowId) { historyEnabled, selectedRag, hydrate, + // For the one write that cannot wait for the watcher: a run_id arriving + // after unmount, once the scope holding that watcher is stopped. + persist, reset, } } diff --git a/gui/views/chat/composables/useMcpRun.js b/gui/views/chat/composables/useMcpRun.js index cbb9d94..c3c4e6e 100644 --- a/gui/views/chat/composables/useMcpRun.js +++ b/gui/views/chat/composables/useMcpRun.js @@ -1,13 +1,18 @@ // One MCP run: POST /plugin/mcp/execute, then poll /plugin/mcp/status until -// FINISHED or FAILED. Returns reactive state plus a start() function. +// FINISHED or FAILED. Returns reactive state plus start() and attach(). // -// Each run is independent — start() resets all state and begins fresh. The -// caller decides how to surface results (e.g. push into a chat transcript). +// The run belongs to the server, so it outlives the browser. attach() is the +// counterpart to start() for one already in flight, which is how a remounted +// view picks its run back up. import { ref, computed } from 'vue' +import { SESSION_EXPIRED } from '../../../composables/request.js' const POLL_INTERVAL_MS = 1000 +// A single failed status GET is usually a blip, not a dead run. +const MAX_CONSECUTIVE_POLL_ERRORS = 3 + export function useMcpRun($api) { const status = ref('idle') // 'idle' | 'RUNNING' | 'FINISHED' | 'FAILED' const stage = ref('') @@ -23,6 +28,20 @@ export function useMcpRun($api) { const isFailed = computed(() => status.value === 'FAILED') let pollTimer = null + let consecutivePollErrors = 0 + // Set by stop() so a request already in flight cannot start a fresh + // interval after the caller unmounted. + let detached = false + // Bumped by reset(). runId cannot mark supersession on its own: it is null + // for the whole /execute POST, which is the window "New chat" opens. + let generation = 0 + + function _clearTimer() { + if (pollTimer) { + clearInterval(pollTimer) + pollTimer = null + } + } function reset() { status.value = 'idle' @@ -33,71 +52,137 @@ export function useMcpRun($api) { finalResult.value = '' trajectory.value = {} errorMessage.value = '' - if (pollTimer) { - clearInterval(pollTimer) - pollTimer = null - } + consecutivePollErrors = 0 + detached = false + generation += 1 + _clearTimer() } + /** + * Submit a run; resolves with the /execute body, or null if "New chat" + * superseded it mid-POST. An unmounted caller still gets the body so the + * run_id can be persisted; only the polling is skipped. + */ async function start(payload) { reset() + const gen = generation status.value = 'RUNNING' prompt.value = payload.text || '' try { const response = await $api.post('/plugin/mcp/execute', payload) + if (gen !== generation) return null runId.value = response.data.run_id - _beginPolling(runId.value) - // Hand the parsed response back so the caller can grab fields like - // session_id that live above the per-run scope. + if (!detached) _beginPolling(runId.value) + // The caller needs fields above the per-run scope, like session_id. return response.data } catch (err) { + if (gen !== generation) return null status.value = 'FAILED' errorMessage.value = err?.response?.data?.error || 'Submission failed.' throw err } } + /** Resume polling a run already in flight on the server, by its `run_id`. */ + async function attach(id) { + reset() + status.value = 'RUNNING' + runId.value = id + // Read once up front so a remounted view paints real state immediately + // instead of holding stale content for a whole interval. + const stillLive = await _pollOnce(id) + if (stillLive && !detached) _beginPolling(id) + } + + // An expired session lands here as a 200 carrying the login page, not a + // snapshot (see gui/composables/request.js). Never persist that as a status. + function _isSnapshot(data) { + return !!data && typeof data === 'object' && typeof data.status === 'string' + } + + function _applySnapshot(data) { + status.value = data.status || 'unknown' + stage.value = data.stage || '' + prompt.value = data.prompt || prompt.value + reasoning.value = data.reasoning || '' + finalResult.value = data.process_result || '' + trajectory.value = data.trajectory || {} + // The run cache carries the workflow's own exception text; without this + // ChatMessage renders "Run failed." with no cause. + if (data.error) errorMessage.value = data.error + } + + // reset() signals "moved on" by clearing runId, so a GET issued for the + // abandoned run must not land on the state that replaced it. + function _superseded(id) { + return runId.value !== id + } + + // Returns false once polling should stop: terminal state, or superseded. + async function _pollOnce(id) { + try { + const res = await $api.get('/plugin/mcp/status', { params: { run_id: id } }) + if (_superseded(id)) return false + if (!_isSnapshot(res.data)) { + status.value = 'FAILED' + errorMessage.value = SESSION_EXPIRED + return false + } + consecutivePollErrors = 0 + _applySnapshot(res.data) + return status.value !== 'FINISHED' && status.value !== 'FAILED' + } catch (err) { + if (_superseded(id)) return false + // 404 means the run left the live cache: server restart, or LRU + // eviction. Neither resolves by asking again. + if (err?.response?.status === 404) { + status.value = 'FAILED' + errorMessage.value = + 'This run is no longer tracked live. ' + + 'Check the History tab for its result.' + return false + } + consecutivePollErrors += 1 + if (consecutivePollErrors < MAX_CONSECUTIVE_POLL_ERRORS) return true + // Losing the poll says nothing about the run, which is still going. + status.value = 'FAILED' + errorMessage.value = + 'Lost contact with the server while this run was in progress. ' + + 'Check the History tab for its result.' + return false + } + } + function _beginPolling(id) { - if (pollTimer) clearInterval(pollTimer) - pollTimer = setInterval(async () => { + _clearTimer() + // Per polling session, so a slow GET cannot stack up behind itself. + let inFlight = false + const timer = setInterval(async () => { + if (inFlight) return + inFlight = true try { - const res = await $api.get('/plugin/mcp/status', { params: { run_id: id } }) - status.value = res.data.status || 'unknown' - stage.value = res.data.stage || '' - prompt.value = res.data.prompt || prompt.value - reasoning.value = res.data.reasoning || '' - finalResult.value = res.data.process_result || '' - trajectory.value = res.data.trajectory || {} - // The backend writes a per-run `error` field into the run cache when - // the workflow raises (see mcp_svc._run_execution's except branch). - // Surface it so ChatMessage can render the actual cause under the - // "Run failed." headline instead of leaving it blank. - if (res.data.error) errorMessage.value = res.data.error - - if (status.value === 'FINISHED' || status.value === 'FAILED') { - clearInterval(pollTimer) - pollTimer = null - } - } catch (e) { - clearInterval(pollTimer) - pollTimer = null - status.value = 'FAILED' - errorMessage.value = 'Polling failed.' + const stillLive = await _pollOnce(id) + // Only ever clear our own timer: clearInterval cannot cancel a + // callback already suspended, so a superseded run can land here + // after a newer one installed its poller. + if (!stillLive && pollTimer === timer) _clearTimer() + } finally { + inFlight = false } }, POLL_INTERVAL_MS) + pollTimer = timer } + /** Stop polling without touching run state; the server run keeps going. */ function stop() { - if (pollTimer) { - clearInterval(pollTimer) - pollTimer = null - } + detached = true + _clearTimer() } return { status, stage, runId, prompt, reasoning, finalResult, trajectory, errorMessage, isRunning, isFinished, isFailed, - start, stop, reset, + start, attach, stop, reset, } }