检测中断类型。 CC 的恢复逻辑: 根据最后一条消息的 role 判断: - "user" → 用户发了消息但 AI 没回复 - "tool" → 工具执行完但 AI 没继续 - "assistant" → 正常结束,无中断 - None → 空会话
(self, session_id: str)
| 122 | ) |
| 123 | |
| 124 | def detect_interruption(self, session_id: str) -> Optional[str]: |
| 125 | """检测中断类型。 |
| 126 | |
| 127 | CC 的恢复逻辑: 根据最后一条消息的 role 判断: |
| 128 | - "user" → 用户发了消息但 AI 没回复 |
| 129 | - "tool" → 工具执行完但 AI 没继续 |
| 130 | - "assistant" → 正常结束,无中断 |
| 131 | - None → 空会话 |
| 132 | """ |
| 133 | path = self._session_path(session_id) |
| 134 | entries = self._read_entries(path) |
| 135 | if not entries: |
| 136 | return None |
| 137 | |
| 138 | chain = self._rebuild_chain(entries) |
| 139 | if not chain: |
| 140 | return None |
| 141 | |
| 142 | last_role = chain[-1].message.get("role", "") |
| 143 | if last_role in ("user", "tool"): |
| 144 | return last_role |
| 145 | return None |
| 146 | |
| 147 | def _append_entry(self, path: Path, entry: StorageEntry) -> None: |
| 148 | """追加一行 JSONL。 |
nothing calls this directly
no test coverage detected