Coverage for merco/core/session.py: 90%

1"""会话管理""" 

2 

3import logging 

4import uuid 

5from datetime import datetime 

6 

7_logger = logging.getLogger("merco.session") 

8 

9 

10class Session: 

11 """单次对话会话 — 数据容器，对接 SessionStore 持久化""" 

12 

13 def __init__(self, session_id: str = None, title: str = "", store=None): 

14 self.id = session_id or _new_id() 

15 self.title = title 

16 self.messages: list[dict] = [] 

17 self.metadata: dict = {} 

18 self._store = store 

19 self._dirty = False # 有未持久化的消息 

20 

21 # ── 消息 ────────────────────────────────────────────── 

22 

23 def add_message(self, role: str, content: str, **kwargs): 

24 """添加消息。不立即写磁盘（由 agent 循环结束时统一 save）""" 

25 r = kwargs.get("reasoning", "") 

26 if r: 

27 _logger.debug("add_message(%s): reasoning=%d chars, content=%d chars", 

28 role, len(r), len(content)) 

29 msg = {"role": role, "content": content} 

30 msg.update(kwargs) 

31 self.messages.append(msg) 

32 self._dirty = True 

33 

34 def get_history(self) -> list[dict]: 

35 return self.messages.copy() 

36 

37 # ── 持久化 ──────────────────────────────────────────── 

38 

39 def save(self): 

40 """将未持久化的消息写入 SQLite。增量：DB 已有 N 条，只写 messages[N:]""" 

41 if not self._store or not self._dirty: 

42 return 

43 

44 self._store.create_session(self.id, self.title) 

45 

46 existing = self._store.count_messages(self.id) 

47 for msg in self.messages[existing:]: 

48 self._store.save_message( 

49 session_id=self.id, 

50 role=msg.get("role", ""), 

51 content=msg.get("content", ""), 

52 tool_call_id=msg.get("tool_call_id", ""), 

53 tool_calls=msg.get("tool_calls"), 

54 reasoning=msg.get("reasoning", ""), 

55 ) 

56 self._dirty = False 

57 

58 def delete(self): 

59 if self._store: 

60 self._store.delete_session(self.id) 

61 

62 # ── 工厂 ────────────────────────────────────────────── 

63 

64 @classmethod 

65 def load(cls, session_id: str, store) -> "Session | None": 

66 """从 store 加载完整会话（含历史消息）""" 

67 data = store.load_session(session_id) 

68 if not data: 

69 return None 

70 

71 s = cls(session_id=data["id"], title=data["title"], store=store) 

72 s.messages = data["messages"] 

73 s.metadata = data.get("metadata", {}) 

74 s._dirty = False 

75 return s 

76 

77 @classmethod 

78 def fork(cls, session_id: str, store, title: str = None) -> "Session | None": 

79 """从 session_id 克隆新会话。返回新 Session 或 None。""" 

80 try: 

81 new_id = store.clone_session(session_id) 

82 except ValueError: 

83 return None 

84 if title is not None: 

85 store.set_title(new_id, title) 

86 return cls.load(new_id, store) 

87 

88 @classmethod 

89 def resume_or_create(cls, store, session_id: str = None) -> "Session": 

90 """恢复指定会话，或自动恢复上次，或新建""" 

91 if session_id: 

92 s = cls.load(session_id, store) 

93 if s: 

94 return s 

95 # 自动恢复上次会话 

96 recent = store.list_sessions(limit=1) 

97 if recent: 

98 s = cls.load(recent[0]["id"], store) 

99 if s: 

100 return s 

101 # 新建 

102 s = cls(store=store) 

103 store.create_session(s.id) 

104 return s 

105 

106 

107def _new_id() -> str: 

108 return str(uuid.uuid4())[:8]