From a688b6977e61918db570d647b7a617126977f591 Mon Sep 17 00:00:00 2001 From: wangjia <809946525@qq.com> Date: Wed, 24 Jun 2026 17:01:58 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20agent=20=E5=AE=9E=E6=97=B6=E8=BE=93?= =?UTF-8?q?=E5=87=BA=E6=B5=81=EF=BC=88SSE=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit B-② SSE 实时流（⑧ ★ GET /api/tasks/:id/stream）： 后端： - GET /api/tasks/:id/stream：SSE 跟随任务最近一条 run 的 transcript 文件， 新写入的 agent 消息逐行推送（活跃 run=实时；已结束=回放后 end）； 500ms 轮询 tail（按 offset 增量读），run 转非 started 即结束流，客户端断开即停 前端： - StreamModal（EventSource 消费 SSE，summarizeMessage 把 SDK 消息→可读行： 助手文本/⚙工具调用/■终态/·maestro 元事件，自动滚动到底） - AgentSection agent 行可点击 → 打开实时流模态；adaptActiveAgents 带出 taskId 验证：typecheck 干净；前端 build 通过；dev 渲染 0 console 错误。 （实时效果需有运行中的 agent；端到端随 daemon 重启验证。） Co-Authored-By: Claude Opus 4.8 --- app/src/adapt.js | 2 +- app/src/app.jsx | 59 ++++++++++++++++++++++++++++++- design/ui_kits/console/Topbar.jsx | 8 +++-- src/api/server.ts | 41 ++++++++++++++++++++- 4 files changed, 105 insertions(+), 5 deletions(-) diff --git a/app/src/adapt.js b/app/src/adapt.js index 91dc877..fd1f59f 100644 --- a/app/src/adapt.js +++ b/app/src/adapt.js @@ -160,7 +160,7 @@ export function adaptActiveAgents(agentsResp, projectId) { if (projectId && a.projectId !== projectId) continue; for (const r of a.active || []) { out.push({ - id: r.runId || r.id, project: a.projectName, + id: r.runId || r.id, taskId: r.taskId, project: a.projectName, title: r.title || r.taskTitle || r.taskId || '（运行中）', kind: (r.kind || 'run').toUpperCase(), time: TIME(r.startedAt || r.at), meta: `${r.worktree || ''} · ${r.model || ''}`, diff --git a/app/src/app.jsx b/app/src/app.jsx index d5f997c..9993736 100644 --- a/app/src/app.jsx +++ b/app/src/app.jsx @@ -48,6 +48,61 @@ function NewProjectModal({ t, lang, onCreate, onClose }) { ); } +// SDK 流式消息 → 一行可读文本（助手文本 / 工具调用 / 终态 / maestro 元事件） +function summarizeMessage(m) { + if (!m || !m.type) return ''; + const content = m.message && m.message.content; + if (m.type === 'assistant' && Array.isArray(content)) { + return content.map((c) => (c.type === 'text' ? c.text : c.type === 'tool_use' ? `⚙ ${c.name}` : '')).filter(Boolean).join(' '); + } + if (m.type === 'user' && Array.isArray(content)) { + return content.some((c) => c.type === 'tool_result') ? '↩ tool result' : ''; + } + if (m.type === 'result') return `■ ${m.subtype || 'done'}`; + if (typeof m.type === 'string' && m.type.startsWith('maestro.')) return `· ${m.type}`; + return ''; +} + +// 实时流模态：EventSource 跟随某 run 的 transcript，逐条渲染 agent 输出 +function StreamModal({ agent, onClose }) { + const [lines, setLines] = React.useState([]); + const [status, setStatus] = React.useState('connecting'); + const boxRef = React.useRef(null); + React.useEffect(() => { + const es = new EventSource(`/api/tasks/${agent.taskId}/stream`); + es.addEventListener('open', () => setStatus('streaming')); + es.onmessage = (e) => { + let text = ''; + try { text = summarizeMessage(JSON.parse(e.data)); } catch { text = ''; } + if (text) setLines((ls) => [...ls.slice(-400), text]); + }; + es.addEventListener('end', () => { setStatus('done'); es.close(); }); + es.onerror = () => { setStatus('disconnected'); es.close(); }; + return () => es.close(); + }, [agent.taskId]); + React.useEffect(() => { if (boxRef.current) boxRef.current.scrollTop = boxRef.current.scrollHeight; }, [lines]); + const dot = { connecting: 'var(--amber)', streaming: 'var(--cyan)', done: 'var(--green)', disconnected: 'var(--faint)' }[status]; + return ( +
+
+
+
+ + {agent.title} + {agent.kind} + 实时输出 · {status} + + +
+
+ {lines.length ? lines.map((l, i) =>
{l}
) + :
等待输出…
} +
+
+
+ ); +} + export function App() { const I18N = window.MAESTRO_I18N; const [lang, setLang] = React.useState(() => { @@ -69,6 +124,7 @@ export function App() { // ── UI state（沿用原型）───────────────────────────── const [showConfig, setShowConfig] = React.useState(false); const [archiveItem, setArchiveItem] = React.useState(null); + const [streamAgent, setStreamAgent] = React.useState(null); const [toasts, setToasts] = React.useState([]); const [theme, setTheme] = React.useState(() => localStorage.getItem('maestro-kit-theme') || 'dark'); React.useEffect(() => { @@ -253,7 +309,7 @@ export function App() { onSave={saveConfig} onSync={onSync} onClose={() => setShowConfig(false)} /> : null} - + {tasks.length ? ( @@ -268,6 +324,7 @@ export function App() { {archiveItem ? setArchiveItem(null)} /> : null} {showNewProject ? setShowNewProject(false)} /> : null} + {streamAgent ? setStreamAgent(null)} /> : null}
{toasts.map((x) => {x.text})}
diff --git a/design/ui_kits/console/Topbar.jsx b/design/ui_kits/console/Topbar.jsx index c8922ff..06af2ee 100644 --- a/design/ui_kits/console/Topbar.jsx +++ b/design/ui_kits/console/Topbar.jsx @@ -154,7 +154,7 @@ function ConfigPanel({ project, onSave, onClose, onSync, t, lang }) { ); } -function AgentSection({ agents, quota, t }) { +function AgentSection({ agents, quota, t, onOpenStream }) { const { SectionHead, QuotaMeter } = window.MaestroDesignSystem_a6a290; return (
@@ -176,7 +176,11 @@ function AgentSection({ agents, quota, t }) {
{agents.map((a) => ( -
+
onOpenStream && a.taskId && onOpenStream(a)} + title={onOpenStream ? '点击查看实时输出' : undefined} + style={{ display: 'flex', gap: 8, alignItems: 'center', fontSize: 12, padding: '3px 4px', borderRadius: 4, cursor: onOpenStream && a.taskId ? 'pointer' : 'default' }} + onMouseEnter={(e) => { if (onOpenStream && a.taskId) e.currentTarget.style.background = 'var(--panel-2)'; }} + onMouseLeave={(e) => { e.currentTarget.style.background = 'transparent'; }}> {a.title} {a.kind} diff --git a/src/api/server.ts b/src/api/server.ts index 16be41d..cda3f92 100644 --- a/src/api/server.ts +++ b/src/api/server.ts @@ -13,7 +13,8 @@ import { git, removeWorktree } from '../executor/worktree.js'; import { cleanupTaskRunArtifacts } from '../executor/cleanup.js'; import { resolveLogo, LOGO_MIME } from './logo.js'; import { readTranscript, TranscriptError } from '../executor/transcript.js'; -import { createReadStream, createWriteStream, mkdirSync } from 'node:fs'; +import { createReadStream, createWriteStream, mkdirSync, readFileSync, statSync } from 'node:fs'; +import { transcriptDir } from '../executor/cc.js'; import { homedir } from 'node:os'; import { join, basename } from 'node:path'; import { pipeline } from 'node:stream/promises'; @@ -284,6 +285,44 @@ export function buildServer(opts: ApiOptions): FastifyInstance { return store.listRuns(id); }); + // 实时流（SSE）：跟随任务「最近一条 run」的 transcript 文件，把新写入的 agent 消息逐行推送。 + // 活跃 run（status=started）= 真实时；已结束 run = 一次性回放后 end。客户端 EventSource 消费。 + app.get('/api/tasks/:id/stream', (req, reply) => { + const { id } = req.params as { id: string }; + const runs = store.listRuns(id); + const run = runs.find((r) => r.status === 'started') ?? runs[runs.length - 1]; + if (!run) return reply.code(404).send({ error: '该任务暂无 run 可流式' }); + const file = join(transcriptDir(), `${run.id}.jsonl`); + reply.hijack(); + reply.raw.writeHead(200, { + 'Content-Type': 'text/event-stream', 'Cache-Control': 'no-cache', + Connection: 'keep-alive', 'X-Accel-Buffering': 'no', + }); + reply.raw.write(`event: open\ndata: {"runId":"${run.id}","kind":"${run.kind}"}\n\n`); + let offset = 0; + let closed = false; + const stop = () => { if (closed) return; closed = true; clearInterval(timer); reply.raw.end(); }; + const tick = (): void => { + if (closed) return; + try { + const size = statSync(file).size; + if (size > offset) { + const chunk = readFileSync(file).subarray(offset).toString('utf8'); + offset = size; + for (const line of chunk.split('\n')) if (line.trim()) reply.raw.write(`data: ${line}\n\n`); + } + } catch { /* 文件尚未创建：等下一拍 */ } + const cur = store.getRun(run.id); + if (!cur || cur.status !== 'started') { + reply.raw.write('event: end\ndata: {}\n\n'); + stop(); + } + }; + const timer = setInterval(tick, 500); + tick(); + req.raw.on('close', stop); + }); + // 执行转录回放：读 run.transcriptRef 指向的 jsonl，逐行解析成脱敏后的结构化消息数组。 // 校验：runId 合法 → 关联任务/项目存在（可访问）→ 有转录记录 → 路径在 transcriptDir 下（防穿越）。 app.get('/api/runs/:id/transcript', (req, reply) => {