diff --git a/web/src/api/taskStream.test.ts b/web/src/api/taskStream.test.ts new file mode 100644 index 00000000..8a992309 --- /dev/null +++ b/web/src/api/taskStream.test.ts @@ -0,0 +1,59 @@ +import test from 'node:test'; +import assert from 'node:assert/strict'; +import { buildTaskRequest, buildResumeRequest } from './taskStream'; + +test('buildTaskRequest: maps five project fields + task_key + user identity', () => { + const req = buildTaskRequest({ + taskKey: 'feasibility_reg', + projectName: '超声主机系统', + customerName: '北京水木东方', + deviceType: '超声吸引刀', + intendedUse: '外科手术导航', + targetMarket: '国内', + userId: 1, + projectId: 7, + }); + assert.equal(req.task_key, 'feasibility_reg'); + assert.equal(req.project_name, '超声主机系统'); + assert.equal(req.customer_name, '北京水木东方'); + assert.equal(req.device_type, '超声吸引刀'); + assert.equal(req.intended_use, '外科手术导航'); + assert.equal(req.target_market, '国内'); + assert.equal(req.user_id, 1); + assert.equal(req.project_id, 7); + assert.equal('work_dir' in req, false); +}); + +test('buildTaskRequest: omits optional fields when absent', () => { + const req = buildTaskRequest({ + taskKey: 'intake', + projectName: 'P', + customerName: 'C', + deviceType: 'D', + }); + assert.equal('intended_use' in req, false); + assert.equal('target_market' in req, false); + assert.equal('user_id' in req, false); + assert.equal('project_id' in req, false); +}); + +test('buildResumeRequest: adds work_dir on top of task request', () => { + const req = buildResumeRequest( + { taskKey: 'feasibility_reg', projectName: 'P', customerName: 'C', deviceType: 'D', userId: 2 }, + '/tmp/ws-x' + ); + assert.equal(req.work_dir, '/tmp/ws-x'); + assert.equal(req.task_key, 'feasibility_reg'); + assert.equal(req.user_id, 2); + assert.equal('add_steps' in req, false); +}); + +test('buildResumeRequest: forwards add_steps when provided', () => { + const req = buildResumeRequest( + { taskKey: 'intake', projectName: 'P', customerName: 'C', deviceType: 'D' }, + '/tmp/ws-y', + [{ id: 'step_dyn_1', role: 'regulatory', title: '补充条款' }] + ); + assert.equal(req.work_dir, '/tmp/ws-y'); + assert.deepEqual(req.add_steps, [{ id: 'step_dyn_1', role: 'regulatory', title: '补充条款' }]); +}); \ No newline at end of file diff --git a/web/src/api/taskStream.ts b/web/src/api/taskStream.ts new file mode 100644 index 00000000..e361b867 --- /dev/null +++ b/web/src/api/taskStream.ts @@ -0,0 +1,85 @@ +import { parseSSEChunk, type StreamEvent } from './chatStream'; + +/** 长任务运行上下文:触发 / 重试 / 重新开始时重发请求所需的最小参数集 */ +export type TaskRunContext = { + taskKey: string; + projectName: string; + customerName: string; + deviceType: string; + intendedUse?: string; + targetMarket?: string; + userId?: number; + projectId?: number; +}; + +/** + * streamTask 消费 POST /api/v1/tasks/run/stream 的 SSE 事件流。 + * 复用 chatStream 的 parseSSEChunk 解析与中止语义。 + */ +export async function streamTask( + body: Record, + handlers: { + onEvent: (ev: StreamEvent) => void; + onRequestId?: (id: string) => void; + }, + signal?: AbortSignal, +): Promise { + const resp = await fetch('/api/v1/tasks/run/stream', { + method: 'POST', + headers: { 'Content-Type': 'application/json', Accept: 'text/event-stream' }, + body: JSON.stringify(body), + signal, + }); + const rid = resp.headers.get('X-Request-Id') || resp.headers.get('x-request-id'); + if (rid) handlers.onRequestId?.(rid); + if (!resp.ok || !resp.body) { + throw Object.assign(new Error(`HTTP ${resp.status}`), { + httpStatus: resp.status, + requestId: rid, + }); + } + const reader = resp.body.getReader(); + const decoder = new TextDecoder(); + let buf = ''; + while (true) { + const { done, value } = await reader.read(); + if (done) break; + buf += decoder.decode(value, { stream: true }); + const parsed = parseSSEChunk(buf); + buf = parsed.rest; + for (const ev of parsed.events) handlers.onEvent(ev); + } + if (buf) { + const flushed = parseSSEChunk(buf, { flush: true }); + for (const ev of flushed.events) handlers.onEvent(ev); + } +} + +/** 构造新任务运行请求(五要素 + task_key + 用户/项目标识) */ +export function buildTaskRequest(task: TaskRunContext): Record { + const body: Record = { + task_key: task.taskKey, + project_name: task.projectName, + customer_name: task.customerName, + device_type: task.deviceType, + }; + if (task.intendedUse) body.intended_use = task.intendedUse; + if (task.targetMarket) body.target_market = task.targetMarket; + if (task.userId) body.user_id = task.userId; + if (task.projectId) body.project_id = task.projectId; + return body; +} + +/** 构造 resume 请求:在任务请求基础上追加 work_dir(可选动态步骤),交由后端续跑既有工作区 */ +export function buildResumeRequest( + task: TaskRunContext, + workDir: string, + addSteps?: Array>, +): Record { + const body = buildTaskRequest(task); + body.work_dir = workDir; + if (addSteps && addSteps.length > 0) { + body.add_steps = addSteps; + } + return body; +} \ No newline at end of file diff --git a/web/src/engine/phaseEventMapper.test.ts b/web/src/engine/phaseEventMapper.test.ts new file mode 100644 index 00000000..e46e897b --- /dev/null +++ b/web/src/engine/phaseEventMapper.test.ts @@ -0,0 +1,90 @@ +import test from 'node:test'; +import assert from 'node:assert/strict'; +import { applyPhaseEvent } from './phaseEventMapper'; + +test('applyPhaseEvent: appends first phase as running node', () => { + const strip = applyPhaseEvent([], { phase_id: 'squad_assembly', message: '组建小队' }); + assert.equal(strip.length, 1); + assert.equal(strip[0].step, 'squad_assembly'); + assert.equal(strip[0].icon, '👥'); + assert.equal(strip[0].status, 'running'); +}); + +test('applyPhaseEvent: completes predecessors when a new phase arrives', () => { + let strip = applyPhaseEvent([], { phase_id: 'squad_assembly', message: 'a' }); + strip = applyPhaseEvent(strip, { phase_id: 'knowledge_retrieval', message: 'b' }); + assert.equal(strip.length, 2); + assert.equal(strip[0].status, 'completed'); + assert.equal(strip[1].status, 'running'); +}); + +test('applyPhaseEvent: same phase refreshes label in place (no duplicate node)', () => { + let strip = applyPhaseEvent([], { phase_id: 'quality_gate', message: 'x', progress: 85 }); + strip = applyPhaseEvent(strip, { phase_id: 'quality_gate', message: 'y', progress: 95 }); + assert.equal(strip.length, 1); + assert.match(strip[0].label, /95%/); +}); + +test('applyPhaseEvent: unknown phase falls back to hourglass meta', () => { + const strip = applyPhaseEvent([], { phase_id: 'mystery_phase' }); + assert.equal(strip[0].icon, '⏳'); + assert.equal(strip[0].label, '任务推进'); +}); + +test('applyPhaseEvent: prefers backend message over short label', () => { + const strip = applyPhaseEvent([], { phase_id: 'squad_assembly', message: '正在组建真实小队...' }); + assert.equal(strip[0].icon, '👥'); + assert.equal(strip[0].label, '正在组建真实小队...'); +}); + +test('applyPhaseEvent: step_completed closes running entry without appending', () => { + let strip = applyPhaseEvent([], { phase_id: 'squad_assembly', message: 'a' }); + strip = applyPhaseEvent(strip, { phase_id: 'step_executing', message: '撰写中' }); + strip = applyPhaseEvent(strip, { phase_id: 'step_completed', message: '完成' }); + assert.equal(strip.length, 2); + assert.equal(strip[0].status, 'completed'); + assert.equal(strip[1].status, 'completed'); + assert.match(strip[1].label, /撰写中/); +}); + +test('applyPhaseEvent: stage_completed closes entries', () => { + let strip = applyPhaseEvent([], { phase_id: 'stage_started', message: 's' }); + strip = applyPhaseEvent(strip, { phase_id: 'stage_completed', message: 'done' }); + assert.equal(strip.length, 1); + assert.equal(strip[0].status, 'completed'); +}); + +test('applyPhaseEvent: task_failed appends warning node', () => { + let strip = applyPhaseEvent([], { phase_id: 'step_executing', message: 'e' }); + strip = applyPhaseEvent(strip, { phase_id: 'task_failed', message: 'boom' }); + assert.equal(strip.length, 2); + assert.equal(strip[0].status, 'completed'); + assert.equal(strip[1].step, 'task_failed'); + assert.equal(strip[1].icon, '⚠️'); + assert.equal(strip[1].status, 'running'); +}); + +test('applyPhaseEvent: decorates label with step index / total steps', () => { + const strip = applyPhaseEvent([], { + phase_id: 'step_executing', + message: 'm', + step_index: 2, + total_steps: 3, + }); + assert.match(strip[0].label, /第 2\/3 步/); +}); + +test('applyPhaseEvent: repeated step_executing with step_index gets separate nodes', () => { + let strip = applyPhaseEvent([], { phase_id: 'step_executing', message: '一', step_index: 1, total_steps: 3 }); + strip = applyPhaseEvent(strip, { phase_id: 'step_executing', message: '二', step_index: 2, total_steps: 3 }); + assert.equal(strip.length, 2); + assert.equal(strip[0].status, 'completed'); + assert.match(strip[0].label, /第 1\/3 步/); + assert.equal(strip[1].status, 'running'); + assert.match(strip[1].label, /第 2\/3 步/); +}); + +test('applyPhaseEvent: ignores event without phase payload', () => { + const strip = applyPhaseEvent([], undefined); + assert.equal(strip.length, 0); +}); \ No newline at end of file diff --git a/web/src/engine/phaseEventMapper.ts b/web/src/engine/phaseEventMapper.ts new file mode 100644 index 00000000..f81d8715 --- /dev/null +++ b/web/src/engine/phaseEventMapper.ts @@ -0,0 +1,80 @@ +import type { PhaseEvent } from '../types/workbench'; + +/** 后端 phase 事件载荷(与 StreamEvent['phase'] 结构兼容,保持 mapper 独立可测) */ +export type PhaseEventPayload = { + phase_id?: string; + message?: string; + role?: string; + step_index?: number; + total_steps?: number; + progress?: number; +}; + +/** phase_id → 图标/短标签映射(与 bizdev-cli/main.go 的 CLI 映射保持一致) */ +const PHASE_META: Record = { + squad_assembly: { icon: '👥', label: '组建小队' }, + knowledge_retrieval: { icon: '📚', label: '知识检索' }, + knowledge_retrieved: { icon: '📚', label: '知识命中' }, + step_executing: { icon: '⚙️', label: '专员执行' }, + step_completed: { icon: '✅', label: '步骤完成' }, + quality_gate: { icon: '🛡️', label: '质量门禁' }, + advisor_briefing: { icon: '💡', label: '参谋简报' }, + stage_transition: { icon: '🚀', label: '阶段跃迁' }, + stage_started: { icon: '🚀', label: '阶段启动' }, + stage_completed: { icon: '🎉', label: '阶段完成' }, + task_failed: { icon: '⚠️', label: '任务异常' }, +}; + +/** 收口型 phase:不追加新节点,仅把当前 running 条目置为 completed */ +const CLOSE_PHASES = new Set(['step_completed', 'stage_completed']); + +function decorateLabel(base: string, phase: PhaseEventPayload): string { + const { step_index, total_steps, progress } = phase; + if (step_index && step_index > 0 && total_steps && total_steps > 0) { + return `${base}(第 ${step_index}/${total_steps} 步)`; + } + if (progress && progress > 0) { + return `${base}(${progress}%)`; + } + return base; +} + +/** 条目 key:同 phase 且无有效 step_index 时刷新文案;带 step_index 的重复 phase 各占一节点 */ +function stepKeyOf(phase: PhaseEventPayload): string { + const idx = phase.step_index ?? 0; + return idx > 0 ? `${phase.phase_id}#${idx}` : `${phase.phase_id}`; +} + +/** + * applyPhaseEvent 把一条后端 phase 事件折叠进 PhaseEventStrip 序列(纯函数)。 + * 进度求真:仅由真实 phase 事件驱动,无任何人为进度推演。 + */ +export function applyPhaseEvent(strip: PhaseEvent[], phase?: PhaseEventPayload): PhaseEvent[] { + if (!phase?.phase_id) return strip; + + const meta = PHASE_META[phase.phase_id] ?? { icon: '⏳', label: '任务推进' }; + // 进度求真:优先展示后端下发的真实进度文案,短标签仅作缺省 + const base = phase.message?.trim() || meta.label; + const label = decorateLabel(base, phase); + const key = stepKeyOf(phase); + + // 已存在的条目:仅刷新文案(同阶段/同步骤进度更新) + const existingIdx = strip.findIndex((e) => e.step === key); + if (existingIdx >= 0) { + return strip.map((e, i) => (i === existingIdx ? { ...e, label } : e)); + } + + // 收口型 phase:running 条目置 completed,不追加新节点 + if (CLOSE_PHASES.has(phase.phase_id)) { + return strip.map((e) => (e.status === 'running' ? { ...e, status: 'completed' as const } : e)); + } + + // 新 phase:先行条目全部置 completed,追加当前节点为 running + const settled = strip.map((e) => + e.status === 'running' ? { ...e, status: 'completed' as const } : e + ); + return [ + ...settled, + { step: key, icon: meta.icon, label, status: 'running' as const }, + ]; +} \ No newline at end of file