Coverage for merco/core/agent.py: 82%

1"""Agent 主循环与核心逻辑""" 

2 

3import json 

4import re 

5import asyncio 

6import logging 

7import shutil 

8import time 

9from typing import Optional 

10from abc import ABC, abstractmethod 

11from rich.console import Console 

12from rich.markdown import Markdown 

13from rich.panel import Panel 

14 

15from .config import MercoConfig 

16from .llm import LLMClient 

17from .session import Session 

18from .message import Message, MessageRole 

19from .context import ContextManager, msg_tokens, estimate_tokens as est_tk 

20from .pipeline import ProcessContext 

21from .interrupt import ( 

22 InterruptCleanupPipeline, CleanupContext, 

23 InjectCancelMessages, TerminateSubprocesses, 

24 CloseMCPConnections, EmitInterruptHooks, SavePartialState 

25) 

26from merco.sandbox.guard import GuardConfirmationRequired, GuardAction 

27 

28console = Console() 

29logger = logging.getLogger("merco.agent") 

30 

31# ── System Prompt 构建器 ───────────────────────────── 

32 

33class PromptChunk: 

34 name: str = "" 

35 def enabled(self, agent) -> bool: return True 

36 def build(self, agent) -> str: raise NotImplementedError 

37 

38class PromptBuilder: 

39 def __init__(self): 

40 self._chunks: list[PromptChunk] = [] 

41 self._disabled: set[str] = set() 

42 def use(self, chunk: PromptChunk) -> "PromptBuilder": 

43 self._chunks.append(chunk); return self 

44 def disable(self, name: str) -> None: self._disabled.add(name) 

45 def enable(self, name: str) -> None: self._disabled.discard(name) 

46 def build(self, agent) -> str: 

47 return "\n\n".join(c.build(agent) for c in self._chunks 

48 if c.name not in self._disabled and c.enabled(agent)) 

49 

50class BasePromptChunk(PromptChunk): 

51 name = "base" 

52 PROMPT = """You are Mercury Code, an AI coding assistant. You can help with: 

53- Writing and editing code 

54- Answering questions about the codebase 

55- Executing shell commands 

56- Searching and managing files 

57 

58When you need to perform actions, use the available tools. 

59Always be concise and helpful.""" 

60 def build(self, agent) -> str: return self.PROMPT 

61 

62class SkillsHintChunk(PromptChunk): 

63 """skill 自动注入：根据当前对话内容匹配相关 skill，注入提示到 system prompt""" 

64 name = "skills_hint" 

65 

66 def enabled(self, agent) -> bool: 

67 return bool(agent.skill_registry and agent.skill_registry.list_skills()) 

68 

69 def build(self, agent) -> str: 

70 registry = agent.skill_registry 

71 if not registry: 

72 return "" 

73 prompt = getattr(agent, "_current_prompt", "") 

74 if not prompt: 

75 return "使用 skill_view 工具可以加载项目相关的技能说明文档。" 

76 

77 relevant = registry.get_relevant(prompt) 

78 if not relevant: 

79 return "" 

80 

81 parts = [] 

82 for skill in relevant: 

83 parts.append( 

84 f"## 相关技能: {skill['name']}\n" 

85 f"{skill['content']}\n" 

86 f"（已自动加载。更多技能用 skill_view 查看。）" 

87 ) 

88 return "\n\n".join(parts) 

89 

90 

91class TimeContextChunk(PromptChunk): 

92 name = "time_context" 

93 def build(self, agent) -> str: 

94 import datetime 

95 now = datetime.datetime.now() 

96 return ( 

97 f"当前时间: {now.strftime('%Y-%m-%d %H:%M:%S')} " 

98 f"{now.astimezone().tzinfo or 'local'} " 

99 f"星期{['一','二','三','四','五','六','日'][now.weekday()]}" 

100 ) 

101 

102 

103def _build_reasoning_panel(text: str) -> Panel: 

104 return Panel(f"[dim]{text.rstrip()}[/dim]", border_style="dim", 

105 title="🧠 思考中…", title_align="left", padding=(0, 1)) 

