Merge branch 'dhk-1' into 'master'

多智能体会商沙箱

See merge request knowledge-base-bj/deerflow-server!67

frontend-web/src/roundtable-planning/api/multi-agent.ts

@@ -260,6 +260,69 @@ export interface StreamMultiAgentResult {
   clarification: ClarificationPayload | null;
 }
 
+/**
+ * Best-effort partial JSON parser for accumulated `tool_call_chunks.args`.
+ *
+ * LangChain emits tool args as a series of string fragments (one delta per
+ * chunk). To get progressive `args.path` / `args.content` during streaming —
+ * the way `useStream` does in the main chat — we concatenate the deltas and
+ * try to parse. When the JSON is still mid-token (e.g. `{"path":"/mnt`) we
+ * close any open string + brackets and retry, so callers can read partially
+ * available keys even before the full payload arrives.
+ *
+ * Returns null when nothing useful can be extracted yet.
+ */
+function parsePartialJson(s: string): Record<string, unknown> | null {
+  if (!s) return null;
+  try {
+    const v = JSON.parse(s);
+    return v && typeof v === "object" ? (v as Record<string, unknown>) : null;
+  } catch {
+    // fall through to best-effort close + retry
+  }
+  let trimmed = s;
+  // A dangling `\` would turn our injected closing `"` into an escape; drop it.
+  if (trimmed.endsWith("\\")) trimmed = trimmed.slice(0, -1);
+  let inString = false;
+  let escaped = false;
+  const stack: string[] = [];
+  for (const ch of trimmed) {
+    if (escaped) {
+      escaped = false;
+      continue;
+    }
+    if (ch === "\\") {
+      escaped = true;
+      continue;
+    }
+    if (inString) {
+      if (ch === '"') inString = false;
+      continue;
+    }
+    if (ch === '"') {
+      inString = true;
+      continue;
+    }
+    if (ch === "{") stack.push("}");
+    else if (ch === "[") stack.push("]");
+    else if (ch === "}" || ch === "]") stack.pop();
+  }
+  let closed = trimmed;
+  if (inString) {
+    closed += '"';
+  } else {
+    // Mid-key / dangling separator (`{"path":`) can't be balanced cleanly.
+    closed = closed.replace(/[,:]\s*$/, "");
+  }
+  while (stack.length) closed += stack.pop();
+  try {
+    const v = JSON.parse(closed);
+    return v && typeof v === "object" ? (v as Record<string, unknown>) : null;
+  } catch {
+    return null;
+  }
+}
+
 /**
  * Identify the final status frame our backend appends after the LangGraph
  * stream. Required keys are `status` (array for leader, string for special)
@@ -394,6 +457,61 @@ export async function streamMultiAgent(
     req.onPhaseChange?.(next, name, args);
   };
 
+  /**
+   * 累积 tool_call_chunks(delta args 字符串) —— 与 LangChain `AIMessageChunk`
+   * 合并的内部行为对齐,补主聊天靠 `useStream` 暗中做的那部分工作。Key 是 chunk.index
+   * (并发 tool 调用通常 0/1/2 区分);chunk.id 变化时视作"同一 index 上的新调用"并重置。
+   * 每帧累积后用 `parsePartialJson` 尽力 parse,成功就把合成的 `tool_calls[]` 注入
+   * 到 messages-tuple 帧的 chunkObj 上,让上层 `onUpstreamFrame` 和 `onPhaseChange`
+   * 都能拿到完整的 `args.path` / `args.content` 来做"流式期间打开沙箱 + 实时预览"。
+   */
+  const toolCallAcc = new Map<
+    number,
+    { id?: string; name: string; argsStr: string }
+  >();
+
+  type SyntheticToolCall = {
+    id?: string;
+    name: string;
+    args: Record<string, unknown>;
+  };
+
+  const ingestToolCallChunks = (chunks: unknown): void => {
+    if (!Array.isArray(chunks)) return;
+    for (const tcc of chunks) {
+      if (!tcc || typeof tcc !== "object") continue;
+      const tccObj = tcc as Record<string, unknown>;
+      const idx = typeof tccObj.index === "number" ? tccObj.index : 0;
+      let entry = toolCallAcc.get(idx);
+      if (!entry) {
+        entry = { name: "", argsStr: "" };
+        toolCallAcc.set(idx, entry);
+      }
+      if (typeof tccObj.id === "string" && tccObj.id) {
+        if (entry.id && entry.id !== tccObj.id) {
+          // 同一 index 上换了新 tool_call,重置累积(不太常见,defensive)。
+          entry.name = "";
+          entry.argsStr = "";
+        }
+        entry.id = tccObj.id;
+      }
+      if (typeof tccObj.name === "string" && tccObj.name) entry.name = tccObj.name;
+      if (typeof tccObj.args === "string") entry.argsStr += tccObj.args;
+    }
+  };
+
+  const buildSyntheticToolCalls = (): SyntheticToolCall[] => {
+    const out: SyntheticToolCall[] = [];
+    for (const entry of toolCallAcc.values()) {
+      if (!entry.name) continue;
+      const parsed = parsePartialJson(entry.argsStr);
+      if (parsed) {
+        out.push({ id: entry.id, name: entry.name, args: parsed });
+      }
+    }
+    return out;
+  };
+
   while (true) {
     const { done, value } = await reader.read();
     if (done) break;
@@ -441,26 +559,51 @@ export async function streamMultiAgent(
             const toolCalls = chunkObj.tool_calls;
             const toolCallChunks = chunkObj.tool_call_chunks;
             const content = chunkObj.content;
+
+            // 1) 累积本帧的 tool_call_chunks delta(若有),再合成可能可读的 tool_calls。
+            //    主聊天 useStream 在 SDK 层做了这件事,我们在自定义 SSE 里手动补上。
+            ingestToolCallChunks(toolCallChunks);
+            const hasCompleteToolCalls =
+              Array.isArray(toolCalls) && toolCalls.length > 0;
+            let synthetic: SyntheticToolCall[] = [];
+            if (!hasCompleteToolCalls && toolCallAcc.size > 0) {
+              synthetic = buildSyntheticToolCalls();
+              if (synthetic.length > 0) {
+                // 把累积出的 tool_calls 注入回帧的 chunkObj —— 这样下面 onUpstreamFrame
+                // 那边的 handleUpstreamFrame 只读 tool_calls 字段就能拿到完整 args,
+                // 不必再单独做一份累积。
+                chunkObj.tool_calls = synthetic;
+              }
+            }
+
+            const effectiveToolCalls: unknown[] | null = hasCompleteToolCalls
+              ? (toolCalls as unknown[])
+              : synthetic.length > 0
+                ? synthetic
+                : null;
+
             const hasToolCall =
-              (Array.isArray(toolCalls) && toolCalls.length > 0) ||
+              (effectiveToolCalls !== null && effectiveToolCalls.length > 0) ||
               (Array.isArray(toolCallChunks) && toolCallChunks.length > 0);
+
             if (chunkType === "tool") {
               setPhase("tool_result");
             } else if (hasToolCall) {
-              // Pick the most informative name we can find in either field.
-              // 优先用 tool_calls(完整对象,含解析好的 args),tool_call_chunks 的 args
-              // 通常是局部 JSON 字符串(LangGraph 流式期间逐字符发),没法直接用。
-              const firstCall = Array.isArray(toolCalls) && toolCalls.length > 0
-                ? (toolCalls[0] as Record<string, unknown>)
-                : Array.isArray(toolCallChunks) && toolCallChunks.length > 0
-                  ? (toolCallChunks[0] as Record<string, unknown>)
-                  : null;
-              const toolName = firstCall && typeof firstCall.name === "string" ? firstCall.name : "";
-              // 仅当来自 tool_calls(完整对象)时透传 args;tool_call_chunks 的 args 是
-              // string 拼接中,不传上去给 UI,等下一个完整 chunk 再带。
-              const fromCompleteToolCalls = Array.isArray(toolCalls) && toolCalls.length > 0;
+              // Pick the most informative name + args we have so far.
+              // 优先用累积出的 / 完整的 tool_calls(含可读 args 对象);若都还不到火候,
+              // 回退到 tool_call_chunks 拿名字(args 还是 string,不透传)。
+              const firstCall =
+                effectiveToolCalls && effectiveToolCalls.length > 0
+                  ? (effectiveToolCalls[0] as Record<string, unknown>)
+                  : Array.isArray(toolCallChunks) && toolCallChunks.length > 0
+                    ? (toolCallChunks[0] as Record<string, unknown>)
+                    : null;
+              const toolName =
+                firstCall && typeof firstCall.name === "string" ? firstCall.name : "";
               const args =
-                fromCompleteToolCalls && firstCall && typeof firstCall.args === "object" && firstCall.args !== null
+                firstCall &&
+                typeof firstCall.args === "object" &&
+                firstCall.args !== null
                   ? (firstCall.args as Record<string, unknown>)
                   : undefined;
               setPhase("tool_calling", toolName, args);

frontend-web/src/roundtable-planning/hooks/useStep2Orchestration.ts

@@ -223,11 +223,21 @@ export function useStep2Orchestration(
     select: selectArtifact,
     setOpen: setArtifactsOpen,
     deselect: deselectArtifact,
+    autoOpen,
+    autoSelect,
   } = useArtifacts();
   const artifactPathsRef = useRef<string[]>([]);
   useEffect(() => {
     artifactPathsRef.current = artifacts;
   }, [artifacts]);
+  // 用户手动关掉沙箱后,autoOpen 会被 ArtifactsProvider 置 false —— 把它们打到 ref
+  // 上,让 openSeatVirtualArtifact 不用进 useCallback deps,避免每次 toggle 重建函数。
+  const autoOpenFlagsRef = useRef({ autoOpen, autoSelect });
+  useEffect(() => {
+    autoOpenFlagsRef.current = { autoOpen, autoSelect };
+  }, [autoOpen, autoSelect]);
+  /** 已自动打开过的 virtualUrl 集合 —— 流式期间一秒会回调几十次,防止重复 schedule 开沙箱。 */
+  const hasAutoOpenedRef = useRef<Set<string>>(new Set());
 
   // ─── State ─────────────────────────────────────────────────────────
   const [activeArtifactThreadId, setActiveArtifactThreadId] = useState<string | null>(
@@ -381,9 +391,20 @@ export function useStep2Orchestration(
   /**
    * 用「虚拟 URL」(write-file: scheme)打开沙箱 —— 内容来自 stub messages 的 args.content,
    * 不污染 `artifacts[]` 真实路径列表(列表后续可由 present_files / 显式 setArtifacts 填充)。
+   *
+   * auto=true 流式期间:
+   *  - 已自动打开过该 virtualUrl 就直接跳过(防止 onPhaseChange / onUpstreamFrame 每帧重复 schedule);
+   *  - 用户手动关闭过沙箱(`autoOpen` 已被 ArtifactsProvider 置 false)也尊重决定,不再自动开。
+   * auto=false(用户点击步骤 / 文件卡)始终强制打开。
    */
   const openSeatVirtualArtifact = useCallback(
     (seatThreadId: string, virtualUrl: string, auto: boolean) => {
+      if (auto) {
+        if (hasAutoOpenedRef.current.has(virtualUrl)) return;
+        const flags = autoOpenFlagsRef.current;
+        if (!flags.autoOpen || !flags.autoSelect) return;
+        hasAutoOpenedRef.current.add(virtualUrl);
+      }
       setActiveArtifactThreadId(seatThreadId);
       const open = () => {
         selectArtifact(virtualUrl, auto);
@@ -625,7 +646,9 @@ export function useStep2Orchestration(
           const steps: Step2Dialogue["steps"] = d.steps ?? [];
           const last = steps[steps.length - 1];
           if (last && last.key === key) {
-            if (args && !last.args) {
+            // 同 phase:tool 重复回调 —— 流式累积期 args 会持续长大(present_files
+            // 多文件分批、write_file 内容增长),拿最新 args 覆盖,不要早早定死。
+            if (args && args !== last.args) {
               const next = [...steps];
               next[next.length - 1] = { ...last, args };
               return { ...d, steps: next };
@@ -804,11 +827,15 @@ export function useStep2Orchestration(
           safetyCounter += 1;
           highlightSpeaker(2);
 
+          // 总控也可能直接 write_file —— 之前只接了 special 流,导致总控写文件时沙箱不自动打开。
+          // 这里把 leader thread id 绑到气泡 + artifact handler 上,与 special 同样的行为。
+          const leaderThreadId = resolveSeatThreadId(ids, coord);
           const leaderBubbleId = appendStreamingDialogue({
             sender: "总控协调",
             avatarType: "blue",
             initialContent: "",
             confidence: "调度中",
+            agentThreadId: leaderThreadId,
           });
 
           const leaderAborter = new AbortController();
@@ -832,7 +859,10 @@ export function useStep2Orchestration(
                   `${phase}:${toolName ?? ""}`,
                   args,
                 );
+                handleArtifactToolPhase(phase, toolName, args, leaderThreadId);
               },
+              onUpstreamFrame: (frame) =>
+                handleUpstreamFrame(frame, leaderThreadId),
             });
           } catch (err) {
             activeAbortersRef.current.delete(leaderAborter);
@@ -1281,6 +1311,8 @@ export function useStep2Orchestration(
       const restoredStubs = snapshot.seatStubMessages ?? {};
       seatStubMessagesRef.current = restoredStubs;
       setSeatStubMessages(restoredStubs);
+      // 草稿里历史 write_file 的虚拟 URL 与当前会话无关 —— 清空守卫,允许新 run 触发自动打开。
+      hasAutoOpenedRef.current = new Set();
       // 调用方根据 furthestStep 决定是否置 init 闸门为已 init;这里默认置上,
       // 调用方需要的话再 resetInitGate()。
       step2InitRef.current = true;
@@ -1307,6 +1339,7 @@ export function useStep2Orchestration(
     setArtifacts([]);
     seatStubMessagesRef.current = {};
     setSeatStubMessages({});
+    hasAutoOpenedRef.current = new Set();
     pendingInjectPromptRef.current = null;
     deselectArtifact();
     step2InitRef.current = false;