From 16900b0a205c2c2ccbc7c10ca7dae0626bce8678 Mon Sep 17 00:00:00 2001 From: JOJO <1498581755@qq.com> Date: Wed, 12 Aug 2026 22:27:31 +0800 Subject: [PATCH] =?UTF-8?q?fix(chat):=20=E4=BF=AE=E5=A4=8D=E5=AD=90?= =?UTF-8?q?=E6=99=BA=E8=83=BD=E4=BD=93=E5=AE=8C=E6=88=90=E9=80=9A=E7=9F=A5?= =?UTF-8?q?=E4=B8=8D=E5=AE=9E=E6=97=B6=E6=98=BE=E7=A4=BA=EF=BC=8C=E5=8E=8B?= =?UTF-8?q?=E7=BC=A9=E6=AE=8B=E7=95=99=E9=94=81=E6=AD=BB=E8=BE=93=E5=85=A5?= =?UTF-8?q?=E4=B8=8E=E5=88=87=E6=8D=A2?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 通知：inject_runtime_user_message 不再把子智能体任务 id 透传为事件顶层 task_id（sender setdefault 不覆盖，前端任务归属守卫丢弃事件；socket 通道 在轮询模式下不渲染，双通道断头导致只能刷新后从历史看到）。后台指令因透传 的是 command_id 而不受影响 - 通知：修复 _persist_and_echo_preceding_notice 回显写在 except 分支内， 批量通知前 N-1 条从不实时回显 - 压缩：移除「压缩中切换/新建对话需确认并取消压缩」拦截（多对话独立运行后 无存在意义） - 压缩：新增 compressionConversationId，输入/发送/停止锁按对话隔离，/new 新建页不再被其他对话的压缩状态锁定 - 压缩：metadata 记录 compression_pid，status/compression_status 读取时按 pid 懒清理进程重启后的残留 in_progress 标记；run_deep_compression 异常时 自动清理标记 --- server/chat_flow_task_main.py | 6 +- server/chat_flow_task_support.py | 8 +- server/conversation.py | 17 ++-- server/deep_compression.py | 86 +++++++++++++++++++ server/status/base.py | 17 +++- static/src/App.vue | 2 +- static/src/app/computed.ts | 23 +++-- static/src/app/methods/conversation/action.ts | 31 +------ static/src/app/methods/conversation/load.ts | 33 +------ static/src/app/methods/message/chat.ts | 2 + static/src/app/methods/message/send.ts | 8 +- static/src/app/methods/resources.ts | 27 +++--- .../app/methods/taskPolling/compression.ts | 5 ++ static/src/app/methods/ui/review.ts | 2 +- static/src/app/state.ts | 3 + utils/context_manager/compression_mixin.py | 3 + 16 files changed, 180 insertions(+), 93 deletions(-) diff --git a/server/chat_flow_task_main.py b/server/chat_flow_task_main.py index 151761cc..b67e9281 100644 --- a/server/chat_flow_task_main.py +++ b/server/chat_flow_task_main.py @@ -495,6 +495,10 @@ def _persist_and_echo_preceding_notice( if cm is not None: cm.add_conversation("user", message, metadata=metadata) except Exception as exc: + debug_log(f"[CompletionNotice] 前置通知写入历史失败: {exc}") + # socketio 回显（在线客户端即时可见）；轮询客户端由后续任务事件流回放， + # 前端按消息内容 dedup，两条通道不会双显。 + try: echo_payload = { "message": message, "conversation_id": conversation_id, @@ -505,7 +509,7 @@ def _persist_and_echo_preceding_notice( echo_payload["starts_work"] = False echo_payload["metadata"] = {**metadata} sender("user_message", echo_payload) - except Exception as exc: + except Exception: pass async def _dispatch_completion_user_notice( diff --git a/server/chat_flow_task_support.py b/server/chat_flow_task_support.py index 80409e91..8e5d04bd 100644 --- a/server/chat_flow_task_support.py +++ b/server/chat_flow_task_support.py @@ -145,7 +145,13 @@ def inject_runtime_user_message( "runtime_guidance": True, "runtime_guidance_original": raw, } - for key in ("task_id", "command_id"): + # 关键：不能把子智能体任务 id 透传为顶层 task_id——本事件由主任务 sender + # 发出、走主任务事件流，若顶层 task_id 是子任务 id，sender 的 setdefault + # 不会覆盖，前端轮询的「任务归属守卫」（task_id !== currentTaskId 即丢弃） + # 会把这条通知扔掉；叠加 socket 通道在轮询模式下不渲染，导致运行期 inline + # 完成通知只能刷新后从历史看到（2026-08-12 定位，后台指令不受影响是因为 + # 它透传的是 command_id 而非 task_id）。子任务 id 仍保留在 metadata 中。 + for key in ("command_id",): value = (extra_metadata or {}).get(key) if value is not None: payload[key] = value diff --git a/server/conversation.py b/server/conversation.py index 1e298314..c08de4f5 100644 --- a/server/conversation.py +++ b/server/conversation.py @@ -90,7 +90,7 @@ from .conversation_stats import ( collect_upload_events, summarize_upload_events, ) -from .deep_compression import run_deep_compression +from .deep_compression import run_deep_compression, heal_stale_compression_flag conversation_bp = Blueprint('conversation', __name__) @@ -1781,18 +1781,23 @@ def compress_conversation(conversation_id, terminal: WebTerminal, workspace: Use def get_conversation_compression_status(conversation_id, terminal: WebTerminal, workspace: UserWorkspace, username: str): try: normalized_id = conversation_id if conversation_id.startswith('conv_') else f"conv_{conversation_id}" - data = terminal.context_manager._get_conversation_manager_for_id(normalized_id).load_conversation(normalized_id) or {} + target_manager = terminal.context_manager._get_conversation_manager_for_id(normalized_id) + data = target_manager.load_conversation(normalized_id) or {} meta = data.get("metadata", {}) or {} + # 与 /api/status 同一套懒清理：进程重启后残留的压缩标记按 pid 判活。 + compression_in_progress = heal_stale_compression_flag( + target_manager, normalized_id, meta, context_manager=terminal.context_manager + ) return jsonify({ "success": True, "data": { "conversation_id": normalized_id, - "compression_in_progress": bool(meta.get("compression_in_progress", False)), - "compression_mode": meta.get("compression_mode"), - "compression_stage": meta.get("compression_stage"), + "compression_in_progress": compression_in_progress, + "compression_mode": meta.get("compression_mode") if compression_in_progress else None, + "compression_stage": meta.get("compression_stage") if compression_in_progress else None, "compression_error": meta.get("compression_error"), "compression_count": int(meta.get("compression_count", 0) or 0), - "compression_job_id": meta.get("compression_job_id"), + "compression_job_id": meta.get("compression_job_id") if compression_in_progress else None, } }) except Exception as e: diff --git a/server/deep_compression.py b/server/deep_compression.py index 2192bf18..5408b4a9 100644 --- a/server/deep_compression.py +++ b/server/deep_compression.py @@ -2,7 +2,9 @@ from __future__ import annotations import asyncio import json +import os from datetime import datetime +from functools import wraps from pathlib import Path from typing import Any, Dict, List, Optional, Tuple @@ -47,6 +49,89 @@ def _emit(sender, event_type: str, payload: Dict[str, Any]): pass +def _clear_compression_state_on_error(func): + """压缩过程异常退出时清理持久化的 compression_in_progress 标记。 + + 压缩中途异常（或调用侧取消）若不清标记，对话 metadata 会永久残留 + in_progress=True，前端据此锁输入栏/拦切换对话（只能删对话解决）。 + """ + @wraps(func) + async def wrapper(*args, **kwargs): + try: + return await func(*args, **kwargs) + except Exception: + try: + web_terminal = kwargs.get("web_terminal") + conversation_id = kwargs.get("conversation_id") + cm = getattr(web_terminal, "context_manager", None) + if cm is not None and conversation_id: + if getattr(cm, "current_conversation_id", None) == conversation_id: + cm.set_compression_state(in_progress=False) + else: + target_manager = ( + cm._get_conversation_manager_for_id(conversation_id) + if hasattr(cm, "_get_conversation_manager_for_id") + else cm.conversation_manager + ) + target_manager.update_conversation_metadata(conversation_id, { + "compression_in_progress": False, + "compression_mode": None, + "compression_stage": None, + "compression_job_id": None, + "compression_resume_payload": None, + "compression_pid": None, + }) + except Exception: + pass + raise + return wrapper + + +def heal_stale_compression_flag( + target_manager, + conversation_id: str, + metadata: Dict[str, Any], + context_manager=None, +) -> bool: + """读取压缩进行态并懒清理残留标记，返回有效的 in_progress。 + + compression_in_progress 持久化在对话 metadata 中。进程在压缩中途被杀时标记 + 残留为 True（压缩随进程死亡，不可能仍在进行），会导致前端误以为仍在压缩。 + 判定依据：标记写入时记录了发起进程的 pid（compression_pid，见 + compression_mixin.set_compression_state）；与当前进程 pid 不一致（或缺失， + 即旧版本写入的标记）即判定为残留并清除（磁盘 + 内存双清）。 + """ + if not bool((metadata or {}).get("compression_in_progress", False)): + return False + flag_pid = (metadata or {}).get("compression_pid") + if flag_pid is not None and flag_pid == os.getpid(): + return True + clear_updates = { + "compression_in_progress": False, + "compression_mode": None, + "compression_stage": None, + "compression_job_id": None, + "compression_resume_payload": None, + "compression_pid": None, + } + try: + target_manager.update_conversation_metadata(conversation_id, clear_updates) + except Exception: + pass + # 同步内存中的 metadata：该对话若正被加载，工具循环的 + # is_compression_in_progress() 读的是内存副本，不清理会一直误判压缩中。 + try: + if ( + context_manager is not None + and getattr(context_manager, "current_conversation_id", None) == conversation_id + and isinstance(getattr(context_manager, "conversation_metadata", None), dict) + ): + context_manager.conversation_metadata.update(clear_updates) + except Exception: + pass + return False + + def _normalize_deep_compression_records(metadata: Dict[str, Any]) -> List[Dict[str, Any]]: records = metadata.get("deep_compression_records") if not isinstance(records, list): @@ -378,6 +463,7 @@ def _mark_history_compacted(history: List[Dict[str, Any]], *, round_index: int, return marked +@_clear_compression_state_on_error async def run_deep_compression( *, web_terminal, diff --git a/server/status/base.py b/server/status/base.py index c0c0a3f0..985c739d 100644 --- a/server/status/base.py +++ b/server/status/base.py @@ -128,13 +128,22 @@ def get_status(terminal, workspace, username): status['conversation']['created_at'] = current_conv_data.get('created_at') status['conversation']['updated_at'] = current_conv_data.get('updated_at') meta = current_conv_data.get("metadata", {}) or {} + # 压缩标记是持久化的，进程在压缩中途被杀会残留 True（压缩随进程 + # 死亡），这里按 pid 懒清理，避免前端误判压缩中锁死输入栏。 + from ..deep_compression import heal_stale_compression_flag + compression_in_progress = heal_stale_compression_flag( + terminal.context_manager._get_conversation_manager_for_id(current_conv), + current_conv, + meta, + context_manager=terminal.context_manager, + ) status['conversation']['compression'] = { - "in_progress": bool(meta.get("compression_in_progress", False)), - "mode": meta.get("compression_mode"), - "stage": meta.get("compression_stage"), + "in_progress": compression_in_progress, + "mode": meta.get("compression_mode") if compression_in_progress else None, + "stage": meta.get("compression_stage") if compression_in_progress else None, "error": meta.get("compression_error"), "count": int(meta.get("compression_count", 0) or 0), - "job_id": meta.get("compression_job_id"), + "job_id": meta.get("compression_job_id") if compression_in_progress else None, } except Exception as exc: log_conn_diag(f"status conversation-meta-failed user={username} error={exc}") diff --git a/static/src/App.vue b/static/src/App.vue index 84382dea..29c6726d 100644 --- a/static/src/App.vue +++ b/static/src/App.vue @@ -269,7 +269,7 @@ :tool-settings="toolSettings" :tool-settings-loading="toolSettingsLoading" :settings-open="settingsOpen" - :compressing="compressing || compressionInProgress" + :compressing="compressionActiveForCurrentConversation" :current-conversation-id="currentConversationId" :icon-style="iconStyle" :tool-category-icon="toolCategoryIcon" diff --git a/static/src/app/computed.ts b/static/src/app/computed.ts index adbbcab3..91157726 100644 --- a/static/src/app/computed.ts +++ b/static/src/app/computed.ts @@ -227,10 +227,23 @@ export const computed = { displayModeSwitchDisabled() { return !!this.policyUiBlocks.block_virtual_monitor; }, + compressionActiveForCurrentConversation() { + // 压缩锁按对话隔离：只有「正在查看的对话就是正在压缩的对话」时才锁； + // 其他对话与 /new 新建页（currentConversationId 为空）不受影响。 + if (!this.compressionInProgress && !this.compressing) { + return false; + } + const cid = this.compressionConversationId; + if (!cid) { + // 未记录来源（旧数据/异常路径兑底）：只锁有打开对话的页面，/new 不锁 + return !!this.currentConversationId; + } + return cid === this.currentConversationId; + }, displayLockEngaged() { // 对话级并行后：运行中的任务不再硬锁输入栏——当前对话需要保持可输入以排队/停止， - // 其他对话需要保持可输入以并行运行。仅压缩期间保持锁定。 - return !!this.compressionInProgress || !!this.compressing; + // 其他对话需要保持可输入以并行运行。仅当前对话压缩期间保持锁定。 + return this.compressionActiveForCurrentConversation; }, currentWorkspaceHasRunningTask() { // 对话级隔离后：仅当「当前对话」有运行中任务时才拦截发送； @@ -258,8 +271,7 @@ export const computed = { this.taskInProgress || monitorLock || this.stopRequested || - this.compressionInProgress || - this.compressing + this.compressionActiveForCurrentConversation ); }, /** @@ -288,8 +300,7 @@ export const computed = { !this.streamingUi && !this.stopRequested && !(this.monitorIsLocked && this.chatDisplayMode === 'monitor') && - !this.compressionInProgress && - !this.compressing + !this.compressionActiveForCurrentConversation ); }, composerStreamingForInput() { diff --git a/static/src/app/methods/conversation/action.ts b/static/src/app/methods/conversation/action.ts index ea4a1197..3af0d786 100644 --- a/static/src/app/methods/conversation/action.ts +++ b/static/src/app/methods/conversation/action.ts @@ -18,35 +18,8 @@ export const actionMethods = { } }, async createNewConversation() { - if (this.compressionInProgress || this.compressing) { - const confirmed = await this.confirmAction({ - title: '压缩进行中', - message: '对话正在压缩中，切换对话会导致压缩失败，确认要继续吗？', - confirmText: '确认', - cancelText: '取消' - }); - if (!confirmed) { - return; - } - try { - if (this.currentConversationId) { - await fetch(`/api/conversations/${this.currentConversationId}/compression_cancel`, { - method: 'POST' - }); - } - } catch (e) { - console.warn('取消压缩失败:', e); - } - this.compressionInProgress = false; - this.compressing = false; - this.compressionMode = ''; - this.compressionStage = ''; - if (this.compressionToastId) { - this.uiDismissToast(this.compressionToastId); - this.compressionToastId = null; - } - } - + // 已移除「压缩中需确认并取消压缩」拦截：多对话独立运行后，某对话压缩中 + // 不影响新建/切换其他对话（压缩锁已按对话隔离）。 debugLog('创建新对话...'); traceLog('createNewConversation:start', { currentConversationId: this.currentConversationId, diff --git a/static/src/app/methods/conversation/load.ts b/static/src/app/methods/conversation/load.ts index 6da83a83..2f11894c 100644 --- a/static/src/app/methods/conversation/load.ts +++ b/static/src/app/methods/conversation/load.ts @@ -184,36 +184,9 @@ export const loadMethods = { // 不会串到新对话）；fire-and-forget，不阻塞加载 this.flushReasoningEffortSave?.(); - if ((this.compressionInProgress || this.compressing) && !force) { - const confirmed = await this.confirmAction({ - title: '压缩进行中', - message: '对话正在压缩中，切换对话会导致压缩失败，确认要继续吗？', - confirmText: '确认', - cancelText: '取消' - }); - if (!confirmed) { - this.suppressTitleTyping = false; - this.titleReady = true; - return; - } - try { - if (this.currentConversationId) { - await fetch(`/api/conversations/${this.currentConversationId}/compression_cancel`, { - method: 'POST' - }); - } - } catch (e) { - console.warn('取消压缩失败:', e); - } - this.compressionInProgress = false; - this.compressing = false; - this.compressionMode = ''; - this.compressionStage = ''; - if (this.compressionToastId) { - this.uiDismissToast(this.compressionToastId); - this.compressionToastId = null; - } - } + // 已移除「压缩中切换对话需确认/取消压缩」拦截：多对话独立运行后，压缩在 + // 后端随对话任务进行，切换视图不影响压缩本身；压缩锁也按对话隔离 + // （compressionActiveForCurrentConversation），不再存在对话级全局锁。 // 注意：加载已有对话时必须保留该对话自身的模型/模式，不能套用用户默认值。 diff --git a/static/src/app/methods/message/chat.ts b/static/src/app/methods/message/chat.ts index 64b0c605..4eab3e62 100644 --- a/static/src/app/methods/message/chat.ts +++ b/static/src/app/methods/message/chat.ts @@ -44,6 +44,7 @@ export const chatMethods = { this.compressing = true; this.compressionInProgress = true; + this.compressionConversationId = this.currentConversationId; this.compressionMode = 'manual'; this.compressionStage = 'requesting'; this.compressionError = ''; @@ -120,6 +121,7 @@ export const chatMethods = { } this.compressing = false; this.compressionInProgress = false; + this.compressionConversationId = null; this.compressionMode = ''; this.compressionStage = ''; } diff --git a/static/src/app/methods/message/send.ts b/static/src/app/methods/message/send.ts index bec586b5..cf15366b 100644 --- a/static/src/app/methods/message/send.ts +++ b/static/src/app/methods/message/send.ts @@ -10,7 +10,7 @@ import { export const sendMethods = { async handleSendOrStop() { - if (this.compressionInProgress) { + if (this.compressionActiveForCurrentConversation) { this.uiPushToast({ title: '对话自动压缩中', message: '当前不可发送/停止，请等待压缩完成', @@ -37,7 +37,7 @@ export const sendMethods = { // 传统模式：waitingForSubAgent=true（taskInProgress=true 。多智能体模式：has_running_multi_agent=true。 // 此时新文本消息应直接发送，触发主智能体下一轮工作，而不是被进队列等任务结束。 const mainIdle = typeof this.mainChatIdle === 'function' ? this.mainChatIdle : ( - !this.streamingUi && !this.stopRequested && !this.compressionInProgress && !this.compressing + !this.streamingUi && !this.stopRequested && !this.compressionActiveForCurrentConversation ); if (this.composerBusy && mainIdle && hasText) { // 如果有 pending 问题（子智能体询问主智能体），仍走问答路径，不走直接发送 @@ -115,7 +115,7 @@ export const sendMethods = { const presetText = typeof options?.presetText === 'string' ? options.presetText : null; const usePresetText = presetText !== null; - if (this.compressionInProgress) { + if (this.compressionActiveForCurrentConversation) { this.uiPushToast({ title: '对话自动压缩中', message: '压缩完成后才能继续发送消息', @@ -473,7 +473,7 @@ export const sendMethods = { return; } this._stopTaskRunning = true; - if (this.compressionInProgress) { + if (this.compressionActiveForCurrentConversation) { this._stopTaskRunning = false; this.uiPushToast({ title: '对话自动压缩中', diff --git a/static/src/app/methods/resources.ts b/static/src/app/methods/resources.ts index 99ea12da..754424e9 100644 --- a/static/src/app/methods/resources.ts +++ b/static/src/app/methods/resources.ts @@ -204,17 +204,24 @@ export const resourceMethods = { this.conversationHasImages = false; this.conversationHasVideos = false; } + // 压缩状态与 has_images/has_videos 同理：status 反映的是后端 terminal 当前 + // 对话的压缩状态。空对话/显式新建路由（/new）上没有打开对话，status 里的 + // 压缩标记属于残留的旧对话上下文，应用会把输入锁错误地带到 /new 页。 const compression = status?.conversation?.compression; - if (compression && typeof compression === 'object') { - this.compressionInProgress = !!compression.in_progress; - this.compressionMode = compression.mode || ''; - this.compressionStage = compression.stage || ''; - this.compressionError = compression.error || ''; - } else { - this.compressionInProgress = false; - this.compressionMode = ''; - this.compressionStage = ''; - this.compressionError = ''; + if (hasConversation && !onExplicitNewRoute) { + if (compression && typeof compression === 'object') { + this.compressionInProgress = !!compression.in_progress; + this.compressionMode = compression.mode || ''; + this.compressionStage = compression.stage || ''; + this.compressionError = compression.error || ''; + this.compressionConversationId = this.compressionInProgress ? this.currentConversationId : null; + } else { + this.compressionInProgress = false; + this.compressionConversationId = null; + this.compressionMode = ''; + this.compressionStage = ''; + this.compressionError = ''; + } } }, diff --git a/static/src/app/methods/taskPolling/compression.ts b/static/src/app/methods/taskPolling/compression.ts index 4c188971..4a7f1114 100644 --- a/static/src/app/methods/taskPolling/compression.ts +++ b/static/src/app/methods/taskPolling/compression.ts @@ -25,6 +25,10 @@ export const compressionMethods = { } const wasInProgress = !!this.compressionInProgress; this.compressionInProgress = !!data.in_progress; + // 记录压缩所属对话：压缩锁只作用于该对话，不影响其他对话与 /new 新建页 + this.compressionConversationId = this.compressionInProgress + ? (data.conversation_id || this.currentConversationId || null) + : null; this.compressionMode = data.mode || ''; this.compressionStage = data.stage || ''; if (this.compressionInProgress && !wasInProgress) { @@ -68,6 +72,7 @@ export const compressionMethods = { this.compressionToastId = null; } this.compressionInProgress = false; + this.compressionConversationId = null; this.compressionMode = ''; this.compressionStage = ''; this.compressionError = ''; diff --git a/static/src/app/methods/ui/review.ts b/static/src/app/methods/ui/review.ts index 3bf98866..2a09912a 100644 --- a/static/src/app/methods/ui/review.ts +++ b/static/src/app/methods/ui/review.ts @@ -28,7 +28,7 @@ export const reviewMethods = { if (this.compressing || this.streamingMessage || !this.isConnected) { return; } - if (this.compressionInProgress) { + if (this.compressionActiveForCurrentConversation) { this.uiPushToast({ title: '对话自动压缩中', message: '当前对话正在压缩，请稍后再试', diff --git a/static/src/app/state.ts b/static/src/app/state.ts index 61650ab7..a9975ded 100644 --- a/static/src/app/state.ts +++ b/static/src/app/state.ts @@ -135,6 +135,9 @@ export function dataState() { // 对话压缩状态 compressing: false, compressionInProgress: false, + // 正在压缩的对话 id：压缩锁只作用于该对话本身，不影响其他对话与 /new 新建页 + // （多对话独立运行后，压缩不再是对话级全局锁）。 + compressionConversationId: null, compressionMode: '', compressionStage: '', compressionError: '', diff --git a/utils/context_manager/compression_mixin.py b/utils/context_manager/compression_mixin.py index 0665ba7a..ee3fa1ba 100644 --- a/utils/context_manager/compression_mixin.py +++ b/utils/context_manager/compression_mixin.py @@ -97,6 +97,9 @@ class CompressionMixin: "compression_error": error if in_progress else None, "compression_resume_payload": resume_payload if in_progress else None, "compression_job_id": job_id if in_progress else None, + # 记录发起压缩的进程 pid：进程重启后该标记必然残留（压缩随进程死亡）， + # status 读取侧据此做懒清理（见 deep_compression.heal_stale_compression_flag）。 + "compression_pid": os.getpid() if in_progress else None, } for k, v in updates.items(): self.conversation_metadata[k] = v