106 

107 

108class ResponseProvider(ABC): 

109 """响应策略基类 — 工厂模式，Agent 不感知流/非流""" 

110 

111 @abstractmethod 

112 async def get_response(self, agent: "Agent", messages: list, 

113 tools: list | None) -> dict: 

114 ... 

115 

116class NonStreamingProvider(ResponseProvider): 

117 """非流式：一次 chat 返回完整响应""" 

118 

119 async def get_response(self, agent: "Agent", messages: list, 

120 tools: list | None) -> dict: 

121 response = await agent.llm.chat( 

122 messages, tools=tools, tool_choice="auto") 

123 reasoning = response.get("reasoning", "") 

124 if reasoning and reasoning.strip(): 

125 agent._render_reasoning(reasoning) 

126 return response 

127 

128class StreamingProvider(ResponseProvider): 

129 """流式：thinking 用 Live Panel 逐 token 显示，content 不流""" 

130 

131 async def get_response(self, agent: "Agent", messages: list, 

132 tools: list | None) -> dict: 

133 import sys, json as _json 

134 from rich.live import Live 

135 from rich.console import Group 

136 

137 assembled: dict = { 

138 "role": "assistant", "content": "", "reasoning": "", 

139 "tool_calls": [], "finish_reason": None, "usage": None} 

140 reasoning_buf = "" 

141 content_buf = "" 

142 tc_buf: dict[int, dict] = {} 

143 

144 stream_think = agent.config.stream_thinking 

145 render_interval = agent.config.stream_render_interval 

146 _last_render = 0.0 

147 _last_content_update = 0.0 

148 _content_update_interval = 0.3 # 300ms throttle for content panel 

149 

150 # ── 初始等待提示（无 reasoning 时显示"⏳ 思考中…"，有则显示推理文字）── 

151 thinking_panel = Panel("[dim]⏳ 思考中…[/dim]", border_style="dim", 

152 title="🧠 思考中…", title_align="left", padding=(0, 1)) 

153 

154 # ── content 面板延迟创建：收到第一个 content chunk 时才创建 ── 

155 content_panel = None # lazy: created on first content chunk 

156 

157 # ── 使用单个 Live 来显示 thinking 面板（content 面板延迟加入 Group）── 

158 live = Live(Group(thinking_panel), console=console, refresh_per_second=4, 

159 transient=agent.config.stream_thinking_transient) 

160 live.start() 

161 

162 # ── 定时刷新任务：防止 API 返回慢时 thinking 面板卡顿 ── 

163 nonlocal_thinking_panel = [thinking_panel] # mutable ref for closure 

164 nonlocal_content_panel: list[Panel | None] = [None] # mutable ref for closure (lazy init) 

165 

166 def _rebuild_group(): 

167 """Rebuild Group with current panels""" 

168 if nonlocal_content_panel[0] is not None: 

169 return Group(nonlocal_thinking_panel[0], nonlocal_content_panel[0]) 

170 return Group(nonlocal_thinking_panel[0]) 

171 

172 async def _refresh_thinking(): 

173 while True: 

174 await asyncio.sleep(0.5) 

175 if reasoning_buf: 

176 nonlocal_thinking_panel[0] = _build_reasoning_panel(reasoning_buf) 

177 live.update(_rebuild_group()) 

178 refresh_task = asyncio.create_task(_refresh_thinking()) 

179 stream_error: Exception | None = None 

180 

181 try: 

182 stream = agent.llm.chat_stream(messages, tools=tools) 

183 async for chunk in stream: 

184 # 取消检查点：如果任务被取消，立即退出 

185 current = asyncio.current_task() 

186 if current and current.cancelled(): 

187 live.stop() 

188 # TODO: 此 checkpoint 无法覆盖「Cancel 在 __anext__ I/O 等待中到达」的情况 

189 # 补救方案：加 except asyncio.CancelledError 兜底 handler，抽离保存逻辑。 

190 # 优先级：低——窗口极小且用户主动取消，丢失的 partial content 是预期行为。 

191 # 保存部分响应 

192 assembled["reasoning"] = reasoning_buf 

