JOJO
01a1658a99
save_conversation 全量覆写改为合并写：快路径前缀追加与现状一致；
慢路径以磁盘为基、同 id 取内存版、内存独有追加在后——旧实例任何
时机的写回都不再丢消息，回退从根免疫。豁免路径（allow_shrink，
仅检查点恢复）保持覆写语义。A 守卫降级为 merged 断言，缩减写回
从拒绝进化为矫正（旧实例独有消息也能救回）。慢路径矫正时打印
🔀 [ConvSaveMerge] 观察日志。详见计划文档 §7。
724 lines
32 KiB
Python
724 lines
32 KiB
Python
# utils/conversation_manager.py - 对话持久化管理器（集成Token统计）
|
||
|
||
import json
|
||
import os
|
||
import time
|
||
import tempfile
|
||
import threading
|
||
from datetime import datetime
|
||
from pathlib import Path
|
||
from typing import Dict, List, Optional, Any
|
||
from dataclasses import dataclass
|
||
try:
|
||
from config import DATA_DIR, HOST_WORKSPACES_FILE
|
||
except ImportError:
|
||
import sys
|
||
from pathlib import Path
|
||
project_root = Path(__file__).resolve().parents[1]
|
||
if str(project_root) not in sys.path:
|
||
sys.path.insert(0, str(project_root))
|
||
from config import DATA_DIR, HOST_WORKSPACES_FILE
|
||
|
||
try:
|
||
from utils.perf_log import perf_log
|
||
except Exception:
|
||
perf_log = None
|
||
|
||
try:
|
||
from config import LOGS_DIR as _CONV_SAVE_LOG_DIR
|
||
except Exception:
|
||
_CONV_SAVE_LOG_DIR = None
|
||
|
||
|
||
def _debug_conversation_save_trace(conversation_id: str, file_path, data: Dict):
|
||
"""[临时调试] 记录每次对话文件写入，重点捕获「非空被空覆盖」的调用栈。
|
||
|
||
输出: logs/conversation_save_debug.log （JSONL）
|
||
排查完「对话消息被清空」问题后应移除。
|
||
"""
|
||
try:
|
||
import traceback
|
||
new_msgs = data.get("messages") if isinstance(data, dict) else None
|
||
new_len = len(new_msgs) if isinstance(new_msgs, list) else -1
|
||
old_len = None
|
||
try:
|
||
p = Path(file_path)
|
||
if p.exists():
|
||
old_data = json.loads(p.read_text(encoding="utf-8"))
|
||
old_msgs = old_data.get("messages")
|
||
old_len = len(old_msgs) if isinstance(old_msgs, list) else None
|
||
except Exception:
|
||
old_len = None
|
||
wipe_suspect = bool(old_len and new_len == 0)
|
||
shrink_suspect = bool(old_len and 0 < new_len < old_len)
|
||
# 只保留项目内调用帧，排除标准库/第三方
|
||
try:
|
||
repo_root = Path(__file__).resolve().parents[2]
|
||
except Exception:
|
||
repo_root = None
|
||
frames = []
|
||
for fr in traceback.extract_stack()[:-2]:
|
||
fn = str(fr.filename)
|
||
if repo_root and not fn.startswith(str(repo_root)):
|
||
continue
|
||
if "/.astrion/" in fn or "site-packages" in fn:
|
||
continue
|
||
rel = fn[len(str(repo_root)) + 1:] if repo_root and fn.startswith(str(repo_root)) else fn
|
||
frames.append(f"{rel}:{fr.lineno}:{fr.name}")
|
||
record = {
|
||
"ts": datetime.now().isoformat(),
|
||
"event": "conversation_save",
|
||
"conversation_id": conversation_id,
|
||
"old_msg_len": old_len,
|
||
"new_msg_len": new_len,
|
||
"wipe_suspect": wipe_suspect,
|
||
"shrink_suspect": shrink_suspect,
|
||
"title": (data.get("title") if isinstance(data, dict) else None),
|
||
"thread": threading.current_thread().name,
|
||
"stack_tail": frames[-14:],
|
||
}
|
||
if _CONV_SAVE_LOG_DIR:
|
||
log_dir = Path(_CONV_SAVE_LOG_DIR)
|
||
else:
|
||
log_dir = Path(DATA_DIR).parent / "logs"
|
||
log_dir.mkdir(parents=True, exist_ok=True)
|
||
with open(log_dir / "conversation_save_debug.log", "a", encoding="utf-8") as f:
|
||
f.write(json.dumps(record, ensure_ascii=False) + "\n")
|
||
if wipe_suspect or shrink_suspect:
|
||
kind = "空列表覆盖" if wipe_suspect else f"缩减 {old_len}->{new_len}"
|
||
print(f"🚨 [ConvSaveDebug] 对话 {conversation_id} 消息将被{kind}！栈: {frames[-14:]}")
|
||
except Exception:
|
||
pass
|
||
|
||
@dataclass
|
||
class ConversationMetadata:
|
||
"""对话元数据"""
|
||
id: str
|
||
title: str
|
||
created_at: str
|
||
updated_at: str
|
||
project_path: Optional[str]
|
||
project_relative_path: Optional[str]
|
||
thinking_mode: bool
|
||
total_messages: int
|
||
total_tools: int
|
||
run_mode: str = "fast"
|
||
model_key: Optional[str] = None
|
||
has_images: bool = False
|
||
has_videos: bool = False
|
||
status: str = "active" # active, archived, error
|
||
|
||
|
||
class CrudMixin:
|
||
"""ConversationManager crud mixin 能力 mixin。"""
|
||
|
||
def create_conversation(
|
||
self,
|
||
project_path: str,
|
||
thinking_mode: bool = False,
|
||
run_mode: str = "fast",
|
||
initial_messages: List[Dict] = None,
|
||
model_key: Optional[str] = None,
|
||
has_images: bool = False,
|
||
has_videos: bool = False,
|
||
metadata_overrides: Optional[Dict[str, Any]] = None
|
||
) -> str:
|
||
"""
|
||
创建新对话
|
||
|
||
Args:
|
||
project_path: 项目路径
|
||
thinking_mode: 思考模式
|
||
initial_messages: 初始消息列表
|
||
|
||
Returns:
|
||
conversation_id: 对话ID
|
||
"""
|
||
t0 = time.perf_counter()
|
||
if perf_log:
|
||
perf_log("cm.create_conversation enter")
|
||
conversation_id = self._generate_conversation_id()
|
||
messages = initial_messages or []
|
||
|
||
# 创建对话数据
|
||
path_metadata = self._prepare_project_path_metadata(project_path)
|
||
normalized_mode = run_mode if run_mode in {"fast", "thinking", "deep"} else ("thinking" if thinking_mode else "fast")
|
||
metadata = {
|
||
"project_path": path_metadata["project_path"],
|
||
"project_relative_path": path_metadata["project_relative_path"],
|
||
"thinking_mode": thinking_mode,
|
||
"run_mode": normalized_mode,
|
||
"model_key": model_key,
|
||
"permission_mode": "unrestricted",
|
||
"execution_mode": "sandbox",
|
||
"pending_permission_mode": None,
|
||
"pending_execution_mode": None,
|
||
"frozen_permission_prompt": None,
|
||
"frozen_execution_prompt": None,
|
||
"has_images": has_images,
|
||
"has_videos": has_videos,
|
||
# 首次对话尚未生成文件树快照，待首次用户消息时填充
|
||
"project_file_tree": None,
|
||
"project_statistics": None,
|
||
"project_snapshot_at": None,
|
||
"total_messages": len(messages),
|
||
"total_tools": self._count_tools_in_messages(messages),
|
||
"status": "active",
|
||
# 压缩相关字段
|
||
"compression_count": 0,
|
||
"is_long_conversation": False,
|
||
"is_ultra_long_conversation": False,
|
||
"tool_call_count": 0,
|
||
"last_shallow_compress_tool_count": 0,
|
||
"compression_in_progress": False,
|
||
"compression_mode": None,
|
||
"compression_stage": None,
|
||
"compression_job_id": None,
|
||
"compression_error": None,
|
||
"compression_resume_payload": None,
|
||
"deep_compression_records": [],
|
||
"last_deep_compression_record": None,
|
||
"skip_auto_title_generation": False,
|
||
}
|
||
if isinstance(metadata_overrides, dict):
|
||
metadata.update(metadata_overrides)
|
||
|
||
conversation_data = {
|
||
"id": conversation_id,
|
||
"title": self._extract_title_from_messages(messages),
|
||
"created_at": datetime.now().isoformat(),
|
||
"updated_at": datetime.now().isoformat(),
|
||
"messages": messages,
|
||
"todo_list": None,
|
||
"metadata": metadata,
|
||
"token_statistics": self._initialize_token_statistics() # 新增
|
||
}
|
||
|
||
# 保存对话文件
|
||
t1 = time.perf_counter()
|
||
self._save_conversation_file(conversation_id, conversation_data)
|
||
if perf_log:
|
||
perf_log("cm.create_conversation save_file", elapsed_ms=(time.perf_counter() - t0) * 1000, extra={"conv_id": conversation_id})
|
||
|
||
# 更新索引
|
||
self._update_index(conversation_id, conversation_data)
|
||
if perf_log:
|
||
perf_log("cm.create_conversation update_index", elapsed_ms=(time.perf_counter() - t0) * 1000, extra={"conv_id": conversation_id})
|
||
|
||
self.current_conversation_id = conversation_id
|
||
elapsed_ms = (time.perf_counter() - t0) * 1000
|
||
if perf_log:
|
||
perf_log("cm.create_conversation done", elapsed_ms=elapsed_ms, extra={"conv_id": conversation_id})
|
||
print(f"📝 创建新对话: {conversation_id} - {conversation_data['title']}")
|
||
|
||
return conversation_id
|
||
|
||
def update_conversation_metadata(self, conversation_id: str, updates: Dict[str, Any]) -> bool:
|
||
"""合并更新对话 metadata。"""
|
||
if not conversation_id or not isinstance(updates, dict):
|
||
return False
|
||
try:
|
||
data = self.load_conversation(conversation_id)
|
||
if not data:
|
||
return False
|
||
metadata = data.get("metadata", {}) or {}
|
||
metadata.update(updates)
|
||
data["metadata"] = metadata
|
||
data["updated_at"] = datetime.now().isoformat()
|
||
self._save_conversation_file(conversation_id, data)
|
||
self._update_index(conversation_id, data)
|
||
return True
|
||
except Exception as exc:
|
||
print(f"⚠️ 更新对话 metadata 失败 {conversation_id}: {exc}")
|
||
return False
|
||
|
||
def update_conversation_title(self, conversation_id: str, title: str) -> bool:
|
||
"""更新对话标题并刷新索引。"""
|
||
if not conversation_id or not title:
|
||
return False
|
||
try:
|
||
data = self.load_conversation(conversation_id)
|
||
if not data:
|
||
return False
|
||
data["title"] = title
|
||
data["updated_at"] = datetime.now().isoformat()
|
||
meta = data.get("metadata", {}) or {}
|
||
meta["title_locked"] = True
|
||
data["metadata"] = meta
|
||
self._save_conversation_file(conversation_id, data)
|
||
self._update_index(conversation_id, data)
|
||
if self.current_conversation_id == conversation_id:
|
||
self.current_conversation_title = title
|
||
return True
|
||
except Exception as exc:
|
||
print(f"⚠️ 更新对话标题失败 {conversation_id}: {exc}")
|
||
return False
|
||
|
||
def _save_conversation_file(self, conversation_id: str, data: Dict):
|
||
"""保存对话文件"""
|
||
file_path = self._get_conversation_file_path(conversation_id)
|
||
try:
|
||
# 确保Token统计数据有效
|
||
data = self._validate_token_statistics(data)
|
||
_debug_conversation_save_trace(conversation_id, file_path, data)
|
||
with self._io_lock:
|
||
self._atomic_write_json(file_path, data)
|
||
except Exception as e:
|
||
print(f"⌘ 保存对话文件失败 {conversation_id}: {e}")
|
||
|
||
def _update_index(self, conversation_id: str, conversation_data: Dict):
|
||
"""更新对话索引"""
|
||
try:
|
||
index = self._load_index()
|
||
|
||
# 创建元数据
|
||
metadata = ConversationMetadata(
|
||
id=conversation_id,
|
||
title=conversation_data["title"],
|
||
created_at=conversation_data["created_at"],
|
||
updated_at=conversation_data["updated_at"],
|
||
project_path=conversation_data["metadata"]["project_path"],
|
||
project_relative_path=conversation_data["metadata"].get("project_relative_path"),
|
||
thinking_mode=conversation_data["metadata"]["thinking_mode"],
|
||
run_mode=conversation_data["metadata"].get("run_mode", "thinking" if conversation_data["metadata"]["thinking_mode"] else "fast"),
|
||
total_messages=conversation_data["metadata"]["total_messages"],
|
||
total_tools=conversation_data["metadata"]["total_tools"],
|
||
status=conversation_data["metadata"].get("status", "active")
|
||
)
|
||
|
||
# 添加到索引
|
||
index[conversation_id] = {
|
||
"title": metadata.title,
|
||
"created_at": metadata.created_at,
|
||
"updated_at": metadata.updated_at,
|
||
"project_path": metadata.project_path,
|
||
"project_relative_path": metadata.project_relative_path,
|
||
"thinking_mode": metadata.thinking_mode,
|
||
"run_mode": metadata.run_mode,
|
||
"model_key": conversation_data["metadata"].get("model_key"),
|
||
"has_images": conversation_data["metadata"].get("has_images", False),
|
||
"has_videos": conversation_data["metadata"].get("has_videos", False),
|
||
"total_messages": metadata.total_messages,
|
||
"total_tools": metadata.total_tools,
|
||
"status": metadata.status,
|
||
"multi_agent_mode": bool(conversation_data["metadata"].get("multi_agent_mode", False))
|
||
}
|
||
|
||
self._save_index(index)
|
||
except Exception as e:
|
||
print(f"⌘ 更新对话索引失败: {e}")
|
||
|
||
@staticmethod
|
||
def _merge_messages_by_id(disk_messages: List[Dict], new_messages: List[Dict]):
|
||
"""按 message_id 合并磁盘与内存消息（方向 C：merge-on-save）。
|
||
|
||
规则：
|
||
- 快路径：内存 id 序列前缀包含磁盘 id 序列（正常追加）→ 直接返回内存版；
|
||
- 慢路径：以磁盘为基，同 id 取内存版（浅压缩打标等修改生效），
|
||
内存独有 id 消息按原序追加在后；无 message_id 的内存消息防御性跳过追加
|
||
（存量消息 100% 有 id，新消息由 add_conversation 生成必有 id）。
|
||
|
||
Returns:
|
||
(merged, appended, healed):
|
||
merged 合并后的消息列表
|
||
appended 内存独有而被追加的消息数
|
||
healed 磁盘有而内存缺失（被矫正保留）的消息数
|
||
"""
|
||
disk_ids = [m.get("message_id") for m in disk_messages]
|
||
new_ids = [m.get("message_id") for m in new_messages]
|
||
# 快路径：前缀追加，与现状全量覆写语义一致
|
||
if len(new_ids) >= len(disk_ids) and new_ids[:len(disk_ids)] == disk_ids:
|
||
return list(new_messages), 0, 0
|
||
|
||
new_by_id = {m.get("message_id"): m for m in new_messages if m.get("message_id")}
|
||
merged: List[Dict] = []
|
||
seen = set()
|
||
healed = 0
|
||
for dm in disk_messages:
|
||
mid = dm.get("message_id")
|
||
if mid and mid in new_by_id:
|
||
merged.append(new_by_id[mid])
|
||
else:
|
||
merged.append(dm)
|
||
if mid:
|
||
healed += 1
|
||
if mid:
|
||
seen.add(mid)
|
||
appended = 0
|
||
for m in new_messages:
|
||
mid = m.get("message_id")
|
||
if not mid:
|
||
continue
|
||
if mid not in seen:
|
||
merged.append(m)
|
||
seen.add(mid)
|
||
appended += 1
|
||
return merged, appended, healed
|
||
|
||
def save_conversation(
|
||
self,
|
||
conversation_id: str,
|
||
messages: List[Dict],
|
||
project_path: str = None,
|
||
thinking_mode: bool = None,
|
||
run_mode: Optional[str] = None,
|
||
todo_list: Optional[Dict] = None,
|
||
model_key: Optional[str] = None,
|
||
has_images: Optional[bool] = None,
|
||
has_videos: Optional[bool] = None,
|
||
project_file_tree: Optional[str] = None,
|
||
project_statistics: Optional[Dict] = None,
|
||
project_snapshot_at: Optional[str] = None,
|
||
allow_shrink: bool = False
|
||
) -> bool:
|
||
"""
|
||
保存对话（更新现有对话）
|
||
|
||
Args:
|
||
conversation_id: 对话ID
|
||
messages: 消息列表
|
||
project_path: 项目路径
|
||
thinking_mode: 思考模式
|
||
allow_shrink: 允许消息数缩减（仅检查点恢复等合法缩减路径使用）
|
||
|
||
Returns:
|
||
bool: 保存是否成功
|
||
"""
|
||
try:
|
||
# 加载现有对话数据
|
||
existing_data = self.load_conversation(conversation_id)
|
||
if not existing_data:
|
||
print(f"⚠️ 对话 {conversation_id} 不存在，无法更新")
|
||
return False
|
||
|
||
# 方向 C（merge-on-save）：正常保存不做全量覆写，按 message_id 合并，
|
||
# 任何实例任何时机的写回都不会丢消息（旧内存回写从根免疫）。
|
||
# 仅检查点恢复等 allow_shrink=True 路径保持覆写语义。
|
||
old_messages = existing_data.get("messages")
|
||
old_len = len(old_messages) if isinstance(old_messages, list) else 0
|
||
new_len = len(messages) if isinstance(messages, list) else 0
|
||
if not allow_shrink and isinstance(old_messages, list) and old_messages:
|
||
merged, appended, healed = self._merge_messages_by_id(old_messages, messages or [])
|
||
if healed or appended:
|
||
print(
|
||
f"🔀 [ConvSaveMerge] {conversation_id}: "
|
||
f"磁盘 {old_len} 条 / 内存 {new_len} 条 → 合并 {len(merged)} 条"
|
||
f"（矫正保留磁盘 {healed} 条，追加内存独有 {appended} 条）"
|
||
)
|
||
messages = merged
|
||
new_len = len(messages)
|
||
# 守卫断言：merge 后恒不缩减（触发即 merge 实现 bug）；豁免路径跳过。
|
||
if new_len < old_len and not allow_shrink:
|
||
print(
|
||
f"🚨 [ConvSaveGuard] 拒绝保存 {conversation_id}: "
|
||
f"消息数 {old_len} -> {new_len} 缩减且未豁免（merge 断言失败，疑似 bug）"
|
||
)
|
||
return False
|
||
|
||
# 更新数据
|
||
existing_data["messages"] = messages
|
||
existing_data["updated_at"] = datetime.now().isoformat()
|
||
|
||
# 更新标题（如未锁定自动标题时，仍按首条消息回退）
|
||
title_locked = existing_data.get("metadata", {}).get("title_locked", False)
|
||
if not title_locked:
|
||
new_title = self._extract_title_from_messages(messages)
|
||
if new_title != "新对话":
|
||
existing_data["title"] = new_title
|
||
|
||
# 更新元数据
|
||
if project_path is not None:
|
||
path_metadata = self._prepare_project_path_metadata(project_path)
|
||
existing_data["metadata"]["project_path"] = path_metadata["project_path"]
|
||
existing_data["metadata"]["project_relative_path"] = path_metadata["project_relative_path"]
|
||
else:
|
||
existing_data["metadata"].setdefault("project_relative_path", None)
|
||
if thinking_mode is not None:
|
||
existing_data["metadata"]["thinking_mode"] = thinking_mode
|
||
if run_mode:
|
||
normalized_mode = run_mode if run_mode in {"fast", "thinking", "deep"} else (
|
||
"thinking" if existing_data["metadata"].get("thinking_mode") else "fast"
|
||
)
|
||
existing_data["metadata"]["run_mode"] = normalized_mode
|
||
elif "run_mode" not in existing_data["metadata"]:
|
||
existing_data["metadata"]["run_mode"] = "thinking" if existing_data["metadata"].get("thinking_mode") else "fast"
|
||
# 推断最新使用的模型（优先参数，其次倒序扫描助手消息）
|
||
inferred_model = None
|
||
if model_key is None:
|
||
for msg in reversed(messages):
|
||
if msg.get("role") != "assistant":
|
||
continue
|
||
msg_meta = msg.get("metadata") or {}
|
||
mk = msg_meta.get("model_key")
|
||
if mk:
|
||
inferred_model = mk
|
||
break
|
||
target_model = model_key if model_key is not None else inferred_model
|
||
if target_model is not None:
|
||
existing_data["metadata"]["model_key"] = target_model
|
||
elif "model_key" not in existing_data["metadata"]:
|
||
existing_data["metadata"]["model_key"] = None
|
||
if has_images is not None:
|
||
existing_data["metadata"]["has_images"] = bool(has_images)
|
||
elif "has_images" not in existing_data["metadata"]:
|
||
existing_data["metadata"]["has_images"] = False
|
||
if has_videos is not None:
|
||
existing_data["metadata"]["has_videos"] = bool(has_videos)
|
||
elif "has_videos" not in existing_data["metadata"]:
|
||
existing_data["metadata"]["has_videos"] = False
|
||
# 文件树快照（如果有新值则更新，若已有则保持）
|
||
if project_file_tree is not None:
|
||
existing_data["metadata"]["project_file_tree"] = project_file_tree
|
||
elif "project_file_tree" not in existing_data["metadata"]:
|
||
existing_data["metadata"]["project_file_tree"] = None
|
||
if project_statistics is not None:
|
||
existing_data["metadata"]["project_statistics"] = project_statistics
|
||
elif "project_statistics" not in existing_data["metadata"]:
|
||
existing_data["metadata"]["project_statistics"] = None
|
||
if project_snapshot_at is not None:
|
||
existing_data["metadata"]["project_snapshot_at"] = project_snapshot_at
|
||
elif "project_snapshot_at" not in existing_data["metadata"]:
|
||
existing_data["metadata"]["project_snapshot_at"] = None
|
||
|
||
existing_data["metadata"]["total_messages"] = len(messages)
|
||
existing_data["metadata"]["total_tools"] = self._count_tools_in_messages(messages)
|
||
|
||
# 更新待办列表
|
||
existing_data["todo_list"] = todo_list
|
||
|
||
# 确保Token统计结构存在（向后兼容）
|
||
if "token_statistics" not in existing_data:
|
||
existing_data["token_statistics"] = self._initialize_token_statistics()
|
||
else:
|
||
existing_data["token_statistics"]["updated_at"] = datetime.now().isoformat()
|
||
|
||
# 保存文件
|
||
self._save_conversation_file(conversation_id, existing_data)
|
||
|
||
# 更新索引
|
||
self._update_index(conversation_id, existing_data)
|
||
|
||
return True
|
||
except Exception as e:
|
||
print(f"⌘ 保存对话失败 {conversation_id}: {e}")
|
||
return False
|
||
|
||
def mark_latest_user_work_completed(
|
||
self,
|
||
conversation_id: str,
|
||
finished_at: Optional[str] = None,
|
||
context_manager: Optional[Any] = None,
|
||
) -> bool:
|
||
"""将对话中最后一条 status 为 working 的用户消息的 work_timer 标记为完成。
|
||
|
||
用于任务停止或正常结束后，刷新页面时不再恢复为"工作中"状态。
|
||
|
||
Args:
|
||
conversation_id: 要处理的对话 ID。
|
||
finished_at: 完成时间 ISO 字符串，默认当前时间。
|
||
context_manager: 可选的当前 ContextManager 实例；若提供，会同步更新
|
||
其内存中的 conversation_history，避免后续 auto_save 把已持久化的
|
||
完成状态覆盖回去。
|
||
"""
|
||
if not conversation_id:
|
||
return False
|
||
try:
|
||
data = self.load_conversation(conversation_id)
|
||
if not data:
|
||
return False
|
||
messages = data.get("messages") or []
|
||
completed = False
|
||
now_iso = finished_at or datetime.now().isoformat()
|
||
for msg in reversed(messages):
|
||
if not isinstance(msg, dict) or msg.get("role") != "user":
|
||
continue
|
||
metadata = msg.get("metadata") or {}
|
||
timer = metadata.get("work_timer")
|
||
if not isinstance(timer, dict):
|
||
continue
|
||
if timer.get("status") != "working":
|
||
continue
|
||
started_at = timer.get("started_at") or msg.get("timestamp") or now_iso
|
||
try:
|
||
start_dt = datetime.fromisoformat(str(started_at).replace("Z", "+00:00"))
|
||
end_dt = datetime.fromisoformat(now_iso.replace("Z", "+00:00"))
|
||
duration_ms = max(0, int((end_dt - start_dt).total_seconds() * 1000))
|
||
except Exception:
|
||
duration_ms = timer.get("duration_ms", 0) or 0
|
||
timer["status"] = "completed"
|
||
timer["started_at"] = started_at
|
||
timer["finished_at"] = now_iso
|
||
timer["duration_ms"] = duration_ms
|
||
msg["metadata"] = metadata
|
||
completed = True
|
||
break
|
||
if not completed:
|
||
return False
|
||
data["updated_at"] = datetime.now().isoformat()
|
||
self._save_conversation_file(conversation_id, data)
|
||
self._update_index(conversation_id, data)
|
||
|
||
# 同步内存副本（如果调用方持有当前对话的内存引用）
|
||
if (
|
||
context_manager
|
||
and getattr(context_manager, "current_conversation_id", None) == conversation_id
|
||
):
|
||
try:
|
||
self._sync_work_timer_in_memory(context_manager, now_iso)
|
||
except Exception:
|
||
pass
|
||
return True
|
||
except Exception as exc:
|
||
print(f"⌘ 标记用户工作完成失败 {conversation_id}: {exc}")
|
||
return False
|
||
|
||
def _sync_work_timer_in_memory(self, context_manager: Any, finished_at: str) -> bool:
|
||
"""将当前 ContextManager 内存中最后一条 working 的 work_timer 标记为完成。"""
|
||
history = getattr(context_manager, "conversation_history", None) or []
|
||
if not history:
|
||
return False
|
||
|
||
for msg in reversed(history):
|
||
if not isinstance(msg, dict) or msg.get("role") != "user":
|
||
continue
|
||
metadata = msg.get("metadata") or {}
|
||
timer = metadata.get("work_timer")
|
||
if not isinstance(timer, dict):
|
||
continue
|
||
if timer.get("status") != "working":
|
||
continue
|
||
|
||
started_at = timer.get("started_at") or msg.get("timestamp") or finished_at
|
||
try:
|
||
start_dt = datetime.fromisoformat(str(started_at).replace("Z", "+00:00"))
|
||
end_dt = datetime.fromisoformat(finished_at.replace("Z", "+00:00"))
|
||
duration_ms = max(0, int((end_dt - start_dt).total_seconds() * 1000))
|
||
except Exception:
|
||
duration_ms = timer.get("duration_ms", 0) or 0
|
||
|
||
timer.update({
|
||
"status": "completed",
|
||
"started_at": started_at,
|
||
"finished_at": finished_at,
|
||
"duration_ms": duration_ms,
|
||
})
|
||
msg["metadata"] = metadata
|
||
return True
|
||
|
||
return False
|
||
|
||
def update_project_snapshot(
|
||
self,
|
||
conversation_id: str,
|
||
project_file_tree: str,
|
||
project_statistics: Optional[Dict],
|
||
project_snapshot_at: Optional[str] = None
|
||
) -> bool:
|
||
"""
|
||
单独更新对话的项目文件树快照，不修改消息内容。
|
||
便于在首次用户消息时写入固定文件结构。
|
||
"""
|
||
try:
|
||
data = self.load_conversation(conversation_id)
|
||
if not data:
|
||
return False
|
||
meta = data.get("metadata", {}) or {}
|
||
meta["project_file_tree"] = project_file_tree
|
||
meta["project_statistics"] = project_statistics
|
||
meta["project_snapshot_at"] = project_snapshot_at
|
||
data["metadata"] = meta
|
||
self._save_conversation_file(conversation_id, data)
|
||
self._update_index(conversation_id, data)
|
||
return True
|
||
except Exception as exc:
|
||
print(f"⌘ 更新项目快照失败 {conversation_id}: {exc}")
|
||
return False
|
||
|
||
def load_conversation(self, conversation_id: str) -> Optional[Dict]:
|
||
"""
|
||
加载对话数据
|
||
|
||
Args:
|
||
conversation_id: 对话ID
|
||
|
||
Returns:
|
||
Dict: 对话数据，如果不存在返回None
|
||
"""
|
||
try:
|
||
file_path = self._get_conversation_file_path(conversation_id)
|
||
if not file_path.exists():
|
||
return None
|
||
|
||
with open(file_path, 'r', encoding='utf-8') as f:
|
||
content = f.read().strip()
|
||
if not content:
|
||
return None
|
||
|
||
data = json.loads(content)
|
||
|
||
metadata = data.get("metadata", {})
|
||
if "project_relative_path" not in metadata:
|
||
metadata["project_relative_path"] = None
|
||
self._save_conversation_file(conversation_id, data)
|
||
print(f"🔧 为对话 {conversation_id} 添加相对路径字段")
|
||
|
||
# 向后兼容：确保Token统计结构存在
|
||
if "token_statistics" not in data:
|
||
data["token_statistics"] = self._initialize_token_statistics()
|
||
# 自动保存修复后的数据
|
||
self._save_conversation_file(conversation_id, data)
|
||
print(f"🔧 为对话 {conversation_id} 添加Token统计结构")
|
||
else:
|
||
# 验证现有Token统计数据
|
||
data = self._validate_token_statistics(data)
|
||
|
||
if "run_mode" not in metadata:
|
||
metadata["run_mode"] = "thinking" if metadata.get("thinking_mode") else "fast"
|
||
self._save_conversation_file(conversation_id, data)
|
||
print(f"🔧 为对话 {conversation_id} 添加运行模式字段")
|
||
|
||
# 确保项目快照字段存在（向后兼容）
|
||
changed = False
|
||
if "project_file_tree" not in metadata:
|
||
metadata["project_file_tree"] = None
|
||
changed = True
|
||
if "project_statistics" not in metadata:
|
||
metadata["project_statistics"] = None
|
||
changed = True
|
||
if "project_snapshot_at" not in metadata:
|
||
metadata["project_snapshot_at"] = None
|
||
changed = True
|
||
if changed:
|
||
data["metadata"] = metadata
|
||
self._save_conversation_file(conversation_id, data)
|
||
print(f"🔧 为对话 {conversation_id} 补齐项目快照字段")
|
||
|
||
# 兼容历史口径：旧版本 total_tools 可能把 role=tool 也计入，导致翻倍。
|
||
expected_total_tools = self._count_tools_in_messages(data.get("messages") or [])
|
||
if int(metadata.get("total_tools", 0) or 0) != expected_total_tools:
|
||
metadata["total_tools"] = expected_total_tools
|
||
data["metadata"] = metadata
|
||
self._save_conversation_file(conversation_id, data)
|
||
self._update_index(conversation_id, data)
|
||
print(f"🔧 修正对话 {conversation_id} 的 total_tools 统计为 {expected_total_tools}")
|
||
|
||
# 回填缺失的模型字段：从最近的助手消息元数据推断
|
||
if metadata.get("model_key") is None:
|
||
inferred_model = None
|
||
for msg in reversed(data.get("messages") or []):
|
||
if msg.get("role") != "assistant":
|
||
continue
|
||
mk = (msg.get("metadata") or {}).get("model_key")
|
||
if mk:
|
||
inferred_model = mk
|
||
break
|
||
if inferred_model is not None:
|
||
metadata["model_key"] = inferred_model
|
||
self._save_conversation_file(conversation_id, data)
|
||
print(f"🔧 为对话 {conversation_id} 回填模型字段: {inferred_model}")
|
||
|
||
return data
|
||
except (json.JSONDecodeError, Exception) as e:
|
||
print(f"⌘ 加载对话失败 {conversation_id}: {e}")
|
||
return None
|