JOJO/agent-Specialization
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

fix(chat): 修复子智能体完成通知不实时显示，压缩残留锁死输入与切换

Browse Source 
- 通知：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 异常时
 自动清理标记
This commit is contained in:
JOJO 2026-08-12 22:27:31 +08:00
parent 34bb5d61cc
commit 16900b0a20
16 changed files with 180 additions and 93 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
server/chat_flow_task_main.py
View File

@ -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(

8
server/chat_flow_task_support.py
View File

@ -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

17
server/conversation.py
View File

@ -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:

86
server/deep_compression.py
View File

@ -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,

17
server/status/base.py
View File

@ -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}")

2
static/src/App.vue
View File

@ -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"

23
static/src/app/computed.ts
View File

@ -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() {

31
static/src/app/methods/conversation/action.ts
View File

@ -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,

33
static/src/app/methods/conversation/load.ts
View File

@ -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），不再存在对话级全局锁。
// 注意：加载已有对话时必须保留该对话自身的模型/模式，不能套用用户默认值。

2
static/src/app/methods/message/chat.ts
View File

@ -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 = '';
}

8
static/src/app/methods/message/send.ts
View File

@ -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: '对话自动压缩中',

27
static/src/app/methods/resources.ts
View File

@ -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 = '';
}
}
},

5
static/src/app/methods/taskPolling/compression.ts
View File

@ -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 = '';

2
static/src/app/methods/ui/review.ts
View File

@ -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: '当前对话正在压缩，请稍后再试',

3
static/src/app/state.ts
View File

@ -135,6 +135,9 @@ export function dataState() {
// 对话压缩状态
compressing: false,
compressionInProgress: false,
// 正在压缩的对话 id：压缩锁只作用于该对话本身，不影响其他对话与 /new 新建页
// （多对话独立运行后，压缩不再是对话级全局锁）。
compressionConversationId: null,
compressionMode: '',
compressionStage: '',
compressionError: '',

3
utils/context_manager/compression_mixin.py
View File

@ -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