193 assembled["content"] = content_buf 

194 if tc_buf: 

195 assembled["tool_calls"] = [ 

196 {"id": v["id"], "name": v["name"], 

197 "arguments": _json.loads(v["arguments"]) 

198 if v["arguments"] else {}} 

199 for v in (tc_buf[i] for i in sorted(tc_buf)) 

200 ] 

201 # 将部分响应添加到 context 和 session 

202 if reasoning_buf or content_buf or tc_buf: 

203 assistant_msg = { 

204 "role": "assistant", 

205 "content": content_buf, 

206 "reasoning": reasoning_buf, 

207 } 

208 if tc_buf: 

209 assistant_msg["tool_calls"] = assembled["tool_calls"] 

210 logger.debug("StreamingProvider 中断: 将 reasoning(%d chars) 存入 context (这是唯一泄漏窗口)", 

211 len(reasoning_buf)) 

212 agent.context.add(assistant_msg) 

213 agent.session.add_message("assistant", content_buf, 

214 reasoning=reasoning_buf, 

215 tool_calls=assembled.get("tool_calls")) 

216 raise asyncio.CancelledError() 

217 r = chunk.get("reasoning", "") 

218 if r: 

219 reasoning_buf += r 

220 if stream_think: 

221 now = time.monotonic() 

222 if render_interval <= 0 or now - _last_render >= render_interval: 

223 _last_render = now 

224 nonlocal_thinking_panel[0] = _build_reasoning_panel(reasoning_buf) 

225 live.update(_rebuild_group()) 

226 content_buf += chunk.get("content", "") 

227 if content_buf.strip() and agent.config.stream_content: 

228 # Lazy init: create content_panel on first content chunk 

229 if content_panel is None: 

230 content_panel = Panel("", border_style="dim", 

231 title_align="left", padding=(0, 1)) 

232 nonlocal_content_panel[0] = content_panel 

233 # Throttle updates to prevent excessive re-rendering 

234 now = time.monotonic() 

235 if now - _last_content_update >= _content_update_interval: 

236 _last_content_update = now 

237 content_panel.renderable = Markdown(content_buf) 

238 live.update(_rebuild_group()) 

239 for tc in chunk.get("tool_calls", []): 

240 idx = tc["index"] 

241 if idx not in tc_buf: 

242 tc_buf[idx] = { 

243 "id": tc.get("id", ""), 

244 "name": tc.get("name", ""), 

245 "arguments": ""} 

246 if tc.get("id"): tc_buf[idx]["id"] = tc["id"] 

247 if tc.get("name"): tc_buf[idx]["name"] = tc["name"] 

248 tc_buf[idx]["arguments"] += tc.get("arguments", "") 

249 if chunk.get("finish_reason"): 

250 assembled["finish_reason"] = chunk["finish_reason"] 

251 if chunk.get("usage"): 

252 assembled["usage"] = chunk["usage"] 

253 # Final update to ensure all content is displayed 

254 if reasoning_buf: 

255 nonlocal_thinking_panel[0] = _build_reasoning_panel(reasoning_buf) 

256 if content_panel and content_buf.strip(): 

257 content_panel.renderable = Markdown(content_buf) 

258 if reasoning_buf or (content_panel and content_buf.strip()): 

259 live.update(_rebuild_group()) 

260 except asyncio.CancelledError: 

261 raise 

262 except Exception as e: 

263 stream_error = e 

264 logger.warning("StreamingProvider API 错误: %s", e, exc_info=True) 

265 from merco.core.llm.error_ui import classify_error, build_error_panel 

266 info = classify_error(e) 

267 nonlocal_thinking_panel[0] = build_error_panel(info) 

268 nonlocal_content_panel[0] = content_panel # preserve partial content 

269 live.update(_rebuild_group()) 

270 await asyncio.sleep(0.15) 

271 finally: 

272 if 'refresh_task' in locals(): 

273 refresh_task.cancel() 

274 try: 

275 await refresh_task 

276 except asyncio.CancelledError: 

277 pass 

278 transient = agent.config.stream_thinking_transient 

