fix(studio): route id-less and compression events away from background task cards

三个事件路由缺陷(并行聊天与后台任务时互相串排):
- context:compression 处理器忽略服务端已带的 executionId,一律写进当前
  聊天流消息:并行时任务 pipeline 的压缩事件会在聊天消息里长出
  context-* 伪工具卡。
- 无 id 的 log / llm:progress 回退到"最近一张运行中工具卡":聊天轮还没有
  自己的工具卡时会落到后台任务卡上(聊天日志串排进任务)。
- 服务端 broadcast("tool:update") 全仓无前端消费者,是死事件。

修法:
- context:compression 带 executionId 时按 id 定位任务卡,把压缩事件作为
  阶段挂上去(不可变更新);卡不存在时丢弃,绝不写进聊天流消息;无 id
  维持现状(聊天轮自己的压缩展示)。
- 无 id 回退改为"最近一张运行中的聊天轮工具卡":跳过带 background 标记的
  任务卡;跳过后没有可挂的卡时丢弃(任务快照重放会带回任务自己的累积
  日志,不丢信息)。
- 删除两处 tool:update 死广播(确认任务 onUpdate 内与聊天轮
  tool_execution_update 转发),已 grep 确认零消费者、无测试引用。
This commit is contained in:
Ma
2026-07-15 15:26:39 +08:00
parent 29b6af4349
commit bfae2d35cf
3 changed files with 169 additions and 16 deletions
-12
View File
@@ -1674,11 +1674,6 @@ async function executeConfirmedProductionAction(args: {
const progress = toolResultText(partialResult, lang);
if (progress) exec.logs = [...(exec.logs ?? []), progress].slice(-80);
void args.onTaskChange(exec).catch(() => undefined);
broadcast("tool:update", {
sessionId: args.streamSessionId,
tool: tool.name,
partialResult,
});
},
);
// 工具可以在结果里带 isError=true 表示"执行完成但结果需要人工处理"
@@ -4965,13 +4960,6 @@ export function createStudioServer(initialConfig: ProjectConfig, root: string, o
stages,
});
}
if (event.type === "tool_execution_update") {
broadcast("tool:update", {
sessionId: streamSessionId,
tool: event.toolName,
partialResult: event.partialResult,
});
}
if (event.type === "tool_execution_end") {
const exec = collectedToolExecs.find(t => t.id === event.toolCallId);
if (exec) {
@@ -852,4 +852,121 @@ describe("chat message actions", () => {
resolveAgent({ response: "聊完了。", session: { sessionId, sessionKind: "short" } });
await sent;
});
it("drops id-less logs and progress instead of attaching them to a background task card", async () => {
const store = createTestStore();
const sessionId = await setupRunningTaskSession(store);
// 快照重放给任务卡一个 active 阶段,验证无 id 进度也不会写进去
fakeEventSources[0]?.emit("task:snapshot", {
sessionId,
execution: {
id: "direct-short_run-1",
tool: "short_fiction_run",
label: "短篇生产",
status: "running",
startedAt: 10,
stages: [{ label: "撰写正文", status: "active" }],
},
});
let resolveAgent!: (value: unknown) => void;
fetchJson.mockClear();
fetchJson.mockImplementationOnce(() => new Promise((resolve) => {
resolveAgent = resolve;
}));
const sent = store.getState().sendMessage(sessionId, "顺便聊两句");
await vi.waitFor(() => expect(fakeEventSources).toHaveLength(2));
// 聊天轮还没有自己的工具卡:无 id 的 log / llm:progress 不能回退到
// 后台任务卡(会反向串排),只能丢弃(任务快照重放会带回累积日志)。
fakeEventSources[1]?.emit("log", {
sessionId,
level: "info",
tag: "studio",
message: "游离的聊天轮日志",
});
fakeEventSources[1]?.emit("llm:progress", {
sessionId,
status: "思考中",
elapsedMs: 900,
totalChars: 120,
chineseChars: 100,
});
expect(findTaskExecution(store, sessionId)?.logs).toBeUndefined();
expect(findTaskExecution(store, sessionId)?.stages?.[0]?.progress).toBeUndefined();
// 聊天轮工具卡出现后:无 id 日志照旧落在聊天卡上,任务卡不受影响
fakeEventSources[1]?.emit("tool:start", {
sessionId,
id: "chat-tool-1",
tool: "sub_agent",
args: { agent: "auditor" },
});
fakeEventSources[1]?.emit("log", {
sessionId,
level: "info",
tag: "studio",
message: "审稿进行中",
});
expect(findChatToolExecution(store, sessionId)?.logs).toEqual(["审稿进行中"]);
expect(findTaskExecution(store, sessionId)?.logs).toBeUndefined();
resolveAgent({ response: "聊完了。", session: { sessionId, sessionKind: "short" } });
await sent;
});
it("routes task-tagged context compression to the task card and never into the chat stream", async () => {
const store = createTestStore();
const sessionId = await setupRunningTaskSession(store);
let resolveAgent!: (value: unknown) => void;
fetchJson.mockClear();
fetchJson.mockImplementationOnce(() => new Promise((resolve) => {
resolveAgent = resolve;
}));
const sent = store.getState().sendMessage(sessionId, "顺便聊两句");
await vi.waitFor(() => expect(fakeEventSources).toHaveLength(2));
const allExecutions = () => (store.getState().sessions[sessionId]?.messages ?? [])
.flatMap((message) => [
...(message.toolExecutions ?? []),
...(message.parts ?? []).flatMap((part) => (part.type === "tool" ? [part.execution] : [])),
]);
// 任务 pipeline 的压缩事件带任务 execution id:作为阶段挂到任务卡上
fakeEventSources[1]?.emit("context:compression", {
sessionId,
executionId: "direct-short_run-1",
category: "story_context",
phase: "start",
protectedTokens: 1200,
});
expect(findTaskExecution(store, sessionId)?.stages).toEqual([
expect.objectContaining({ label: "压缩故事上下文", status: "active" }),
]);
// 不产生聊天流内容:没有 context-* 伪工具卡被写进消息
expect(allExecutions().some((execution) => execution.id.startsWith("context-"))).toBe(false);
fakeEventSources[1]?.emit("context:compression", {
sessionId,
executionId: "direct-short_run-1",
category: "story_context",
phase: "end",
});
expect(findTaskExecution(store, sessionId)?.stages).toEqual([
expect.objectContaining({ label: "压缩故事上下文", status: "completed" }),
]);
// id 指向的卡不存在:事件丢弃,同样不写进聊天流
fakeEventSources[1]?.emit("context:compression", {
sessionId,
executionId: "direct-unknown-9",
category: "session_context",
phase: "start",
});
expect(allExecutions().some((execution) => execution.id.startsWith("context-"))).toBe(false);
resolveAgent({ response: "聊完了。", session: { sessionId, sessionKind: "short" } });
await sent;
});
});
@@ -27,6 +27,7 @@ type ContextCompressionPhase = "start" | "end" | "error";
interface ContextCompressionEventPayload {
readonly sessionId?: string;
readonly executionId?: string;
readonly category?: ContextCompressionCategory;
readonly phase?: ContextCompressionPhase;
readonly message?: string;
@@ -110,9 +111,27 @@ export function appendBoundedToolLogs(
}
/**
*
* streamTs
* update null no-op
* background
* id 退
*/
function findRunningChatToolPart(
parts: ReadonlyArray<MessagePart>,
): (MessagePart & { type: "tool" }) | undefined {
for (let i = parts.length - 1; i >= 0; i -= 1) {
const part = parts[i]!;
if (part.type === "tool" && part.execution.status === "running" && !part.execution.background) {
return part;
}
}
return undefined;
}
/**
*
* execution.background id
* 退 null
*
* update null no-op
*/
export function updateLatestRunningToolMessage(
messages: ReadonlyArray<Message>,
@@ -120,7 +139,7 @@ export function updateLatestRunningToolMessage(
): ReadonlyArray<Message> | null {
for (let i = messages.length - 1; i >= 0; i -= 1) {
const message = messages[i]!;
const running = findRunningToolPart([...(message.parts ?? [])]);
const running = findRunningChatToolPart(message.parts ?? []);
if (!running) continue;
const updated = update(running.execution);
if (!updated) return null;
@@ -573,8 +592,18 @@ export function attachSessionStreamListeners({
if (!sessionMatchesEvent(sessionId, data) || !data?.category || !data.phase) return;
const category = data.category;
const phase = data.phase;
const executionId = eventExecutionId(data);
set((state) => ({
sessions: updateSession(state.sessions, sessionId, (runtime) => {
// 带 executionId 的压缩事件(后台生产任务的 pipeline):作为阶段挂到
// 对应的任务卡上;卡不存在时丢弃这条(任务快照重放会带回状态),
// 绝不写进聊天流消息——并行时会把任务状态串排进聊天轮。
if (executionId) {
const messages = updateToolPartById(runtime.messages, executionId, (execution) =>
applyContextCompressionToExecution(execution, category, phase, data),
);
return messages ? { messages } : {};
}
const [messages, stream] = getOrCreateStream(runtime.messages, streamTs);
const parts = [...(stream.parts ?? [])];
applyContextCompressionToParts(parts, category, phase, data);
@@ -639,6 +668,25 @@ function findRunningExecution(parts: MessagePart[]): ToolExecution | undefined {
return running?.execution;
}
/** 按 id 定位到的任务卡:把压缩事件作为阶段挂上去(不可变更新)。 */
function applyContextCompressionToExecution(
execution: ToolExecution,
category: ContextCompressionCategory,
phase: ContextCompressionPhase,
data: ContextCompressionEventPayload,
): ToolExecution {
const stages = upsertCompressionStage(execution.stages, category, phase, data);
if (phase === "error") {
return {
...execution,
stages,
status: "error",
error: data.message ?? `${compressionLabel(category)}${tr("失败", " failed")}`,
};
}
return { ...execution, stages };
}
function applyContextCompressionToParts(
parts: MessagePart[],
category: ContextCompressionCategory,