feat(web): asset upload, task run observer, and session lineage on tasks

Upload binary attachments as project assets, subscribe by run_id after refresh,
and pass session_id/project_id through long-task requests.

Co-authored-by: Cursor <cursoragent@cursor.com>

web/src/App.tsx

@@ -17,6 +17,7 @@ import {
   unshareFreeformDoc,
   deleteFreeformDoc,
 } from './api/freeformDoc';
+import { fetchChatSessions, upsertChatSession } from './api/chatSessions';
 import { createProject, fetchProjectLineage, fetchProjects, getProject } from './api/projectWorkflow';
 import { DocumentLineage } from './types/projectWorkflow';
 import { WorkflowPrerequisiteBanner } from './components/workflow/WorkflowPrerequisiteBanner';
@@ -39,6 +40,7 @@ import { DevRoleSwitcher } from './components/dev/DevRoleSwitcher';
 import { authClient } from './api/auth';
 import { streamChat, mapErrorInfo } from './api/chatStream';
 import { streamTask, buildTaskRequest, buildResumeRequest } from './api/taskStream';
+import type { StreamEvent } from './api/chatStream';
 import { shouldAutoRetryChat, shouldAutoRetryTask, autoRetryDelayMs, fallbackErrorClass } from './api/chatRetry';
 import { applyPhaseEvent } from './engine/phaseEventMapper';
 import { LoginResult, SystemAuthConfig } from './types/auth';
@@ -121,6 +123,7 @@ import { fetchChatTurnStatus } from './api/chatTurn';
 import { createClientTurnId } from './lib/createClientTurnId';
 import { mergeSessionUser } from './lib/mergeSessionUser';
 import { mapServerProjectToProjectInfo, mergeProjectLists } from './lib/projectAdapter';
+import { mapServerSessionToProjectSession, mergeSessionLists } from './lib/sessionAdapter';
 import { SendMessageResult } from './types/sendMessage';
 import { CategoryFilterBar } from './components/history/CategoryFilterBar';
 import { HistoryCalendar } from './components/history/HistoryCalendar';
@@ -133,40 +136,50 @@ import { CategoryDock } from './components/chat/CategoryDock';
 import { fetchChatCategories, fetchChatDates } from './api/chatWindow';
 import { ChatCategoryStat, ChatDateStat, QuotedMessage } from './types/workbench';
 import { saveFile, restoreFileRawToDeliverables } from './data/fileStore';