279 if live: 

280 live.stop() 

281 if transient and reasoning_buf: 

282 console.print(_build_reasoning_panel(reasoning_buf)) 

283 if transient and content_buf and agent.config.stream_content: 

284 console.print(Panel(Markdown(content_buf), border_style="dim", 

285 title_align="left", padding=(0, 1))) 

286 # Error path: print static red Panel if needed, then re-raise 

287 if stream_error is not None: 

288 from merco.core.llm.error_ui import classify_error, build_error_panel 

289 need_static = transient or (not reasoning_buf and not content_buf) 

290 if need_static: 

291 console.print(build_error_panel(classify_error(stream_error))) 

292 agent._error_displayed_in_stream = True 

293 raise stream_error 

294 

295 assembled["reasoning"] = reasoning_buf 

296 assembled["content"] = content_buf 

297 if tc_buf: 

298 assembled["tool_calls"] = [ 

299 {"id": v["id"], "name": v["name"], 

300 "arguments": _json.loads(v["arguments"]) 

301 if v["arguments"] else {}} 

302 for v in (tc_buf[i] for i in sorted(tc_buf)) 

303 ] 

304 logger.debug( 

305 "stream done: finish=%s content=%d reasoning=%d tool_calls=%d%s", 

306 assembled.get("finish_reason"), len(assembled["content"]), 

307 len(assembled["reasoning"]), len(assembled["tool_calls"]), 

308 f" {[tc['name'] for tc in assembled['tool_calls']]}" if assembled["tool_calls"] else "") 

309 return assembled 

310 

311class Agent: 

312 """AI Agent 核心类，负责对话循环与工具调度""" 

313 

314 

315 def __init__(self, config: MercoConfig, tool_registry=None): 

316 self.config = config 

317 self.session = Session() 

318 from merco.sandbox import snapshot 

319 snapshot.set_current_session(self.session.id) 

320 self.context = ContextManager(max_tokens=config.max_input_tokens) 

321 self.tool_registry = tool_registry 

322 self.skill_registry = None 

323 

324 # 初始化 LLM 客户端 

325 api_key = config.model.api_key or self._get_api_key(config.model.provider) 

326 self.llm = LLMClient( 

327 api_key=api_key, 

328 model=config.model.model, 

329 base_url=config.model.base_url, 

330 temperature=config.model.temperature, 

331 max_tokens=config.model.max_tokens, 

332 cooldown=0.3, # 请求冷却（秒），0=禁用；共享网关可调大 

333 extra_params=config.model.extra_params, 

334 headers=config.model.headers, 

335 stream_options=config.model.stream_options, 

336 ) 

337 

338 self._tool_calls_count = 0 

339 self._max_tool_calls = self.config.max_tool_calls 

340 self._current_prompt = "" 

341 

342 # Flag set by response providers when they've already displayed an error 

343 # panel inline (avoids duplicate Panel print at REPL layer). 

344 self._error_displayed_in_stream = False 

345 

346 # ── 可观察性 ── 

347 from merco.hooks.registry import HookRegistry 

348 from merco.observability.observer import Observer 

349 self.hooks = HookRegistry() 

350 self.observer = Observer(self.hooks) 

351 

352 # ── 守卫：敏感命令执行前确认 ── 

353 from merco.sandbox.guard import ( 

354 ToolGuard, PolicyPipeline, BuiltinDefaultPolicy 

355 ) 

356 self._security_pipeline = PolicyPipeline() 

357 self._security_pipeline.use(BuiltinDefaultPolicy( 

358 mode=config.sandbox_mode, 

359 user_rules=config.sandbox_rules, 

360 )) 

361 self.guard = ToolGuard(pipeline=self._security_pipeline) 

362 

363 # ── Middleware：Guard + EditApply + ErrorHandling 装配到 ToolRegistry ── 

364 from merco.tools.middleware import GuardMiddleware, EditApplyMiddleware, ErrorHandlingMiddleware 

365 self.tool_registry.use(GuardMiddleware(self.guard)) 

366 self.tool_registry.use(EditApplyMiddleware(diff_view=config.diff_view)) 

