feat: 补齐智能体对话思考和工具过程渲染

src/pages/CloudAgents/CloudChannelConversations.tsx

@@ -1,8 +1,7 @@
 import { useCallback, useEffect, useState } from 'react';
 import { Download, Folder, RefreshCw } from 'lucide-react';
 import { Button } from '@/components/ui/button';
-import ReactMarkdown from 'react-markdown';
-import remarkGfm from 'remark-gfm';
+import { CloudMessageBody } from './CloudMessageBody';
 import { cloudAgentsApi } from '@/lib/cloud-agents-api';
 import type { CloudAgentOperations, CloudChannelSession, CloudChannelConversation, CloudFile, CloudPendingOperation } from '../../../shared/cloud-agents';
 import { CloudApproval, cloudStatus } from './CloudChat';
@@ -185,7 +184,7 @@ function ChannelSession({ sessionId, visible, refreshRevision, onBusy, onNewSess
     {error && <p role="alert" className="text-sm text-destructive">{error}</p>}
     <div className="space-y-4 rounded-xl border border-border p-4" aria-label="微信消息">
       {data?.messages.filter(message => message.role !== 'system').map(message => <article key={message.id} className={message.role === 'user' ? 'ml-6 rounded-xl bg-muted/40 p-3' : 'p-1'}>
-        <p className="mb-2 text-xs text-muted-foreground">{message.role === 'user' ? '你 · 微信' : '智能体'}</p><div className="prose prose-sm max-w-none break-words"><ReactMarkdown remarkPlugins={[remarkGfm]} components={{ a: ({ href, children }) => href && /^(https?:|mailto:)/i.test(href) ? <a href={href} target="_blank" rel="noreferrer">{children}</a> : <span>{children}</span> }}>{message.content}</ReactMarkdown></div></article>)}
+        <p className="mb-2 text-xs text-muted-foreground">{message.role === 'user' ? '你 · 微信' : '智能体'}</p><CloudMessageBody {...message} /></article>)}
       {data && !data.messages.length && <p className="text-sm text-muted-foreground">还没有消息。</p>}
     </div>
     {!recoveryReady && <Button variant="outline" size="sm" onClick={() => void loadRecovery().catch(error => setError(failureText(error)))}>重新读取操作记录</Button>}

src/pages/CloudAgents/CloudChat.tsx

@@ -6,8 +6,8 @@ import { cloudAgentsApi } from '@/lib/cloud-agents-api';
 import type { CloudHistory, CloudInterrupt, CloudPrompt, CloudRequest, CloudRun, CloudThread, CloudScheduleProposal } from '../../../shared/cloud-agents';
 import { CloudFiles } from './CloudFiles';
 import './desktop-workspace.css';
-import ReactMarkdown, { defaultUrlTransform } from 'react-markdown';
-import remarkGfm from 'remark-gfm';
+import { CloudMessageBody } from './CloudMessageBody';
+import { reduceCloudStream, type CloudStream } from './cloud-chat-stream';
 import { CloudPendingContext, useCloudPendingState, usePendingCloudInput } from './CloudPending';
 import { agentExamples } from './agent-guidance';
 