+import { uploadProjectAsset } from './api/projectAssets';
+import { subscribeTaskRunEvents, createTaskRun, cancelTaskRun } from './api/taskRuns';
 
 /**
- * 将用户上传附件列表转换为任务输入材料（TaskRunMaterial[]），
- * 文本类格式（.txt .md .csv .json .xml）读取全文后 Base64 编码；
- * 其余格式仅传文件名作来源注记，content 为空字符串。
+ * 将附件转为任务材料：二进制先上传 CAS 登记 asset_id；小文本可 inline Base64。
  */
 async function readAttachmentsAsMaterials(
-  attachments: { name: string; type: string; docId?: string; fileRaw?: File | Blob }[]
+  attachments: { name: string; type: string; docId?: string; fileRaw?: File | Blob; assetId?: number }[],
+  projectId?: number,
+  userId?: number,
 ): Promise<TaskRunMaterial[]> {
   const TEXT_EXTS = new Set(['txt', 'md', 'csv', 'json', 'xml', 'yaml', 'yml', 'log']);
   const results: TaskRunMaterial[] = [];
   for (const att of attachments) {
+    if (att.assetId && att.assetId > 0) {
+      results.push({ filename: att.name, asset_id: att.assetId });
+      continue;
+    }
     if (!att.fileRaw) continue;
     const extMatch = att.name.match(/\.([^.]+)$/);
     const ext = extMatch ? extMatch[1].toLowerCase() : '';
-    let content = '';
-    if (TEXT_EXTS.has(ext)) {
-      try {
-        content = await new Promise<string>((resolve, reject) => {
-          const reader = new FileReader();
-          reader.onload = () => {
-            const text = reader.result as string;
-            // btoa 仅支持 Latin1，需先 encodeURIComponent 转义 Unicode
-            const b64 = btoa(unescape(encodeURIComponent(text)));
-            resolve(b64);
-          };
-          reader.onerror = () => reject(reader.error);
-          reader.readAsText(att.fileRaw as Blob, 'utf-8');
-        });
-      } catch {
-        content = '';
+    if (!TEXT_EXTS.has(ext)) {
+      if (!projectId || projectId <= 0) {
+        throw new Error(`附件 ${att.name} 需要先登记为项目资产（缺少 project_id）`);
       }
+      const uploaded = await uploadProjectAsset(projectId, att.fileRaw as Blob, att.name, userId);
+      results.push({ filename: uploaded.filename || att.name, asset_id: uploaded.asset_id });
+      continue;
+    }
+    try {
+      const content = await new Promise<string>((resolve, reject) => {
+        const reader = new FileReader();
+        reader.onload = () => {
+          const text = reader.result as string;
+          const b64 = btoa(unescape(encodeURIComponent(text)));
+          resolve(b64);
+        };
+        reader.onerror = () => reject(reader.error);
+        reader.readAsText(att.fileRaw as Blob, 'utf-8');
+      });
+      results.push({ filename: att.name, content });
+    } catch {
+      throw new Error(`读取文本附件 ${att.name} 失败`);
     }
-    results.push({ filename: att.name, content });
   }
   return results;
 }
@@ -563,6 +576,7 @@ export default function App() {
   const [createProjectModalOpen, setCreateProjectModalOpen] = useState(false);
   const [createProjectError, setCreateProjectError] = useState<string | null>(null);
   const [projectSyncNotice, setProjectSyncNotice] = useState<string | null>(null);
+  const [sessionSyncNotice, setSessionSyncNotice] = useState<string | null>(null);
   const [projectMembersModalOpen, setProjectMembersModalOpen] = useState(false);
 
   const isAdmin = currentUser.roleId === 'role-admin';
@@ -676,6 +690,42 @@ export default function App() {
     };
   }, [session, allUsers]);
 
+  useEffect(() => {
+    if (!session || !currentUser.id) {
+      setSessionSyncNotice(null);
+      return;
+    }
+
+    let cancelled = false;
+    void (async () => {
+      try {
+        const serverSessions = await fetchChatSessions(currentUser.id);
+        if (cancelled) return;
+        setSessions((prev) => {
+          const mapped = serverSessions.map((item) =>
+            mapServerSessionToProjectSession(
+              item,
+              projects,
+              prev.find((sessionItem) => sessionItem.sessionId === item.session_id)
+            )
+          );
+          return mergeSessionLists(prev, mapped);
+        });
+        setSessionSyncNotice(null);
+      } catch (err) {
+        if (cancelled) return;
+        console.warn('[sessions] 服务端会话列表同步失败，继续使用本地缓存:', err);
+        setSessionSyncNotice(
+          '会话列表暂用本地缓存：无法连接服务端，新建会话可能无法在其他设备同步。'
+        );
+      }
+    })();
+
+    return () => {
+      cancelled = true;
+    };
+  }, [session, currentUser.id, projects]);
+
   const handleLogout = () => {
     authClient.logout();
     setSession(null);
@@ -747,7 +797,7 @@ export default function App() {
 
   const workflowBanner = (
     <>
-      {projectSyncNotice && (
+      {(projectSyncNotice || sessionSyncNotice) && (
         <div
           style={{
             padding: '8px 12px',
@@ -759,7 +809,7 @@ export default function App() {
             fontSize: '12px',
           }}
         >
-          {projectSyncNotice}
+          {[projectSyncNotice, sessionSyncNotice].filter(Boolean).join(' ')}
         </div>
       )}
       <WorkflowPrerequisiteBanner
@@ -1970,6 +2020,7 @@ export default function App() {
   };
 
   const taskStreamAbortRef = useRef<AbortController | null>(null);
+  const activeTaskRunIdRef = useRef<string | null>(null);
 
   // P2：长任务尝试模式。'run' 新任务 | 'resume' 续跑失败工作区 | 'restart' 按原意图新工作区
   const runTaskAttempt = (
@@ -1980,6 +2031,7 @@ export default function App() {
     opts?: { phaseEvents?: PhaseEvent[]; workDir?: string }
   ) => {
     taskStreamAbortRef.current?.abort();
+    activeTaskRunIdRef.current = null;
     const controller = new AbortController();
     taskStreamAbortRef.current = controller;
 
@@ -2028,73 +2080,87 @@ export default function App() {
     });
     setTaskThinking(true);
 
-    const body =
-      mode === 'resume' && opts?.workDir
-        ? buildResumeRequest(task, opts.workDir)
-        : buildTaskRequest(task);
-
-    streamTask(
-      body,
-      {
-        onEvent: (ev) => {
-          if (controller.signal.aborted || settled) return;
-          switch (ev.type) {
-            case 'phase': {
-              // live PhaseEventStrip：由真实 phase 事件驱动，禁止假进度
-              patchAsst((m) => ({
-                status: 'streaming_status',
-                livePhaseLabel: ev.phase?.message || m.livePhaseLabel,
-                phaseEvents: applyPhaseEvent(opts?.phaseEvents || m.phaseEvents || [], ev.phase),
-              }));
-              break;
-            }
-            case 'status': {
-              if (ev.status?.message) {
-                patchAsst({ status: 'streaming_status', livePhaseLabel: ev.status.message });
-              }
-              break;
-            }
-            case 'delta': {
-              accumulated += ev.delta || '';
-              patchAsst({ status: 'streaming_status', content: accumulated });
-              break;
-            }
-            case 'deliverable': {
-              const title = ev.deliverable?.title;
-              if (title) {
-                accumulated += (accumulated ? '\n\n' : '') + `📄 已生成文档：**${title}**`;
-                patchAsst({ status: 'streaming_status', content: accumulated });
-              }
-              break;
-            }
-            case 'error': {
-              failOrRetry(mapErrorInfo(ev));
-              break;
-            }
-            case 'done': {
-              settled = true;
-              patchAsst({
-                status: 'completed',
-                livePhaseLabel: undefined,
-                taskWorkspaceId: ev.done?.workspace_id,
-              });
-              setTaskThinking(false);
-              break;
-            }
-            default:
-              break;
+    const onEvent = (ev: StreamEvent) => {
+      if (controller.signal.aborted || settled) return;
+      switch (ev.type) {
+        case 'phase': {
+          patchAsst((m) => ({
+            status: 'streaming_status',
+            livePhaseLabel: ev.phase?.message || m.livePhaseLabel,
+            phaseEvents: applyPhaseEvent(opts?.phaseEvents || m.phaseEvents || [], ev.phase),
+          }));
+          break;
+        }
+        case 'status': {
+          if (ev.status?.message) {
+            patchAsst({ status: 'streaming_status', livePhaseLabel: ev.status.message });
           }
-        },
-      },
-      controller.signal
-    ).catch((_err) => {
+          break;
+        }
+        case 'delta': {
+          accumulated += ev.delta || '';
+          patchAsst({ status: 'streaming_status', content: accumulated });
+          break;
+        }
+        case 'deliverable': {
+          const title = ev.deliverable?.title;
+          if (title) {
+            accumulated += (accumulated ? '\n\n' : '') + `📄 已生成文档：**${title}**`;
+            patchAsst({ status: 'streaming_status', content: accumulated });
+          }
+          break;
+        }
+        case 'error': {
+          failOrRetry(mapErrorInfo(ev));
+          break;
+        }
+        case 'done': {
+          settled = true;
+          activeTaskRunIdRef.current = null;
+          patchAsst({
+            status: 'completed',
+            livePhaseLabel: undefined,
+            taskWorkspaceId: ev.done?.workspace_id,
+          });
+          setTaskThinking(false);
+          break;
+        }
+        default:
+          break;
+      }
+    };
+
+    const startObserver = async () => {
+      const body =
+        mode === 'resume' && opts?.workDir
+          ? buildResumeRequest(task, opts.workDir)
+          : buildTaskRequest(task);
+
+      if (task.sessionId && task.projectId && task.projectId > 0 && mode === 'run') {
+        const created = await createTaskRun({
+          ...body,
+          session_id: task.sessionId,
+          project_id: task.projectId,
+          user_id: task.userId,
+          client_turn_id: task.clientTurnId,
+          user_message: task.userMessage,
+        });
+        activeTaskRunIdRef.current = created.run_id;
+        patchAsst({ taskRunId: created.run_id });
+        await subscribeTaskRunEvents(created.run_id, { onEvent }, controller.signal);
+        return;
+      }
+
+      await streamTask(body, { onEvent }, controller.signal);
+    };
+
+    void startObserver().catch((_err) => {
       if (settled) return;
       if (controller.signal.aborted) {
         patchAsst({ status: 'cancelled', livePhaseLabel: undefined });
         setTaskThinking(false);
         return;
       }
-      // fetch/网络层抛出：合成瞬态错误，走与后端 error 事件一致的自动重试路径
       failOrRetry(buildNetworkErrorInfo());
     });
   };
@@ -2139,6 +2205,11 @@ export default function App() {
   const handleStopStreaming = () => {
     turnRuntime.abortActiveSessionTurns();
     taskStreamAbortRef.current?.abort();
+    const runId = activeTaskRunIdRef.current;
+    if (runId) {
+      void cancelTaskRun(runId).catch(() => undefined);
+      activeTaskRunIdRef.current = null;
+    }
   };
 
   // 发送消息与对话驱动双轨逻辑
@@ -2325,13 +2396,17 @@ export default function App() {
       const asstId = 'ast-' + Date.now();
       const currentProj = sessions.find((s) => s.sessionId === activeSessionId);
       // 项目五要素暂取会话标题/设备类型；客户名沿用 bizdev-cli 演示默认值，五要素完整建模留待后续
+      const clientTurnId = `turn_${Date.now()}`;
       const baseTask: TaskRunContext = {
         taskKey: matchedTemplate.taskKey,
         projectName: currentProj?.title || '医疗器械研发演示项目',
         customerName: '北京水木东方',
         deviceType: currentProj?.deviceType || '医疗器械',
+        sessionId: activeSessionId,
         userId: currentUser.id,
         projectId: currentProj?.projectId || 1,
+        clientTurnId,
+        userMessage: text,
       };
       const placeholder: ChatMessage = {
         id: asstId,
@@ -2347,7 +2422,11 @@ export default function App() {
       setMessages((prev) => [...prev, placeholder]);
       // 异步读取附件内容后再触发任务，避免文档材料丢失
       void (async () => {
-        const materials = await readAttachmentsAsMaterials(processedAttachments || []);
+        const materials = await readAttachmentsAsMaterials(
+          processedAttachments || [],
+          baseTask.projectId,
+          baseTask.userId,
+        );
         const task: TaskRunContext = materials.length > 0 ? { ...baseTask, materials } : baseTask;
         runTaskAttempt(task, asstId, 'run', 0);
       })();
@@ -2878,6 +2957,19 @@ export default function App() {
     setProjects((prev) => [newProject, ...prev]);
     setSessions((prev) => [...newSessionsToAppend, ...prev]);
     setMessages((prev) => [...prev, ...newMessagesToAppend]);
+
+    void Promise.all(
+      newSessionsToAppend.map((sessionItem) =>
+        upsertChatSession({
+          session_id: sessionItem.sessionId,
+          title: sessionItem.title,
+          user_id: sessionItem.userId ?? currentUser.id,
+          project_id: sessionItem.projectId,
+        })
+      )
+    ).catch((err) => {
+      console.warn('[sessions] 服务端会话注册失败，其他设备可能暂不可见:', err);
+    });
     setDeliverables((prev) => [...prev, ...clonedDocs]);
     setDocCache((prev) => ({ ...prev, ...clonedAsts }));
 

web/src/api/chatWindow.ts

@@ -79,7 +79,16 @@ export function mapRecordToMessage(r: RawChatMessageRecord): ChatMessage {
     interactiveCard: meta.interactive_card,
     recommendedCases: meta.recommended_cases,
     quotes: meta.quotes || (meta.quote ? [meta.quote] : undefined),
-    status: r.role === 'assistant' ? 'completed' : undefined,
+    status:
+      r.role === 'assistant'
+        ? meta.task_status === 'running'
+          ? 'streaming_status'
+          : meta.task_status === 'failed'
+            ? 'failed'
+            : 'completed'
+        : undefined,
+    taskRunId: typeof meta.run_id === 'string' ? meta.run_id : undefined,
+    taskWorkspaceId: typeof meta.workspace_id === 'string' ? meta.workspace_id : undefined,
     contentKind: r.content_kind,
     exportCapability: r.export_capability,
   };

web/src/api/projectAssets.ts

@@ -0,0 +1,39 @@
+export type ProjectAsset = {
+  asset_id: number;
+  filename: string;
+  sha256: string;
+  mime_type?: string;
+  size_bytes?: number;
+  project_id: number;
+};
+
+export async function uploadProjectAsset(
+  projectId: number,
+  file: File | Blob,
+  filename: string,
+  userId?: number,
+): Promise<ProjectAsset> {
+  const form = new FormData();
+  form.append('file', file, filename);
+  const headers: Record<string, string> = {};
+  if (userId && userId > 0) {
+    headers['X-User-ID'] = String(userId);
+  }
+  const res = await fetch(`/api/v1/projects/${projectId}/assets`, {
+    method: 'POST',
+    headers,
+    body: form,
+  });
+  if (!res.ok) {
+    throw new Error(`upload asset failed: HTTP ${res.status}`);
+  }
+  return res.json();
+}
+
+export async function listProjectAssets(projectId: number): Promise<ProjectAsset[]> {
+  const res = await fetch(`/api/v1/projects/${projectId}/assets`);
+  if (!res.ok) {
+    throw new Error(`list assets failed: HTTP ${res.status}`);
+  }
+  return res.json();
+}

web/src/api/taskRuns.ts

@@ -0,0 +1,76 @@
+import { parseSSEChunk, type StreamEvent } from './chatStream';
+
+export type CreateTaskRunBody = Record<string, unknown> & {
+  session_id: string;
+  project_id: number;
+  user_id?: number;
+  client_turn_id?: string;
+  user_message?: string;
+};
+
+export type CreateTaskRunResponse = {
+  run_id: string;
+  status: string;
+  workspace_id?: string;
+};
+
+export async function createTaskRun(body: CreateTaskRunBody): Promise<CreateTaskRunResponse> {
+  const res = await fetch('/api/v1/tasks/runs', {
+    method: 'POST',
+    headers: { 'Content-Type': 'application/json' },
+    body: JSON.stringify(body),
+  });
+  if (!res.ok) {
+    throw new Error(`create task run failed: HTTP ${res.status}`);
+  }
+  return res.json();
+}
+
+export async function subscribeTaskRunEvents(
+  runId: string,
+  handlers: { onEvent: (ev: StreamEvent) => void; afterSeq?: number },
+  signal?: AbortSignal,
+): Promise<void> {
+  const after = handlers.afterSeq ?? 0;
+  const url = `/api/v1/tasks/runs/${encodeURIComponent(runId)}/events?after_seq=${after}`;
+  const resp = await fetch(url, {
+    method: 'GET',
+    headers: { Accept: 'text/event-stream' },
+    signal,
+  });
+  if (!resp.ok || !resp.body) {
+    throw new Error(`subscribe task run failed: HTTP ${resp.status}`);
+  }
+  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);
+  }
+}
+
+export async function cancelTaskRun(runId: string): Promise<void> {
+  const res = await fetch(`/api/v1/tasks/runs/${encodeURIComponent(runId)}/cancel`, {
+    method: 'POST',
+  });
+  if (!res.ok) {
+    throw new Error(`cancel task run failed: HTTP ${res.status}`);
+  }
+}
+
+export async function getTaskRun(runId: string) {
+  const res = await fetch(`/api/v1/tasks/runs/${encodeURIComponent(runId)}`);
+  if (!res.ok) {
+    throw new Error(`get task run failed: HTTP ${res.status}`);
+  }
+  return res.json();
+}

web/src/api/taskStream.ts

@@ -56,8 +56,10 @@ export function buildTaskRequest(task: TaskRunContext): Record<string, unknown>
   };
   if (task.intendedUse) body.intended_use = task.intendedUse;
   if (task.targetMarket) body.target_market = task.targetMarket;
+  if (task.sessionId) body.session_id = task.sessionId;
   if (task.userId) body.user_id = task.userId;
   if (task.projectId) body.project_id = task.projectId;
+  if (task.clientTurnId) body.client_turn_id = task.clientTurnId;
   if (task.materials && task.materials.length > 0) {
     body.materials = task.materials;
   }

web/src/types/workbench.ts

@@ -128,7 +128,8 @@ export interface ChatErrorInfo {
  */
 export type TaskRunMaterial = {
   filename: string;
-  content: string; // Base64 编码的文件内容
+  content?: string; // 小文本可选 Base64；二进制应使用 asset_id
+  asset_id?: number;
 };
 
 /** 长任务运行上下文：触发 / 重试 / 重新开始时重发请求所需的最小参数集 */
@@ -139,8 +140,11 @@ export type TaskRunContext = {
   deviceType: string;
   intendedUse?: string;
   targetMarket?: string;
+  sessionId?: string;
   userId?: number;
   projectId?: number;
+  clientTurnId?: string;
+  userMessage?: string;
   /** 用户上传的参考文档材料，注入任务工作区 materials_text/ 目录 */
   materials?: TaskRunMaterial[];
 };
@@ -172,6 +176,8 @@ export interface ChatMessage {
   sourceUserMessageId?: string;
   /** P2: 长任务上下文（存在即表示该气泡由 task stream 驱动） */
   taskRequest?: TaskRunContext;
+  /** 观察者模型 run_id，刷新后可重订阅 */
+  taskRunId?: string;
   /** P2: 最近一次任务成功/收尾时的工作区目录（done.workspace_id） */
   taskWorkspaceId?: string;