367 self.tool_registry.use(ErrorHandlingMiddleware()) 

368 

369 # ── 会话持久化 ── 

370 from merco.memory.session_store import SessionStore 

371 from merco.memory.session_search import SessionSearch 

372 self._session_store = SessionStore(_get_db_path()) 

373 self._search = SessionSearch(self._session_store) 

374 self.session = Session.resume_or_create(self._session_store) 

375 # _restore_context() 在 Agent.create()._initialize_async_plugins() 中执行 

376 

377 # ── 工厂：根据 config 选响应策略 ── 

378 if self.config.streaming: 

379 self._provider: ResponseProvider = StreamingProvider() 

380 else: 

381 self._provider: ResponseProvider = NonStreamingProvider() 

382 

383 # ── Pipeline 初始化 ── 

384 from .pipeline import ResultPipeline, RecoveryPipeline, EmptyResponsePipeline 

385 from merco.tools.processors.truncation import TruncationProcessor 

386 from merco.skills.processors import SkillViewProcessor 

387 from merco.core.recovery.wait import WaitRecovery 

388 from merco.context.recovery import ContextCompressRecovery 

389 from merco.core.empty_response import CallbackEmptyResponse 

390 self.result_pipeline = ResultPipeline() 

391 self.result_pipeline.use(TruncationProcessor(max_bytes=16000)) 

392 self.result_pipeline.use(SkillViewProcessor()) 

393 self.recovery_pipeline = RecoveryPipeline() 

394 self.recovery_pipeline.use(WaitRecovery(delay=3.0)) 

395 self.recovery_pipeline.use(ContextCompressRecovery()) 

396 self.empty_response_pipeline = EmptyResponsePipeline() 

397 self.empty_response_pipeline.use(CallbackEmptyResponse()) 

398 

399 # ── Prompt 构建器 ── 

400 self.prompt_builder = PromptBuilder() 

401 self.prompt_builder.use(BasePromptChunk()) 

402 self.prompt_builder.use(SkillsHintChunk()) 

403 self.prompt_builder.use(TimeContextChunk()) 

404 

405 # ── Memory 召回 ── 

406 from merco.memory.recall import HybridRecaller, FTS5Recaller, MemoryRecaller 

407 from merco.memory.store import MemoryStore 

408 from merco.memory.backend import MemoryBackendRegistry 

409 from merco.memory.backends.json_backend import JSONBackend 

410 

411 self.memory_backends = MemoryBackendRegistry() 

412 self.memory_backends.register(JSONBackend(config.memory_path)) 

413 

414 backend_name = config.memory_backend or "json" 

415 selected_backend = self.memory_backends.get(backend_name) or self.memory_backends.get("json") 

416 

417 _fts5 = FTS5Recaller(self._search) 

418 _mem = MemoryRecaller(MemoryStore(backend=selected_backend)) 

419 self.recaller = ( 

420 HybridRecaller(limit=config.memory_recall_limit, max_chars=config.memory_recall_max_chars) 

421 .add(_fts5) 

422 .add(_mem) 

423 ) 

424 

425 # ── Memory 保存链（让 /remember 和 session 结束抽取可写入）── 

426 from merco.memory.save_pipeline import MemorySavePipeline 

427 from merco.memory.strategy import ( 

428 ExplicitRememberStrategy, SessionEndExtractStrategy, 

429 ) 

430 

431 self._memory_store = MemoryStore(backend=selected_backend) 

432 self.memory_save_pipeline = MemorySavePipeline( 

433 store=self._memory_store, 

434 hooks=self.hooks, 

435 ) 

436 self.memory_strategies = [ 

437 ExplicitRememberStrategy(self.memory_save_pipeline), 

438 ] 

439 if self.config.memory_auto_extract_on_session_end: 

440 self.memory_strategies.append( 

441 SessionEndExtractStrategy( 

442 self.memory_save_pipeline, self.llm, 

443 session_store=self._session_store, 

444 max_per_session=self.config.memory_extract_max_per_session, 

445 min_messages=self.config.memory_extract_min_messages, 

446 ) 

447 ) 