@@ -39,7 +39,7 @@ export function CloudChat({ slug, previewRevision: latestRevision, initialThread
   const [intent, setIntent] = useState<CloudPrompt | null>(null);
   const [busy, setBusy] = useState(false);
   const [error, setError] = useState('');
-  const [liveText, setLiveText] = useState('');
+  const [liveStream, setLiveStream] = useState<CloudStream>({ runId: '', cursor: '', messages: [] });
   const [progress, setProgress] = useState('');
   const [interrupt, setInterrupt] = useState<CloudInterrupt | null>(null);
   const [loading, setLoading] = useState(true);
@@ -151,6 +151,9 @@ export function CloudChat({ slug, previewRevision: latestRevision, initialThread
 
   const runId = request && active(request.status) ? request.run_id ?? undefined : run?.agent_run_id;
   const running = active(run?.status) || active(request?.status);
+  const liveHistoryReady = history?.run?.agent_run_id === liveStream.runId
+    ? !active(history.run.status)
+    : Boolean(history?.messages.some(item => item.run_id === liveStream.runId));
   useEffect(() => {
     if (!runId || !running) return;
     let live = true;
@@ -163,16 +166,17 @@ export function CloudChat({ slug, previewRevision: latestRevision, initialThread
         const payload = object(envelope.payload);
         const chunk = object(payload.chunk);
         const targetThread = thread?.thread_id ?? request?.thread_id;
+        if (envelope.run_id && envelope.run_id !== runId) return;
         if (envelope.thread_id && targetThread && envelope.thread_id !== targetThread) return;
-        for (const item of Array.isArray(payload.items) ? payload.items : [chunk]) {
-          const semantic = object(object(item).stream_event);
-          if (semantic.type === 'message_delta' && typeof semantic.content === 'string') setLiveText(v => v + semantic.content);
-          if (semantic.type === 'tool_call' && typeof semantic.name === 'string') setProgress('正在使用 ' + semantic.name);
-        }
+        setLiveStream(current => reduceCloudStream(
+          current.runId === runId ? current : { runId, cursor: '', messages: [] },
+          envelope, targetThread ?? '', event.lastEventId,
+        ));
         if (event.type === 'interrupt') setInterrupt(chunk as CloudInterrupt);
         if (event.type === 'end') {
           source?.close();
-          setLiveText(''); setProgress('');
+          setLiveStream(current => ({ ...current, finished: true }));
+          setProgress('');
           const id = thread?.thread_id ?? request?.thread_id;
           if (id) void readHistory(id, g).catch(e => { if (live) setError(message(e)); });
         }
@@ -191,14 +195,14 @@ export function CloudChat({ slug, previewRevision: latestRevision, initialThread
         if (!live || g !== generation.current) return;
         setRun(next);
         if (!active(next.status)) {
-          source?.close(); setLiveText(''); setProgress(''); setInterrupt(next.interrupt ?? null);
+          source?.close(); setProgress(''); setInterrupt(next.interrupt ?? null);
           void readHistory(next.thread_id, g).catch(e => { if (live) setError(message(e)); });
         }
       }).catch(e => { if (live) setError(message(e)); });
     }, 5000);
     return () => { live = false; source?.close(); window.clearInterval(timer); };
   }, [runId, running, thread?.thread_id, request?.thread_id, readHistory]);
-  useEffect(() => { if (following.current && transcript.current) transcript.current.scrollTop = transcript.current.scrollHeight; }, [history?.messages.length, liveText, interrupt]);
+  useEffect(() => { if (following.current && transcript.current) transcript.current.scrollTop = transcript.current.scrollHeight; }, [history?.messages.length, liveStream, interrupt]);
   const stalePreview = previewRevision !== undefined && latestRevision !== undefined && previewRevision !== latestRevision;
 
   const rememberThread = (next: CloudThread) => {
@@ -218,7 +222,7 @@ export function CloudChat({ slug, previewRevision: latestRevision, initialThread
     try {
       const accepted = await cloudAgentsApi.call(previewRevision === undefined ? 'submit' : 'preview', { slug, ...input });
       if (!alive.current || g !== generation.current) return;
-      setRequest(accepted); setIntent(null); setQuery(''); setLiveText(''); setAttachmentIds([]);
+      setRequest(accepted); setIntent(null); setQuery(''); setLiveStream({ runId: '', cursor: '', messages: [] }); setAttachmentIds([]);
       rememberThread({ thread_id: accepted.thread_id, client_thread_id: clientId, agent_slug: slug, title: input.query.slice(0, 80),
         mode, draft_revision: previewRevision, updated_at: new Date().toISOString(), run_id: accepted.run_id, status: accepted.status, unread: false });
       await readHistory(accepted.thread_id, g);
@@ -231,7 +235,7 @@ export function CloudChat({ slug, previewRevision: latestRevision, initialThread
     if (mode === 'preview') setPreviewRevision(next?.draft_revision ?? revision);
     const g = ++generation.current;
     setThread(next); setClientId(next?.client_thread_id ?? crypto.randomUUID()); setHistory(null);
-    setRequest(null); setQueued([]); setRun(null); setLiveText(''); setInterrupt(null); setError(''); if (!keepQuery) setQuery('');
+    setRequest(null); setQueued([]); setRun(null); setLiveStream({ runId: '', cursor: '', messages: [] }); setInterrupt(null); setError(''); if (!keepQuery) setQuery('');
     setAttachmentIds([]);
     if (next) { setLoading(true); try { await readHistory(next.thread_id, g); } catch(e) { setError(message(e)); } finally { setLoading(false); } }
   };
@@ -280,7 +284,7 @@ export function CloudChat({ slug, previewRevision: latestRevision, initialThread
     <div ref={transcript} data-testid="cloud-chat-messages" className="agent-chat-scroll"
       onScroll={e => { const el = e.currentTarget; following.current = el.scrollHeight - el.scrollTop - el.clientHeight < 64; }}><div className="space-y-5 py-5 pr-2">
       {loading ? <p role="status" className="text-sm text-muted-foreground">正在读取对话…</p>
-        : !history?.messages.length && !liveText && <div className="space-y-4 py-6 text-sm leading-7">
+        : !history?.messages.length && !liveStream.messages.length && <div className="space-y-4 py-6 text-sm leading-7">
           <h2 className="font-medium">{mode === 'preview' ? '问它一个小问题' : '今天想让它帮你做什么？'}</h2>
           <p className="text-muted-foreground">{mode === 'preview' ? '看看它的回答是否符合你的要求。不满意就改一改，再试一次。' : '从下面选一句，或写下你自己的想法。'}</p>
           <div className="grid gap-2">{examples.map(example => <Button key={example} type="button" variant="outline"
@@ -293,8 +297,17 @@ export function CloudChat({ slug, previewRevision: latestRevision, initialThread
         <p className="my-2 text-xs text-muted-foreground">这是修改前的回答。用同一个问题再问一次，更容易看出变化。</p>
         <TrialMessages history={previousTrial.history} onError={setError} />
       </details>}
-      {history && <TrialMessages history={history} onError={setError} />}
-      {liveText && <div className="whitespace-pre-wrap break-words p-2 text-sm leading-7">{liveText}</div>}
+      {history && <TrialMessages history={{ ...history, messages: history.messages.filter(saved =>
+        !(!liveHistoryReady && saved.run_id === liveStream.runId && liveStream.messages.some(item => item.message_id === saved.message_id)),
+      ) }} onError={setError} />}
+      {liveStream.messages.filter(item => !liveHistoryReady || !history?.messages.some(saved => saved.run_id === liveStream.runId && saved.message_id === item.message_id)).map(item =>
+        <article key={liveStream.runId + ':' + item.message_id} className="min-w-0 p-2">
+          <p className="mb-2 text-xs text-muted-foreground">助手</p>
+          <CloudMessageBody {...item} active={running && !liveStream.finished && runId === liveStream.runId} onDownload={path => {
+            const id = thread?.thread_id ?? request?.thread_id;
+            if (id) void cloudAgentsApi.download(id, path).catch(e => setError(message(e)));
+          }} />
+        </article>)}
       {onSchedule && history?.schedule_proposals?.map(proposal => <article key={proposal.id} className="space-y-3 rounded-xl border border-violet-200 bg-violet-50/30 p-4 text-sm">
         <h3 className="font-medium">自动任务建议 · {proposal.name}</h3><p className="whitespace-pre-wrap">{proposal.prompt}</p>
         <p className="text-xs text-muted-foreground">{proposal.cron_expression} · {proposal.timezone} · 尚未启用</p>
@@ -302,7 +315,7 @@ export function CloudChat({ slug, previewRevision: latestRevision, initialThread
       </article>)}
       {run?.error && run.status === 'failed' && <p role="alert" className="text-sm text-destructive">{run.error.message}</p>}
       {interrupt && runId && !thread?.archived && <CloudApproval key={runId} interrupt={interrupt} runId={runId} onResumed={nextRun => {
-        setInterrupt(null); setLiveText(''); setRun(nextRun); setRequest(null);
+        setInterrupt(null); setLiveStream({ runId: '', cursor: '', messages: [] }); setRun(nextRun); setRequest(null);
       }} />}
 
     {queued.length > 0 && <div className="mb-3 space-y-2 rounded-lg bg-muted/40 p-3 text-sm">
@@ -352,11 +365,7 @@ export function CloudChat({ slug, previewRevision: latestRevision, initialThread
 function TrialMessages({ history, onError }: { history: CloudHistory; onError: (error: string) => void }) {
   return history.messages.map(item => <article key={item.id} className={item.role === 'user' ? 'ml-8 rounded-xl bg-muted/60 p-4' : 'p-2'}>
     <p className="mb-2 text-xs text-muted-foreground">{item.role === 'user' ? '你' : '助手'}</p>
-    <div className="prose prose-sm max-w-none break-words"><ReactMarkdown urlTransform={url => url.startsWith('sandbox:/') ? url.slice(8) : defaultUrlTransform(url)} remarkPlugins={[remarkGfm]} components={{
-      a: ({ href, children }) => href && !/^(https?:|mailto:|#)/i.test(href)
-        ? <button className="text-primary underline" onClick={() => void cloudAgentsApi.download(history.thread_id, href.replace(/^\/api\/chat\/thread\/[^/]+\/artifacts\//, '/')).catch(e => onError(message(e)))}>{children}</button>
-        : <a href={href} target="_blank" rel="noreferrer">{children}</a>,
-    }}>{item.content}</ReactMarkdown></div>
+    <CloudMessageBody {...item} onDownload={path => void cloudAgentsApi.download(history.thread_id, path).catch(e => onError(message(e)))} />
   </article>);
 }
 

src/pages/CloudAgents/CloudMessageBody.tsx

@@ -0,0 +1,34 @@
+import ReactMarkdown, { defaultUrlTransform } from 'react-markdown';
+import remarkGfm from 'remark-gfm';
+import type { CloudToolCall } from '../../../shared/cloud-agents';
+
+const display = (value: unknown) => typeof value === 'string' ? value : value == null ? '' : JSON.stringify(value, null, 2);
+
+export function CloudMessageBody({ content, reasoning_content, tool_calls = [], active = false, onDownload }: {
+  content: string; reasoning_content?: string; tool_calls?: CloudToolCall[]; active?: boolean;
+  onDownload?: (path: string) => void;
+}) {
+  return <div className="min-w-0 space-y-3">
+    {(reasoning_content || tool_calls.length > 0) && <details open={active && !content} className="rounded-lg border border-border bg-muted/20 p-3 text-xs text-muted-foreground" data-testid="cloud-message-process">
+      <summary className="cursor-pointer font-medium">{reasoning_content ? '思考过程' : '工具调用'}{tool_calls.length > 0 && ' · ' + tool_calls.length + ' 个工具'}{active && !content ? ' · 进行中' : ''}</summary>
+      <div className="mt-3 space-y-3">
+        {reasoning_content && <div aria-label="智能体思考过程" className="max-h-72 overflow-auto whitespace-pre-wrap break-words border-l-2 pl-3 leading-6">{reasoning_content}</div>}
+        {tool_calls.map(tool => <details key={tool.id} className="rounded-md border border-border p-2">
+          <summary className="cursor-pointer break-words">{tool.name} · {tool.status === 'success' ? '已完成' : tool.status === 'error' ? '失败' : active ? tool.status === 'running' ? '执行中' : '准备调用' : '未完成'}</summary>
+          <div className="mt-2 space-y-2">
+            <p>输入</p><pre className="max-h-48 overflow-auto whitespace-pre-wrap break-words">{display(tool.args)}</pre>
+            {tool.tool_call_result != null && <><p>结果</p><pre className="max-h-60 overflow-auto whitespace-pre-wrap break-words">{display(tool.tool_call_result.content)}</pre></>}
+            {tool.error_message && <p className="whitespace-pre-wrap break-words text-destructive">{tool.error_message}</p>}
+          </div>
+        </details>)}
+      </div>
+    </details>}
+    {content && <div className="prose prose-sm max-w-none break-words [&_pre]:max-w-full [&_pre]:overflow-x-auto [&_table]:block [&_table]:overflow-x-auto">
+      <ReactMarkdown urlTransform={url => url.startsWith('sandbox:/') ? url.slice(8) : defaultUrlTransform(url)} remarkPlugins={[remarkGfm]} components={{
+        a: ({ href, children }) => href && !/^(https?:|mailto:|#)/i.test(href)
+          ? onDownload ? <button className="text-primary underline" onClick={() => onDownload(href.replace(/^\/api\/chat\/thread\/[^/]+\/artifacts\//, '/'))}>{children}</button> : <span>{children}</span>
+          : <a href={href} target="_blank" rel="noreferrer">{children}</a>,
+      }}>{content}</ReactMarkdown>
+    </div>}
+  </div>;
+}

src/pages/CloudAgents/cloud-chat-stream.ts

@@ -0,0 +1,74 @@
+import type { CloudToolCall } from '../../../shared/cloud-agents';
+
+export interface CloudLiveMessage {
+  message_id: string;
+  content: string;
+  reasoning_content: string;
+  tool_calls: (CloudToolCall & { index?: number })[];
+}
+export interface CloudStream {
+  runId: string;
+  cursor: string;
+  messages: CloudLiveMessage[];
+  finished?: boolean;
+}
+export const object = (value: unknown): Record<string, unknown> =>
+  value !== null && typeof value === 'object' && !Array.isArray(value) ? value as Record<string, unknown> : {};
+
+/** Redis SSE cursors order replayed batches without treating subthread output as an answer. */
+export function reduceCloudStream(state: CloudStream, data: unknown, threadId: string, cursor: string): CloudStream {
+  if (cursor && /^\d+-\d+$/.test(cursor) && state.cursor) {
+    const [time, sequence] = cursor.split('-').map(BigInt);
+    const [previousTime, previousSequence] = state.cursor.split('-').map(BigInt);
+    if (time < previousTime || (time === previousTime && sequence <= previousSequence)) return state;
+  }
+  const envelope = object(data);
+  if (envelope.run_id && envelope.run_id !== state.runId) return state;
+  const next = { ...state, cursor: /^\d+-\d+$/.test(cursor) ? cursor : state.cursor };
+  if (envelope.thread_id && envelope.thread_id !== threadId) return next;
+  const payload = object(envelope.payload);
+  const chunks = Array.isArray(payload.items) ? payload.items : [payload.chunk];
+  let messages = state.messages;
+  for (const value of chunks) {
+    const chunk = object(value);
+    const event = object(chunk.stream_event);
+    if ((chunk.thread_id && chunk.thread_id !== threadId) || (event.thread_id && event.thread_id !== threadId)) continue;
+    if (['message_delta', 'tool_call', 'tool_call_delta'].includes(String(event.type)) && typeof event.message_id === 'string') {
+      const id = event.message_id;
+      const current = messages.find(message => message.message_id === id)
+        ?? { message_id: id, content: '', reasoning_content: '', tool_calls: [] };
+      const updated = { ...current };
+      if (event.type === 'message_delta') {
+        updated.content += typeof event.content === 'string' ? event.content : '';
+        updated.reasoning_content += typeof event.reasoning_content === 'string' ? event.reasoning_content
+          : typeof event.additional_reasoning_content === 'string' ? event.additional_reasoning_content : '';
+      } else {
+        const index = typeof event.index === 'number' ? event.index : 0;
+        const existing = current.tool_calls.find(tool => (event.tool_call_id && tool.id === event.tool_call_id) || tool.index === index);
+        const tool = { ...existing, id: typeof event.tool_call_id === 'string' ? event.tool_call_id : existing?.id ?? id + ':' + index,
+          name: typeof event.name === 'string' ? event.name : existing?.name ?? '工具',
+          args: existing && existing.status !== 'pending' ? existing.args
+            : event.type === 'tool_call' ? event.args ?? {} : String(existing?.args ?? '') + String(event.args_delta ?? ''),
+          status: existing?.status ?? 'pending', index };
+        updated.tool_calls = existing ? current.tool_calls.map(item => item === existing ? tool : item) : [...current.tool_calls, tool];
+      }
+      messages = messages.some(message => message.message_id === id)
+        ? messages.map(message => message.message_id === id ? updated : message) : [...messages, updated];
+    }
+    const toolEvent = object(chunk.event);
+    const result = object(toolEvent.data);
+    if (toolEvent.method === 'tools' && typeof result.tool_call_id === 'string') {
+      messages = messages.map(message => ({ ...message, tool_calls: message.tool_calls.map(tool => {
+        if (tool.id !== result.tool_call_id) return tool;
+        if (result.event === 'tool-started') return { ...tool, args: result.input ?? tool.args, status: 'running' };
+        if (result.event !== 'tool-finished') return tool;
+        const output = object(result.output);
+        const error = result.error || (output.status === 'error' ? output.content : null);
+        return { ...tool, status: error ? 'error' : 'success',
+          error_message: error ? (typeof error === 'string' ? error : JSON.stringify(error)) : null,
+          tool_call_result: { content: output.content ?? result.output ?? '' } };
+      }) }));
+    }
+  }
+  return { ...next, messages };
+}