448 for strat in self.memory_strategies: 

449 strat.subscribe(self.hooks) 

450 

451 # ── 插件系统 ── 

452 from merco.plugins.base import PluginContext 

453 from merco.plugins.manager import PluginManager 

454 from merco.plugins.builtin.observability.plugin import ObservabilityPlugin 

455 from merco.plugins.builtin.skills.plugin import SkillPlugin 

456 from merco.plugins.builtin.mcp.plugin import MCPPlugin 

457 from merco.plugins.builtin.subagent.plugin import SubAgentPlugin 

458 from merco.plugins.builtin.web.plugin import WebPlugin 

459 from merco.plugins.builtin.scheduler.plugin import SchedulerPlugin 

460 from merco.plugins.builtin.superpower.plugin import SuperpowerPlugin 

461 

462 # ── Context Pipeline ── 

463 from merco.context.pipeline import ContextPipeline 

464 from merco.context.processors.compress import CompressProcessor 

465 from merco.context.processors.cache_optimize import CacheOptimizeProcessor 

466 

467 self.context_pipeline = ContextPipeline() 

468 self.context_pipeline.use(CacheOptimizeProcessor()) 

469 self.context_pipeline.use(CompressProcessor( 

470 max_tokens=config.max_input_tokens, 

471 threshold=config.compression_threshold, 

472 )) 

473 

474 # ── AgentProfile Registry ── 

475 from merco.agents.profile import AgentProfileRegistry, BUILTIN_PROFILES 

476 

477 self.agent_profiles = AgentProfileRegistry() 

478 for p in BUILTIN_PROFILES: 

479 self.agent_profiles.register(p) 

480 

481 # Todo + SubAgent 由 SubAgentPlugin 激活时接管 

482 from merco.todo.manager import TodoManager 

483 from merco.agents.subagent import SubAgentManager 

484 self.todo_manager = TodoManager(f"{config.memory_path}/../todos.db") 

485 self.sub_agent_manager = SubAgentManager(self, self.agent_profiles) 

486 

487 # ── Loop Policy ── 

488 from merco.core.loop_policy import LoopPolicyRegistry, DefaultLoopPolicy 

489 self.loop_policies = LoopPolicyRegistry() 

490 self.loop_policies.register(DefaultLoopPolicy()) 

491 self.loop_policies.set_active("default") 

492 

493 self._plugin_ctx = PluginContext( 

494 hooks=self.hooks, 

495 tool_registry=self.tool_registry, 

496 prompt_builder=self.prompt_builder, 

497 recovery_pipeline=self.recovery_pipeline, 

498 result_pipeline=self.result_pipeline, 

499 memory_save_pipeline=self.memory_save_pipeline, 

500 recaller=self.recaller, 

501 config=config, 

502 observer=self.observer, 

503 todo_manager=self.todo_manager, 

504 sub_agent_manager=self.sub_agent_manager, 

505 context_pipeline=self.context_pipeline, 

506 memory_backends=self.memory_backends, 

507 agent_profiles=self.agent_profiles, 

508 loop_policies=self.loop_policies, 

509 ) 

510 self._plugin_ctx.agent = self 

511 self.plugin_manager = PluginManager(self._plugin_ctx) 

512 

513 # 注册内置插件 

514 self.plugin_manager.register(ObservabilityPlugin()) 

515 self.plugin_manager.register(SkillPlugin()) 

516 self.plugin_manager.register(MCPPlugin()) 

517 self.plugin_manager.register(SubAgentPlugin()) 

518 self.plugin_manager.register(WebPlugin()) 

519 self.plugin_manager.register(SchedulerPlugin()) 

520 self.plugin_manager.register(SuperpowerPlugin()) 

521 

522 # ── MCP 客户端（由 MCPPlugin 激活时创建；legacy 路径下保持 None）── 

523 self.mcp_manager = None 

524 

525 # ── 中断清理管线 ── 

526 self._cleanup_pipeline = (InterruptCleanupPipeline() 

527 .use(InjectCancelMessages()) 

528 .use(TerminateSubprocesses()) 

529 .use(CloseMCPConnections()) 

530 .use(EmitInterruptHooks()) 

531 .use(SavePartialState())) 

532 

533 @classmethod 

534 async def create(cls, config: MercoConfig, tool_registry=None) -> "Agent": 

535 """Create an Agent with deterministic async plugin initialization.""" 

536 agent = cls(config=config, tool_registry=tool_registry) 

537 await agent._initialize_async_plugins() 

538 return agent 

539 

540 async def _initialize_async_plugins(self) -> None: 

541 """Initialize plugins in deterministic order for Agent.create().""" 

542 await self.plugin_manager.activate("observability") 

543 self.observer = self._plugin_ctx.observer 

544 assert self.observer is not None 

545 self._restore_context() 

546 await self.plugin_manager.activate("skills") 

547 await self.plugin_manager.activate("mcp") 

548 await self.plugin_manager.activate("subagent") 

549 await self.plugin_manager.activate("web") 

550 await self.plugin_manager.activate("scheduler") 

551 await self.plugin_manager.activate_all() 

552 

553 async def run(self, prompt: str) -> str: 

554 """执行一次 Agent 循环""" 

555 self._current_prompt = prompt 

556 self._error_displayed_in_stream = False 

557 

558 # ── 生命周期事件：session.create（首次激活时）── 

559 await self.hooks.emit("session.create", session_id=self.session.id) 

560 # ── 生命周期事件：agent.start ── 

561 await self.hooks.emit("agent.start", session_id=self.session.id) 

562 

563 # 添加用户消息 

564 self.session.add_message("user", prompt) 

565 self.context.add({"role": "user", "content": prompt}) 

566 

567 # ── 消息事件：message.receive ── 

568 await self.hooks.emit("message.receive", message=prompt) 

569 

570 tools = self.tool_registry.get_definitions() if self.tool_registry else [] 

571 self.context.set_overhead(await self._build_system_prompt(), len(tools)) 

572 if self.context.needs_compression(): 

573 await self._compress_context() 

574 

575 # 执行 Agent 循环 

576 try: 

577 result = await self._agent_loop() 

578 # 正常结束时：保存 session 

579 self._auto_title(prompt) 

580 self.session.metadata["observer"] = self.observer.snapshot() 

581 self.session.save() 

582 self._session_store.save_metadata(self.session.id, self.session.metadata) 

583 await self.hooks.emit("conversation.turn") 

584 except asyncio.CancelledError: 

585 # 使用 InterruptCleanupPipeline 替换 _inject_interrupted_tool_results 

586 cancelled_tool_calls = self._find_orphan_tool_calls() 

587 cleanup_ctx = CleanupContext( 

588 agent=self, 

589 cancelled_tool_calls=cancelled_tool_calls, 

590 session_id=self.session.id, 

591 ) 

592 await self._cleanup_pipeline.process(cleanup_ctx) 

593 raise 

594 finally: 

595 # ── 生命周期事件：agent.stop（所有退出路径都触发）── 

596 await self.hooks.emit("agent.stop", session_id=self.session.id) 

597 # ── 生命周期事件：session.destroy（清理 session）── 

598 await self.hooks.emit("session.destroy", session_id=self.session.id) 

599 return result 

600 

601 def _restore_context(self): 

602 """清空上下文，然后从持久化会话恢复消息""" 

603 self.context = ContextManager(max_tokens=self.config.max_input_tokens) 

604 snap = self.session.metadata.get("observer") 

605 if snap: 

606 self.observer.restore(snap) 

607 

608 checkpoint = self.session.metadata.get("compress_checkpoint") 

609 if checkpoint: 

610 # Restore from checkpoint: summary + tail only 

611 summary = checkpoint.get("summary", "") 

612 tail_count = checkpoint.get("tail_count", 5) 

613 original_count = checkpoint.get("original_count", 0) 

614 all_msgs = self.session.messages 

615 

616 # 如果之后新增了大量消息，checkpoint 已过时 → 全量恢复后重新压缩 

617 if original_count > 0 and len(all_msgs) > original_count + 20: 

618 logger.debug("_restore_context: checkpoint 过时 (original=%d now=%d)，全量恢复", 

619 original_count, len(all_msgs)) 

620 del self.session.metadata["compress_checkpoint"] 

621 # fall through to full restore below 

622 else: 

623 tail = all_msgs[-tail_count * 2:] if len(all_msgs) > tail_count * 2 else all_msgs 

624 

625 if summary: 

626 self.context.add({"role": "system", "content": summary}) 

627 for msg in tail: 

628 entry = {"role": msg["role"], "content": msg.get("content", "")} 

629 if "tool_call_id" in msg: 

630 entry["tool_call_id"] = msg["tool_call_id"] 

631 if msg.get("tool_calls"): 

632 entry["tool_calls"] = msg["tool_calls"] 

633 self.context.add(entry) 

634 return 

635 

636 for msg in self.session.messages: 

637 r = msg.get("reasoning", "") 

638 if r: 

639 logger.debug("_restore_context: session 消息含 reasoning (%d chars, 已丢弃)", 

640 len(r)) 

641 entry = {"role": msg["role"], "content": msg.get("content", "")} 

642 if "tool_call_id" in msg: 

643 entry["tool_call_id"] = msg["tool_call_id"] 

644 if msg.get("tool_calls"): 

645 entry["tool_calls"] = msg["tool_calls"] 

646 self.context.add(entry) 

647 

648 def _auto_title(self, user_input: str): 

649 """用首条用户消息的前 40 字作为会话标题""" 

650 if self.session.title or not user_input: 

651 return 

652 title = user_input.strip()[:40] 

653 self.session.title = title 

654 self._session_store.update_title(self.session.id, title) 

655 

656 def _wrap_up_messages(self, messages): 

657 """构造收尾消息列表：追加总结请求。""" 

658 return messages + [{ 

659 "role": "user", 

660 "content": "已达到最大工具调用次数。请基于已有信息给出最终回复，不要再调用工具。" 

661 }] 

662 

663 async def _wrap_up_call(self, messages): 

664 """收尾调用：无工具文字回应。""" 

665 try: 

666 resp = await self.llm.chat(messages, tools=[], tool_choice="none") 

667 except Exception: 

668 return "模型调用失败" 

669 content = resp.get("content", "") or "已达到调用上限。" 

670 self.session.add_message("assistant", content) 

671 self.context.add({"role": "assistant", "content": content}) 

672 return content 

673 

674 

675 async def _agent_loop(self) -> str: 

676 """Agent 主循环。工具预算耗尽时直接收尾。""" 

677 self._tool_calls_count = 0 

678 self._max_tool_calls = self.config.max_tool_calls 

679 _empty_retries = 0 

680 _recovery_attempts = 0 

681 tools = [] 

682 

683 while True: 

684 messages = await self._build_messages() 

685 tools = list(self.tool_registry.get_definitions() if self.tool_registry else []) 

686 

687 if self._tool_calls_count >= self._max_tool_calls: 

688 return await self._wrap_up_call(self._wrap_up_messages(messages)) 

689 

690 logger.debug("→ Agent 循环 #%d: %d 条消息", self._tool_calls_count, len(messages)) 

691 t0 = time.monotonic() 

692 try: 

693 before = await self.hooks.emit("llm.before_chat", messages=messages, tools=tools) 

694 if before and before.data: 

695 messages = before.data.get("messages", messages) 

696 tools = before.data.get("tools", tools) 

697 if before.stop: 

698 response = before.data["response"] 

699 else: 

700 response = await self._provider.get_response( 

701 self, messages, tools or None) 

702 else: 

703 response = await self._provider.get_response( 

704 self, messages, tools or None) 

705 except Exception as e: 

706 _recovery_attempts += 1 

707 if _recovery_attempts > 3: 

708 from merco.core.llm.errors import llm_error 

709 return llm_error(e) 

710 from openai import APIStatusError 

711 from .pipeline import RecoveryContext 

712 ctx = RecoveryContext( 

713 error=e, 

714 status_code=e.status_code if isinstance(e, APIStatusError) else 0,