重构(运行器): 拆分图片提示词与工具运行职责

保留 AgentRunner 的公开入口、属性和调用语义，将图片优化、提示词组装、工具策略执行拆入独立内部组件；同时将 KnowledgeConfig 移入核心配置模块并保留旧知识工具使用方式，补充组件级回归测试。

agent/agent/core/knowledge_config.py

@@ -0,0 +1,71 @@
+"""Knowledge extraction and injection configuration.
+
+This value object belongs to the runner configuration layer.  Keeping it out
+of the builtin HTTP tool module prevents core configuration imports from
+triggering builtin tool registration.
+"""
+
+from __future__ import annotations
+
+import subprocess
+from dataclasses import dataclass
+from typing import Dict, List, Optional
+
+from agent.core.prompts import COMPLETION_REFLECT_PROMPT, build_reflect_prompt
+
+
+@dataclass
+class KnowledgeConfig:
+    """Configure knowledge extraction, review, and prompt injection."""
+
+    enable_extraction: bool = True
+    reflect_prompt: str = ""
+    enable_completion_extraction: bool = True
+    completion_reflect_prompt: str = ""
+    reflect_auto_commit: bool = False
+    enable_injection: bool = True
+    owner: str = ""
+    default_tags: Optional[Dict[str, str]] = None
+    default_scopes: Optional[List[str]] = None
+    default_search_types: Optional[List[str]] = None
+    default_search_owner: str = ""
+
+    def get_reflect_prompt(self) -> str:
+        """Return the configured compression-reflection prompt."""
+
+        return self.reflect_prompt or build_reflect_prompt()
+
+    def get_completion_reflect_prompt(self) -> str:
+        """Return the configured completion-reflection prompt."""
+
+        return self.completion_reflect_prompt or COMPLETION_REFLECT_PROMPT
+
+    @property
+    def resolved_owner(self) -> str:
+        """Resolve owner from explicit config, Git identity, then ``agent``."""
+
+        if self.owner:
+            return self.owner
+        try:
+            result = subprocess.run(
+                ["git", "config", "user.email"],
+                capture_output=True,
+                text=True,
+                timeout=2,
+            )
+            if result.returncode == 0 and result.stdout.strip():
+                return result.stdout.strip()
+        except (OSError, subprocess.SubprocessError):
+            pass
+        return "agent"
+
+    def get_owner(self, agent_id: str = "agent") -> str:
+        """Return the resolved owner, namespaced for non-default agents."""
+
+        owner = self.resolved_owner
+        if owner == "agent" and agent_id != "agent":
+            return f"agent:{agent_id}"
+        return owner
+
+
+__all__ = ["KnowledgeConfig"]

agent/agent/core/runner.py

@@ -1,17 +1,8 @@
-"""
-Agent Runner - Agent 执行引擎
-
-核心职责：
-1. 执行 Agent 任务（循环调用 LLM + 工具）
-2. 记录执行轨迹（Trace + Messages + GoalTree）
-3. 加载和注入技能（Skill）
-4. 管理执行计划（GoalTree）
-5. 支持续跑（continue）和回溯重跑（rewind）
-
-参数分层：
-- Infrastructure: AgentRunner 构造时设置（trace_store, llm_call 等）
-- RunConfig: 每次 run 时指定（model, trace_id, after_sequence 等）
-- Messages: OpenAI SDK 格式的任务消息
+"""Public execution facade for Agent loops and resumable Trace state.
+
+``AgentRunner`` retains the stable calling surface while delegating prompt and
+tool-policy concerns to focused internal components.  It still owns loop state,
+Trace continuity, context management, and rewind semantics.
 """
 
 import asyncio
@@ -48,28 +39,22 @@ from agent.trace.compaction import (
     estimate_tokens,
 )
 from agent.skill.models import Skill
-from agent.skill.skill_loader import load_skills_from_dir
 from agent.tools import ToolRegistry, get_tool_registry
-from agent.tools.builtin.knowledge import KnowledgeConfig
+from agent.core.knowledge_config import KnowledgeConfig
 from agent.core.memory import MemoryConfig
-from agent.core.presets import get_preset
 from agent.orchestration.models import AgentRole, CompletionPolicy
 from agent.orchestration.policy import (
     DefaultToolPolicy,
     ResolvedAgentPolicy,
-    PLANNER_TOOLS,
-    WORKER_TOOLS,
-    VALIDATOR_TOOLS,
 )
+from agent.core.runner_tools import RunnerToolRuntime
+from agent.core.runner_prompts import RunnerPromptRuntime
+from agent.core.runner_images import RunnerImageRuntime
 from agent.core.prompts import (
-    DEFAULT_SYSTEM_PREFIX,
     TRUNCATION_HINT,
     AGENT_INTERRUPTED_SUMMARY,
-    TASK_NAME_GENERATION_SYSTEM_PROMPT,
-    TASK_NAME_FALLBACK,
     build_tool_interrupted_message,
     build_agent_continue_hint,
-    ROLE_CONTRACTS,
 )
 
 logger = logging.getLogger(__name__)
@@ -197,8 +182,7 @@ class RunConfig:
     # None = 默认 Agent（无长期记忆）；赋值 MemoryConfig 使该 Agent 成为 memory-bearing Agent
     memory: Optional["MemoryConfig"] = None
 
-    # BUILTIN_TOOLS 硬编码列表已移除（2026-04）。
-    # 工具可用性现在由 @tool(groups=[...]) 声明 + RunConfig.tool_groups 过滤控制。
+    # Tool availability is resolved from registry groups and role policy.
 
 
 @dataclass
@@ -281,12 +265,17 @@ class AgentRunner:
         # 图片优化缓存（避免重复处理）
         # key: 图片内容的 hash, value: {"downscaled": ..., "description": ...}
         self._image_opt_cache: Dict[str, Dict[str, Any]] = {}
+        self._image_runtime = RunnerImageRuntime(self)
 
         # 当前 run 的 MemoryConfig（由 run() 根据 RunConfig.memory 设置）
         # dream 工具从 context.runner 读取此字段，判断是否 memory-bearing
         self._current_memory_config: Optional[MemoryConfig] = None
         self.task_coordinator = task_coordinator
         self.tool_policy = tool_policy or DefaultToolPolicy()
+        self._tool_runtime = RunnerToolRuntime(self.tools, self.tool_policy, self.log)
+        self._prompt_runtime = RunnerPromptRuntime(
+            logger=self.log,
+        )
 
     # ===== 核心公开方法 =====
 
@@ -713,7 +702,7 @@ class AgentRunner:
 
             await broadcast_trace_status_changed(config.trace_id, "running")
         except Exception:
-            pass
+            self.log.debug("Trace status broadcast failed", exc_info=True)
 
         return trace_obj, goal_tree, sequence
 
@@ -1279,7 +1268,7 @@ class AgentRunner:
 
                         await broadcast_trace_status_changed(trace_id, "stopped")
                     except Exception:
-                        pass
+                        self.log.debug("Stopped status broadcast failed", exc_info=True)
                     trace_obj = await self.trace_store.get_trace(trace_id)
                     if trace_obj:
                         yield trace_obj
@@ -2301,8 +2290,11 @@ class AgentRunner:
                                         }
                                     )
                             img_count = len(tool_content_for_llm) - 1  # 减去 text 块
-                            print(
-                                f"[Runner] 多模态工具反馈: tool={tool_name}, images={img_count}, text_len={len(tool_result_text)}"
+                            self.log.debug(
+                                "Multimodal tool result: tool=%s images=%d text_len=%d",
+                                tool_name,
+                                img_count,
+                                len(tool_result_text),
                             )
                         else:
                             tool_result_text = tool_text
@@ -3184,645 +3176,47 @@ class AgentRunner:
     # ===== 辅助方法 =====
 
     async def _optimize_images(self, messages: List[Dict], model: str) -> List[Dict]:
-        """
-        分级优化已处理的图片，节省 token
-
-        策略（基于图片距离最后一条 assistant 的"轮次"）：
-        1. 最近 1-2 轮：保留原图
-        2. 3-5 轮：降低分辨率和压缩（节省 token 但保留视觉信息）
-        3. 5 轮以上：调用小模型生成文本描述 + 保留 URL
-
-        处理结果会缓存，避免重复的 PIL 解码/编码和 LLM 调用。
-
-        Args:
-            messages: 原始消息列表
-            model: 当前使用的模型（用于选择描述生成模型）
-
-        Returns:
-            优化后的消息列表（深拷贝）
-        """
-        if not messages:
-            return messages
-
-        # 找到最后一条 assistant message 的位置
-        last_assistant_idx = -1
-        for i in range(len(messages) - 1, -1, -1):
-            if messages[i].get("role") == "assistant":
-                last_assistant_idx = i
-                break
-
-        # 如果没有 assistant message，说明还没开始对话，不优化
-        if last_assistant_idx == -1:
-            return messages
-
-        # 统计从每个位置到最后一条 assistant 之间的 assistant 数量（作为"轮次"）
-        assistant_count_after = [0] * len(messages)
-        count = 0
-        for i in range(len(messages) - 1, -1, -1):
-            assistant_count_after[i] = count
-            if messages[i].get("role") == "assistant":
-                count += 1
-
-        # 深拷贝避免修改原始数据
-        import copy
-        import hashlib
-        import asyncio
-        import base64 as b64mod
-        import httpx
-        import mimetypes
-
-        messages = copy.deepcopy(messages)
-
-        # 预处理：将所有 HTTP(S) URL 图片下载并转为 base64 data URL
-        # Qwen API 无法访问外部签名 URL（如 BFL、火山引擎 TOS），必须在本地转换
-        url_download_jobs = []  # [(msg_idx, block_idx, url)]
-        for i, msg in enumerate(messages):
-            if msg.get("role") != "tool":
-                continue
-            content = msg.get("content")
-            if not isinstance(content, list):
-                continue
-            for block_idx, block in enumerate(content):
-                if isinstance(block, dict) and block.get("type") == "image_url":
-                    url = block.get("image_url", {}).get("url", "")
-                    if url.startswith(("http://", "https://")):
-                        url_download_jobs.append((i, block_idx, url))
-
-        if url_download_jobs:
-
-            async def _download_image_to_data_url(url: str) -> str | None:
-                try:
-                    async with httpx.AsyncClient(timeout=60, trust_env=False) as client:
-                        resp = await client.get(url)
-                        resp.raise_for_status()
-                        ct = resp.headers.get("content-type", "").split(";")[0].strip()
-                        if not ct.startswith("image/"):
-                            ct = (
-                                mimetypes.guess_type(url.split("?")[0])[0]
-                                or "image/png"
-                            )
-                        b64 = b64mod.b64encode(resp.content).decode()
-                        return f"data:{ct};base64,{b64}"
-                except Exception:
-                    return None
-
-            results = await asyncio.gather(
-                *[_download_image_to_data_url(url) for _, _, url in url_download_jobs],
-                return_exceptions=True,
-            )
-            converted = 0
-            for (msg_idx, block_idx, original_url), result in zip(
-                url_download_jobs, results
-            ):
-                if isinstance(result, str) and result.startswith("data:"):
-                    messages[msg_idx]["content"][block_idx]["image_url"]["url"] = result
-                    converted += 1
-            if converted:
-                self.log.info(
-                    f"[Image Optimization] URL→base64 预转换: {converted}/{len(url_download_jobs)} 张"
-                )
-
-        # 统计优化情况
-        stats = {"kept": 0, "downscaled": 0, "described": 0, "cache_hit": 0}
-
-        # 收集需要降分辨率或尺寸补齐的图片（用于并发处理）
-        process_jobs = []  # [(msg_idx, block_idx, image_url, cache_key, max_size, cache_field)]
-
-        # 第一遍：扫描并收集需要处理的图片
-        for i in range(last_assistant_idx):
-            msg = messages[i]
-            if msg.get("role") != "tool":
-                continue
-
-            content = msg.get("content")
-            if not isinstance(content, list):
-                continue
-
-            rounds_ago = assistant_count_after[i]
-
-            for block_idx, block in enumerate(content):
-                if isinstance(block, dict) and block.get("type") == "image_url":
-                    image_url_obj = block.get("image_url", {})
-                    image_url = image_url_obj.get("url", "")
-
-                    if image_url.startswith("data:"):
-                        cache_key = hashlib.md5(image_url[:200].encode()).hexdigest()
-                    else:
-                        cache_key = hashlib.md5(image_url.encode()).hexdigest()
-
-                    # 1-5 轮都需要检查尺寸
-                    if rounds_ago <= 5:
-                        cached = self._image_opt_cache.get(cache_key, {})
-                        cache_field = "pad_only" if rounds_ago <= 2 else "downscaled"
-
-                        if cache_field not in cached and image_url.startswith("data:"):
-                            max_size = None if rounds_ago <= 2 else 512
-                            process_jobs.append(
-                                (
-                                    i,
-                                    block_idx,
-                                    image_url,
-                                    cache_key,
-                                    max_size,
-                                    cache_field,
-                                )
-                            )
-
-        # 并发处理所有尺寸任务
-        if process_jobs:
-            process_results = await asyncio.gather(
-                *[
-                    self._process_image_size(url, max_size=ms)
-                    for _, _, url, _, ms, _ in process_jobs
-                ],
-                return_exceptions=True,
-            )
-            for (_, _, _, cache_key, _, cache_field), result in zip(
-                process_jobs, process_results
-            ):
-                if not isinstance(result, Exception) and result is not None:
-                    self._image_opt_cache.setdefault(cache_key, {})[cache_field] = (
-                        result
-                    )
-
-        # 第二遍：应用处理结果
-        for i in range(last_assistant_idx):
-            msg = messages[i]
-            if msg.get("role") != "tool":
-                continue
-
-            content = msg.get("content")
-            if not isinstance(content, list):
-                continue
-
-            # 计算这条消息距离最后一条 assistant 的"轮次"
-            rounds_ago = assistant_count_after[i]
-
-            # 处理每个 content block
-            new_content = []
-            for block in content:
-                if isinstance(block, dict) and block.get("type") == "image_url":
-                    image_url_obj = block.get("image_url", {})
-                    image_url = image_url_obj.get("url", "")
-
-                    # 生成缓存 key（URL 图片用 URL 本身，base64 用前 64 字符 hash）
-                    if image_url.startswith("data:"):
-                        cache_key = hashlib.md5(image_url[:200].encode()).hexdigest()
-                    else:
-                        cache_key = hashlib.md5(image_url.encode()).hexdigest()
-
-                    # 根据距离决定处理策略
-                    if rounds_ago <= 2:
-                        # 最近 1-2 轮：只补齐过小图片，保留原分辨率
-                        cached = self._image_opt_cache.get(cache_key, {})
-                        if "pad_only" in cached:
-                            new_content.append(
-                                {
-                                    "type": "image_url",
-                                    "image_url": {"url": cached["pad_only"]},
-                                }
-                            )
-                            stats["kept"] += 1
-                            stats["cache_hit"] += 1
-                        elif image_url.startswith("data:"):
-                            processed = await self._process_image_size(
-                                image_url, max_size=None
-                            )
-                            if processed:
-                                self._image_opt_cache.setdefault(cache_key, {})[
-                                    "pad_only"
-                                ] = processed
-                                new_content.append(
-                                    {
-                                        "type": "image_url",
-                                        "image_url": {"url": processed},
-                                    }
-                                )
-                            else:
-                                new_content.append(block)
-                            stats["kept"] += 1
-                        else:
-                            new_content.append(block)
-                            stats["kept"] += 1
-
-                    elif rounds_ago <= 5:
-                        # 3-5 轮：降低分辨率（优先从缓存取）
-                        cached = self._image_opt_cache.get(cache_key, {})
-                        if "downscaled" in cached:
-                            new_content.append(
-                                {
-                                    "type": "image_url",
-                                    "image_url": {"url": cached["downscaled"]},
-                                }
-                            )
-                            stats["downscaled"] += 1
-                            stats["cache_hit"] += 1
-                        elif image_url.startswith("data:"):
-                            processed = await self._process_image_size(
-                                image_url, max_size=512
-                            )
-                            if processed:
-                                # 缓存结果
-                                self._image_opt_cache.setdefault(cache_key, {})[
-                                    "downscaled"
-                                ] = processed
-                                new_content.append(
-                                    {
-                                        "type": "image_url",
-                                        "image_url": {"url": processed},
-                                    }
-                                )
-                                stats["downscaled"] += 1
-                            else:
-                                new_content.append(block)
-                                stats["kept"] += 1
-                        else:
-                            # URL 图片：无法直接处理，保留原图
-                            new_content.append(block)
-                            stats["kept"] += 1
-
-                    else:
-                        # 5 轮以上：生成文本描述（优先从缓存取）
-                        cached = self._image_opt_cache.get(cache_key, {})
-                        if "description" in cached:
-                            new_content.append(cached["description"])
-                            stats["described"] += 1
-                            stats["cache_hit"] += 1
-                        else:
-                            description = await self._generate_image_description(
-                                image_url, model
-                            )
-                            url_info = (
-                                f" (URL: {image_url[:100]}...)"
-                                if not image_url.startswith("data:")
-                                else ""
-                            )
-                            desc_block = {
-                                "type": "text",
-                                "text": f"[Image description: {description}]{url_info}",
-                            }
-                            # 缓存结果
-                            self._image_opt_cache.setdefault(cache_key, {})[
-                                "description"
-                            ] = desc_block
-                            new_content.append(desc_block)
-                            stats["described"] += 1
-                else:
-                    new_content.append(block)
-
-            msg["content"] = new_content
-        # print(f"[Image Opt Check] 扫描到 {stats['kept'] + stats['downscaled'] + stats['described']} 张图片上下文")
-        if stats["downscaled"] > 0 or stats["described"] > 0:
-            self.log.info(
-                f"[Image Optimization] 保留 {stats['kept']} 张，"
-                f"降分辨率 {stats['downscaled']} 张，"
-                f"文本描述 {stats['described']} 张，"
-                f"缓存命中 {stats['cache_hit']} 次"
-            )
-
-        return messages
+        return await self._image_runtime._optimize_images(messages, model)
 
     async def _process_image_size(
         self, base64_url: str, max_size: Optional[int] = 512, min_size: int = 11
     ) -> Optional[str]:
-        """
-        处理 base64 图片的尺寸：
-        - 若 max_size 不为 None 且大于该值，则等比例缩放
-        - 若任意一边小于 min_size，则补充白边 (Padding)
-        """
-        try:
-            from PIL import Image
-            import io
-            import base64
-
-            # 解析 base64 数据
-            if not base64_url.startswith("data:"):
-                return None
-
-            _header, data = base64_url.split(",", 1)
-
-            # 解码图片
-            img_data = base64.b64decode(data)
-            img = Image.open(io.BytesIO(img_data))
-
-            width, height = img.size
-
-            needs_downscale = max_size is not None and (
-                width > max_size or height > max_size
-            )
-            needs_pad = width < min_size or height < min_size
-
-            # 尺寸正常，无需处理
-            if not needs_downscale and not needs_pad:
-                return base64_url
-
-            new_width, new_height = width, height
-
-            # 1. 降分辨率
-            if needs_downscale:
-                if width > height:
-                    new_width = max_size
-                    new_height = int(height * max_size / width)
-                else:
-                    new_height = max_size
-                    new_width = int(width * max_size / height)
-
-            if (new_width, new_height) != (width, height):
-                img_resized = img.resize(
-                    (new_width, new_height), Image.Resampling.BILINEAR
-                )
-            else:
-                img_resized = img
-
-            # 2. 补齐白边 (Padding)
-            pad_width = max(new_width, min_size)
-            pad_height = max(new_height, min_size)
-
-            if pad_width > new_width or pad_height > new_height:
-                # 创建白色背景
-                padded_img = Image.new(
-                    "RGBA" if img_resized.mode in ("RGBA", "P") else "RGB",
-                    (pad_width, pad_height),
-                    (255, 255, 255, 255),
-                )
-                offset_x = (pad_width - new_width) // 2
-                offset_y = (pad_height - new_height) // 2
-                padded_img.paste(img_resized, (offset_x, offset_y))
-                img_resized = padded_img
-
-            # 转换为 RGB（JPEG不支持 RGBA, P 等具有透明度或索引的模式）
-            if img_resized.mode != "RGB":
-                if img_resized.mode == "RGBA" or img_resized.mode == "P":
-                    # Create a white background for transparent images
-                    background = Image.new("RGB", img_resized.size, (255, 255, 255))
-                    if img_resized.mode == "P" and "transparency" in img_resized.info:
-                        img_resized = img_resized.convert("RGBA")
-                    if img_resized.mode == "RGBA":
-                        background.paste(img_resized, mask=img_resized.split()[3])
-                        img_resized = background
-                img_resized = img_resized.convert("RGB")
-
-            # 重新编码为 JPEG（如果只是补齐没有缩放，可以稍微保留高点质量）
-            buffer = io.BytesIO()
-            quality = 60 if needs_downscale else 85
-            img_resized.save(buffer, format="JPEG", quality=quality, optimize=False)
-            new_data = base64.b64encode(buffer.getvalue()).decode("utf-8")
-
-            return f"data:image/jpeg;base64,{new_data}"
-
-        except Exception as e:
-            self.log.warning(f"[Image Process] 处理图片尺寸失败: {e}")
-            return None
+        return await self._image_runtime._process_image_size(
+            base64_url,
+            max_size=max_size,
+            min_size=min_size,
+        )
 
     async def _generate_image_description(
         self, image_url: str, current_model: str
     ) -> str:
-        """
-        使用小模型生成图片的文本描述
-
-        Args:
-            image_url: 图片 URL（base64 或 http(s)）
-            current_model: 当前使用的模型
-
-        Returns:
-            图片描述文本
-        """
-        try:
-            # 使用 qwen-vl-max（通义千问视觉模型）生成描述
-            # 注意：qwen-vl 系列专门支持视觉输入
-            description_model = "qwen-vl-max"
-
-            # 构建描述请求
-            messages = [
-                {
-                    "role": "user",
-                    "content": [
-                        {"type": "image_url", "image_url": {"url": image_url}},
-                        {
-                            "type": "text",
-                            "text": "请用 1-2 句话简洁描述这张图片的主要内容。",
-                        },
-                    ],
-                }
-            ]
-
-            # 调用 LLM
-            result = await self.llm_call(
-                messages=messages,
-                model=description_model,
-                tools=None,
-                temperature=0.3,
-            )
-
-            description = result.get("content", "").strip()
-            return description if description else "图片内容"
-
-        except Exception as e:
-            self.log.warning(f"[Image Description] 生成描述失败: {e}")
-            return "图片内容"
+        return await self._image_runtime._generate_image_description(
+            image_url,
+            current_model,
+        )
 
     def _add_cache_control(
         self, messages: List[Dict], model: str, enable: bool
     ) -> List[Dict]:
-        """
-        为支持的模型添加 Prompt Caching 标记
-
-        策略：固定位置 + 延迟缓存
-        1. 如果有未处理的图片（最后一条 assistant 之后的 tool messages 中有图片），跳过缓存
-        2. system message 添加缓存（如果足够长）
-        3. 固定位置缓存点（20, 40, 60, 80），确保每个缓存点间隔 >= 1024 tokens
-        4. 最多使用 4 个缓存点（含 system）
-
-        Args:
-            messages: 原始消息列表
-            model: 模型名称
-            enable: 是否启用缓存
-
-        Returns:
-            添加了 cache_control 的消息列表（深拷贝）
-        """
-        if not enable:
-            return messages
-
-        # 只对 Claude 模型启用
-        if "claude" not in model.lower():
-            return messages
-
-        # 延迟缓存：检查是否有未处理的图片
-        last_assistant_idx = -1
-        for i in range(len(messages) - 1, -1, -1):
-            if messages[i].get("role") == "assistant":
-                last_assistant_idx = i
-                break
-
-        # 检查最后一条 assistant 之后是否有包含图片的 tool messages
-        has_unprocessed_images = False
-        if last_assistant_idx >= 0:
-            for i in range(last_assistant_idx + 1, len(messages)):
-                msg = messages[i]
-                if msg.get("role") == "tool":
-                    content = msg.get("content")
-                    if isinstance(content, list):
-                        has_unprocessed_images = any(
-                            isinstance(block, dict) and block.get("type") == "image_url"
-                            for block in content
-                        )
-                        if has_unprocessed_images:
-                            break
-
-        if has_unprocessed_images:
-            self.log.debug("[Cache] 检测到未处理的图片，延迟缓存建立")
-            return messages
-
-        # 深拷贝避免修改原始数据
-        import copy
-
-        messages = copy.deepcopy(messages)
-
-        # 策略 1: 为 system message 添加缓存
-        system_cached = False
-        for msg in messages:
-            if msg.get("role") == "system":
-                content = msg.get("content", "")
-                if isinstance(content, str) and len(content) > 1000:
-                    msg["content"] = [
-                        {
-                            "type": "text",
-                            "text": content,
-                            "cache_control": {"type": "ephemeral"},
-                        }
-                    ]
-                    system_cached = True
-                    self.log.debug(
-                        f"[Cache] 为 system message 添加缓存标记 (len={len(content)})"
-                    )
-                break
-
-        # 策略 2: 固定位置缓存点
-        CACHE_INTERVAL = 20
-        MAX_POINTS = 3 if system_cached else 4
-        MIN_TOKENS = 1024
-        AVG_TOKENS_PER_MSG = 70
-
-        total_msgs = len(messages)
-        if total_msgs == 0:
-            return messages
-
-        cache_positions = []
-        last_cache_pos = 0
-
-        for i in range(1, MAX_POINTS + 1):
-            target_pos = i * CACHE_INTERVAL - 1  # 19, 39, 59, 79
-
-            if target_pos >= total_msgs:
-                break
-
-            # 从目标位置开始查找合适的 user/assistant 消息
-            for j in range(target_pos, total_msgs):
-                msg = messages[j]
-
-                if msg.get("role") not in ("user", "assistant"):
-                    continue
-
-                content = msg.get("content", "")
-                if not content:
-                    continue
-
-                # 检查 content 是否非空
-                is_valid = False
-                if isinstance(content, str):
-                    is_valid = len(content) > 0
-                elif isinstance(content, list):
-                    is_valid = any(
-                        isinstance(block, dict)
-                        and block.get("type") == "text"
-                        and len(block.get("text", "")) > 0
-                        for block in content
-                    )
-
-                if not is_valid:
-                    continue
-
-                # 检查 token 距离
-                msg_count = j - last_cache_pos
-                estimated_tokens = msg_count * AVG_TOKENS_PER_MSG
-
-                if estimated_tokens >= MIN_TOKENS:
-                    cache_positions.append(j)
-                    last_cache_pos = j
-                    self.log.debug(
-                        f"[Cache] 在位置 {j} 添加缓存点 (估算 {estimated_tokens} tokens)"
-                    )
-                    break
-
-        # 应用缓存标记
-        for idx in cache_positions:
-            msg = messages[idx]
-            content = msg.get("content", "")
-
-            if isinstance(content, str):
-                msg["content"] = [
-                    {
-                        "type": "text",
-                        "text": content,
-                        "cache_control": {"type": "ephemeral"},
-                    }
-                ]
-                self.log.debug(
-                    f"[Cache] 为 message[{idx}] ({msg.get('role')}) 添加缓存标记"
-                )
-            elif isinstance(content, list):
-                # 在最后一个 text block 添加 cache_control
-                for block in reversed(content):
-                    if isinstance(block, dict) and block.get("type") == "text":
-                        block["cache_control"] = {"type": "ephemeral"}
-                        self.log.debug(
-                            f"[Cache] 为 message[{idx}] ({msg.get('role')}) 添加缓存标记"
-                        )
-                        break
-
-        self.log.debug(
-            f"[Cache] 总消息: {total_msgs}, "
-            f"缓存点: {len(cache_positions)} at {cache_positions}"
-        )
-        return messages
+        return self._image_runtime._add_cache_control(messages, model, enable)
 
     def _resolve_run_policy(
         self,
         config: RunConfig,
         trace: Optional[Trace] = None,
     ) -> ResolvedAgentPolicy:
-        """Resolve role and effective tools from the preset, never caller context."""
-        preset = get_preset(config.agent_type)
-        policy = self.tool_policy.resolve(config, preset, self.tools)
-        if trace and trace.agent_role != policy.role.value:
-            raise ValueError(
-                f"Trace role mismatch: stored={trace.agent_role}, requested={policy.role.value}"
-            )
-        if policy.completion_policy == CompletionPolicy.EXPLICIT_VALIDATION:
-            if policy.role == AgentRole.LEGACY:
-                raise ValueError(
-                    f"Preset '{config.agent_type}' has legacy role and cannot run explicit_validation"
-                )
-        return policy
+        return self._tool_runtime.resolve_policy(config, trace)
 
     def _get_run_tool_schemas(
         self,
         config: RunConfig,
         policy: Optional[ResolvedAgentPolicy] = None,
     ) -> List[Dict]:
-        policy = policy or self._resolve_run_policy(config)
-        if policy.completion_policy == CompletionPolicy.LEGACY_AUTO:
-            schemas = self._get_tool_schemas(
-                config.tools, config.tool_groups, config.exclude_tools
-            )
-            explicit_only = PLANNER_TOOLS | WORKER_TOOLS | VALIDATOR_TOOLS
-            return [
-                schema
-                for schema in schemas
-                if schema.get("function", {}).get("name") not in explicit_only
-            ]
-        return self.tools.get_schemas(sorted(policy.effective_tools))
+        return self._tool_runtime.get_run_tool_schemas(
+            config,
+            policy or self._resolve_run_policy(config),
+            get_tool_schemas=self._get_tool_schemas,
+        )
 
     def _build_protected_tool_context(
         self,
@@ -3832,31 +3226,16 @@ class AgentRunner:
         sequence: int,
         side_branch: Optional[Dict[str, Any]] = None,
     ) -> Dict[str, Any]:
-        """Merge host context first; framework identity fields always win."""
-        context = dict(config.context or {})
-        trace_context = trace.context or {}
-        context.update(
-            {
-                "store": self.trace_store,
-                "trace_id": trace.trace_id,
-                "root_trace_id": trace_context.get("root_trace_id", trace.trace_id),
-                "goal_id": goal_tree.current_id if goal_tree else None,
-                "runner": self,
-                "goal_tree": goal_tree,
-                "knowledge_config": config.knowledge,
-                "sequence": sequence,
-                "side_branch": side_branch,
-                "role": trace.agent_role,
-                "completion_policy": CompletionPolicy(config.completion_policy).value,
-                "task_id": trace_context.get("task_id"),
-                "spec_version": trace_context.get("spec_version"),
-                "attempt_id": trace_context.get("attempt_id"),
-                "snapshot_id": trace_context.get("snapshot_id"),
-                "validation_id": trace_context.get("validation_id"),
-                "coordinator": self.task_coordinator,
-            }
+        return self._tool_runtime.build_protected_context(
+            config,
+            trace,
+            goal_tree,
+            sequence,
+            runner=self,
+            trace_store=self.trace_store,
+            task_coordinator=self.task_coordinator,
+            side_branch=side_branch,
         )
-        return context
 
     async def _execute_authorized_tool(
         self,
@@ -3869,26 +3248,17 @@ class AgentRunner:
         sequence: int,
         side_branch: Optional[Dict[str, Any]] = None,
     ) -> Any:
-        policy = self._resolve_run_policy(config, trace)
-        if policy.completion_policy == CompletionPolicy.EXPLICIT_VALIDATION:
-            authorization = self.tool_policy.authorize(policy.role, tool_name, policy)
-            if not authorization.allowed:
-                self.log.warning(
-                    "Rejected unauthorized tool call: %s", authorization.reason
-                )
-                return {
-                    "text": f"Error: {authorization.reason}",
-                    "error": "unauthorized_tool",
-                }
-        context = self._build_protected_tool_context(
-            config, trace, goal_tree, sequence, side_branch=side_branch
-        )
-        context["tool_call_id"] = tool_call_id
-        return await self.tools.execute(
+        return await self._tool_runtime.execute_authorized_tool(
             tool_name,
             tool_args,
-            uid=config.uid or "",
-            context=context,
+            tool_call_id,
+            config,
+            trace,
+            goal_tree,
+            sequence,
+            resolve_policy=self._resolve_run_policy,
+            build_context=self._build_protected_tool_context,
+            side_branch=side_branch,
         )
 
     def _get_tool_schemas(
@@ -3897,28 +3267,11 @@ class AgentRunner:
         tool_groups: Optional[List[str]] = None,
         exclude_tools: Optional[List[str]] = None,
     ) -> List[Dict]:
-        """
-        获取工具 Schema
-
-        合并策略（取并集）：
-        - tool_groups 非空: 按分组白名单过滤得到基础工具集
-        - tools 非空: 追加指定的工具名（与 tool_groups 结果取并集）
-        - 两者都为 None: 返回所有已注册工具
-
-        最后再用 exclude_tools 减去禁用的工具（如远程 agent 禁止 agent/evaluate）。
-        """
-        if tool_groups is not None:
-            tool_names = set(self.tools.get_tool_names(groups=tool_groups))
-        else:
-            tool_names = set(self.tools.get_tool_names())
-        if tools is not None:
-            tool_names |= set(tools)
-        if exclude_tools:
-            tool_names -= set(exclude_tools)
-        return self.tools.get_schemas(list(tool_names))
-
-    # 默认 system prompt 前缀（当 config.system_prompt 和前端都未提供 system message 时使用）
-    # 注意：此常量已迁移到 agent.core.prompts，这里保留引用以保持向后兼容
+        return self._tool_runtime.get_tool_schemas(
+            tools,
+            tool_groups,
+            exclude_tools,
+        )
 
     async def _build_system_prompt(
         self, config: RunConfig, base_prompt: Optional[str] = None
@@ -3940,108 +3293,20 @@ class AgentRunner:
             base_prompt: 已有 system 内容（来自消息），
                          None 时使用 config.system_prompt 或 preset.system_prompt
         """
-        from agent.core.presets import AGENT_PRESETS
-
-        # 确定 system_prompt 来源
-        if base_prompt is not None:
-            system_prompt = base_prompt
-        elif config.system_prompt is not None:
-            system_prompt = config.system_prompt
-        else:
-            # 尝试从 preset 获取 system_prompt
-            preset = AGENT_PRESETS.get(config.agent_type)
-            system_prompt = (
-                preset.system_prompt if preset and preset.system_prompt else None
-            )
-
-        # 确定要加载哪些 skills
-        skills_filter: Optional[List[str]] = config.skills
-        if skills_filter is None:
-            preset = AGENT_PRESETS.get(config.agent_type)
-            if preset is not None:
-                skills_filter = preset.skills  # 可能仍为 None（加载全部）
-
-        # 加载并过滤
-        all_skills = load_skills_from_dir(self.skills_dir)
-        if skills_filter is not None:
-            skills = [s for s in all_skills if s.name in skills_filter]
-        else:
-            skills = all_skills
-
-        skills_text = self._format_skills(skills) if skills else ""
-
-        if system_prompt:
-            if skills_text:
-                system_prompt += f"\n\n## Skills\n{skills_text}"
-        else:
-            system_prompt = DEFAULT_SYSTEM_PREFIX
-            if skills_text:
-                system_prompt += f"\n\n## Skills\n{skills_text}"
-
-        if config.max_iterations and config.max_iterations > 0:
-            system_prompt += f"\n\n## Execution Constraint\n这是一项有严格步数限制的任务。你最多可以用 {config.max_iterations} 轮交互来解决问题。\n请务必【边查边写、随时存档】！每当你收集或得出一个有价值的独立结果（如收集到一个独立 Case），请立刻调用工具写入或追加到结果文件中，绝对不要等到所有任务都做完再最后一次性输出。这样即使触达步数上限被强制打断，你已经收集的成果也能安全保留！"
-        # Memory 注入（memory-bearing Agent）——在 system prompt 末尾追加
-        # 初版选择 system prompt 追加（见 agent/docs/memory.md 待定问题 1）。
-        # 好处：run 启动一次性注入、所有后续轮次都能看到、与 skills 注入方式一致。
-        # 代价：若记忆文件很大会持续占 prompt tokens —— 待观察后决定是否切换方案。
-        if config.memory:
-            try:
-                from agent.core.memory import load_memory_files, format_memory_injection
-
-                files = load_memory_files(config.memory)
-                memory_text = format_memory_injection(files)
-                if memory_text:
-                    system_prompt += f"\n\n{memory_text}"
-            except Exception as e:
-                self.log.warning(f"[Memory] 加载记忆失败，跳过注入: {e}")
-
-        # 框架角色契约最后追加，Host prompt、skills 和 memory 都不能覆盖不变量。
-        policy = self._resolve_run_policy(config)
-        if policy.completion_policy == CompletionPolicy.EXPLICIT_VALIDATION:
-            contract = ROLE_CONTRACTS.get(policy.role.value)
-            if contract:
-                system_prompt += f"\n\n{contract}"
-
-        return system_prompt
+        return await self._prompt_runtime.build_system_prompt(
+            config,
+            base_prompt,
+            skills_dir=self.skills_dir,
+            resolve_policy=self._resolve_run_policy,
+            format_skills=self._format_skills,
+        )
 
     async def _generate_task_name(self, messages: List[Dict]) -> str:
         """生成任务名称：优先使用 utility_llm，fallback 到文本截取"""
-        # 提取 messages 中的文本内容
-        text_parts = []
-        for msg in messages:
-            content = msg.get("content", "")
-            if isinstance(content, str):
-                text_parts.append(content)
-            elif isinstance(content, list):
-                for part in content:
-                    if isinstance(part, dict) and part.get("type") == "text":
-                        text_parts.append(part.get("text", ""))
-        raw_text = " ".join(text_parts).strip()
-
-        if not raw_text:
-            return TASK_NAME_FALLBACK
-
-        # 尝试使用 utility_llm 生成标题
-        if self.utility_llm_call:
-            try:
-                result = await self.utility_llm_call(
-                    messages=[
-                        {
-                            "role": "system",
-                            "content": TASK_NAME_GENERATION_SYSTEM_PROMPT,
-                        },
-                        {"role": "user", "content": raw_text[:2000]},
-                    ],
-                    model="gpt-4o-mini",  # 使用便宜模型
-                )
-                title = result.get("content", "").strip()
-                if title and len(title) < 100:
-                    return title
-            except Exception:
-                pass
-
-        # Fallback: 截取前 50 字符
-        return raw_text[:50] + ("..." if len(raw_text) > 50 else "")
+        return await self._prompt_runtime.generate_task_name(
+            messages,
+            utility_llm_call=self.utility_llm_call,
+        )
 
     def _format_skills(self, skills: List[Skill]) -> str:
         if not skills:

agent/agent/core/runner_images.py

@@ -0,0 +1,636 @@
+"""Image optimization and prompt-cache support for AgentRunner."""
+
+from __future__ import annotations
+
+from typing import Any, Dict, List, Optional
+
+
+class RunnerImageRuntime:
+    """Own image processing while preserving AgentRunner compatibility hooks."""
+
+    def __init__(self, owner: Any) -> None:
+        self._owner = owner
+
+    @property
+    def _image_opt_cache(self) -> Dict[str, Dict[str, Any]]:
+        return self._owner._image_opt_cache
+
+    @property
+    def log(self):
+        return self._owner.log
+
+    @property
+    def llm_call(self):
+        return self._owner.llm_call
+
+    async def _optimize_images(self, messages: List[Dict], model: str) -> List[Dict]:
+        """
+        分级优化已处理的图片，节省 token
+
+        策略（基于图片距离最后一条 assistant 的"轮次"）：
+        1. 最近 1-2 轮：保留原图
+        2. 3-5 轮：降低分辨率和压缩（节省 token 但保留视觉信息）
+        3. 5 轮以上：调用小模型生成文本描述 + 保留 URL
+
+        处理结果会缓存，避免重复的 PIL 解码/编码和 LLM 调用。
+
+        Args:
+            messages: 原始消息列表
+            model: 当前使用的模型（用于选择描述生成模型）
+
+        Returns:
+            优化后的消息列表（深拷贝）
+        """
+        if not messages:
+            return messages
+
+        # 找到最后一条 assistant message 的位置
+        last_assistant_idx = -1
+        for i in range(len(messages) - 1, -1, -1):
+            if messages[i].get("role") == "assistant":
+                last_assistant_idx = i
+                break
+
+        # 如果没有 assistant message，说明还没开始对话，不优化
+        if last_assistant_idx == -1:
+            return messages
+
+        # 统计从每个位置到最后一条 assistant 之间的 assistant 数量（作为"轮次"）
+        assistant_count_after = [0] * len(messages)
+        count = 0
+        for i in range(len(messages) - 1, -1, -1):
+            assistant_count_after[i] = count
+            if messages[i].get("role") == "assistant":
+                count += 1
+
+        # 深拷贝避免修改原始数据
+        import copy
+        import hashlib
+        import asyncio
+        import base64 as b64mod
+        import httpx
+        import mimetypes
+
+        messages = copy.deepcopy(messages)
+
+        # 预处理：将所有 HTTP(S) URL 图片下载并转为 base64 data URL
+        # Qwen API 无法访问外部签名 URL（如 BFL、火山引擎 TOS），必须在本地转换
+        url_download_jobs = []  # [(msg_idx, block_idx, url)]
+        for i, msg in enumerate(messages):
+            if msg.get("role") != "tool":
+                continue
+            content = msg.get("content")
+            if not isinstance(content, list):
+                continue
+            for block_idx, block in enumerate(content):
+                if isinstance(block, dict) and block.get("type") == "image_url":
+                    url = block.get("image_url", {}).get("url", "")
+                    if url.startswith(("http://", "https://")):
+                        url_download_jobs.append((i, block_idx, url))
+
+        if url_download_jobs:
+
+            async def _download_image_to_data_url(url: str) -> str | None:
+                try:
+                    async with httpx.AsyncClient(timeout=60, trust_env=False) as client:
+                        resp = await client.get(url)
+                        resp.raise_for_status()
+                        ct = resp.headers.get("content-type", "").split(";")[0].strip()
+                        if not ct.startswith("image/"):
+                            ct = (
+                                mimetypes.guess_type(url.split("?")[0])[0]
+                                or "image/png"
+                            )
+                        b64 = b64mod.b64encode(resp.content).decode()
+                        return f"data:{ct};base64,{b64}"
+                except Exception:
+                    self.log.debug(
+                        "Remote image normalization failed: %s", url, exc_info=True
+                    )
+                    return None
+
+            results = await asyncio.gather(
+                *[_download_image_to_data_url(url) for _, _, url in url_download_jobs],
+                return_exceptions=True,
+            )
+            converted = 0
+            for (msg_idx, block_idx, original_url), result in zip(
+                url_download_jobs, results
+            ):
+                if isinstance(result, str) and result.startswith("data:"):
+                    messages[msg_idx]["content"][block_idx]["image_url"]["url"] = result
+                    converted += 1
+            if converted:
+                self.log.info(
+                    f"[Image Optimization] URL→base64 预转换: {converted}/{len(url_download_jobs)} 张"
+                )
+
+        # 统计优化情况
+        stats = {"kept": 0, "downscaled": 0, "described": 0, "cache_hit": 0}
+
+        # 收集需要降分辨率或尺寸补齐的图片（用于并发处理）
+        process_jobs = []  # [(msg_idx, block_idx, image_url, cache_key, max_size, cache_field)]
+
+        # 第一遍：扫描并收集需要处理的图片
+        for i in range(last_assistant_idx):
+            msg = messages[i]
+            if msg.get("role") != "tool":
+                continue
+
+            content = msg.get("content")
+            if not isinstance(content, list):
+                continue
+
+            rounds_ago = assistant_count_after[i]
+
+            for block_idx, block in enumerate(content):
+                if isinstance(block, dict) and block.get("type") == "image_url":
+                    image_url_obj = block.get("image_url", {})
+                    image_url = image_url_obj.get("url", "")
+
+                    if image_url.startswith("data:"):
+                        cache_key = hashlib.md5(image_url[:200].encode()).hexdigest()
+                    else:
+                        cache_key = hashlib.md5(image_url.encode()).hexdigest()
+
+                    # 1-5 轮都需要检查尺寸
+                    if rounds_ago <= 5:
+                        cached = self._image_opt_cache.get(cache_key, {})
+                        cache_field = "pad_only" if rounds_ago <= 2 else "downscaled"
+
+                        if cache_field not in cached and image_url.startswith("data:"):
+                            max_size = None if rounds_ago <= 2 else 512
+                            process_jobs.append(
+                                (
+                                    i,
+                                    block_idx,
+                                    image_url,
+                                    cache_key,
+                                    max_size,
+                                    cache_field,
+                                )
+                            )
+
+        # 并发处理所有尺寸任务
+        if process_jobs:
+            process_results = await asyncio.gather(
+                *[
+                    self._owner._process_image_size(url, max_size=ms)
+                    for _, _, url, _, ms, _ in process_jobs
+                ],
+                return_exceptions=True,
+            )
+            for (_, _, _, cache_key, _, cache_field), result in zip(
+                process_jobs, process_results
+            ):
+                if not isinstance(result, Exception) and result is not None:
+                    self._image_opt_cache.setdefault(cache_key, {})[cache_field] = (
+                        result
+                    )
+
+        # 第二遍：应用处理结果
+        for i in range(last_assistant_idx):
+            msg = messages[i]
+            if msg.get("role") != "tool":
+                continue
+
+            content = msg.get("content")
+            if not isinstance(content, list):
+                continue
+
+            # 计算这条消息距离最后一条 assistant 的"轮次"
+            rounds_ago = assistant_count_after[i]
+
+            # 处理每个 content block
+            new_content = []
+            for block in content:
+                if isinstance(block, dict) and block.get("type") == "image_url":
+                    image_url_obj = block.get("image_url", {})
+                    image_url = image_url_obj.get("url", "")
+
+                    # 生成缓存 key（URL 图片用 URL 本身，base64 用前 64 字符 hash）
+                    if image_url.startswith("data:"):
+                        cache_key = hashlib.md5(image_url[:200].encode()).hexdigest()
+                    else:
+                        cache_key = hashlib.md5(image_url.encode()).hexdigest()
+
+                    # 根据距离决定处理策略
+                    if rounds_ago <= 2:
+                        # 最近 1-2 轮：只补齐过小图片，保留原分辨率
+                        cached = self._image_opt_cache.get(cache_key, {})
+                        if "pad_only" in cached:
+                            new_content.append(
+                                {
+                                    "type": "image_url",
+                                    "image_url": {"url": cached["pad_only"]},
+                                }
+                            )
+                            stats["kept"] += 1
+                            stats["cache_hit"] += 1
+                        elif image_url.startswith("data:"):
+                            processed = await self._owner._process_image_size(
+                                image_url, max_size=None
+                            )
+                            if processed:
+                                self._image_opt_cache.setdefault(cache_key, {})[
+                                    "pad_only"
+                                ] = processed
+                                new_content.append(
+                                    {
+                                        "type": "image_url",
+                                        "image_url": {"url": processed},
+                                    }
+                                )
+                            else:
+                                new_content.append(block)
+                            stats["kept"] += 1
+                        else:
+                            new_content.append(block)
+                            stats["kept"] += 1
+
+                    elif rounds_ago <= 5:
+                        # 3-5 轮：降低分辨率（优先从缓存取）
+                        cached = self._image_opt_cache.get(cache_key, {})
+                        if "downscaled" in cached:
+                            new_content.append(
+                                {
+                                    "type": "image_url",
+                                    "image_url": {"url": cached["downscaled"]},
+                                }
+                            )
+                            stats["downscaled"] += 1
+                            stats["cache_hit"] += 1
+                        elif image_url.startswith("data:"):
+                            processed = await self._owner._process_image_size(
+                                image_url, max_size=512
+                            )
+                            if processed:
+                                # 缓存结果
+                                self._image_opt_cache.setdefault(cache_key, {})[
+                                    "downscaled"
+                                ] = processed
+                                new_content.append(
+                                    {
+                                        "type": "image_url",
+                                        "image_url": {"url": processed},
+                                    }
+                                )
+                                stats["downscaled"] += 1
+                            else:
+                                new_content.append(block)
+                                stats["kept"] += 1
+                        else:
+                            # URL 图片：无法直接处理，保留原图
+                            new_content.append(block)
+                            stats["kept"] += 1
+
+                    else:
+                        # 5 轮以上：生成文本描述（优先从缓存取）
+                        cached = self._image_opt_cache.get(cache_key, {})
+                        if "description" in cached:
+                            new_content.append(cached["description"])
+                            stats["described"] += 1
+                            stats["cache_hit"] += 1
+                        else:
+                            description = await self._owner._generate_image_description(
+                                image_url, model
+                            )
+                            url_info = (
+                                f" (URL: {image_url[:100]}...)"
+                                if not image_url.startswith("data:")
+                                else ""
+                            )
+                            desc_block = {
+                                "type": "text",
+                                "text": f"[Image description: {description}]{url_info}",
+                            }
+                            # 缓存结果
+                            self._image_opt_cache.setdefault(cache_key, {})[
+                                "description"
+                            ] = desc_block
+                            new_content.append(desc_block)
+                            stats["described"] += 1
+                else:
+                    new_content.append(block)
+
+            msg["content"] = new_content
+        if stats["downscaled"] > 0 or stats["described"] > 0:
+            self.log.info(
+                f"[Image Optimization] 保留 {stats['kept']} 张，"
+                f"降分辨率 {stats['downscaled']} 张，"
+                f"文本描述 {stats['described']} 张，"
+                f"缓存命中 {stats['cache_hit']} 次"
+            )
+
+        return messages
+
+    async def _process_image_size(
+        self, base64_url: str, max_size: Optional[int] = 512, min_size: int = 11
+    ) -> Optional[str]:
+        """
+        处理 base64 图片的尺寸：
+        - 若 max_size 不为 None 且大于该值，则等比例缩放
+        - 若任意一边小于 min_size，则补充白边 (Padding)
+        """
+        try:
+            from PIL import Image
+            import io
+            import base64
+
+            # 解析 base64 数据
+            if not base64_url.startswith("data:"):
+                return None
+
+            _header, data = base64_url.split(",", 1)
+
+            # 解码图片
+            img_data = base64.b64decode(data)
+            img = Image.open(io.BytesIO(img_data))
+
+            width, height = img.size
+
+            needs_downscale = max_size is not None and (
+                width > max_size or height > max_size
+            )
+            needs_pad = width < min_size or height < min_size
+
+            # 尺寸正常，无需处理
+            if not needs_downscale and not needs_pad:
+                return base64_url
+
+            new_width, new_height = width, height
+
+            # 1. 降分辨率
+            if needs_downscale:
+                if width > height:
+                    new_width = max_size
+                    new_height = int(height * max_size / width)
+                else:
+                    new_height = max_size
+                    new_width = int(width * max_size / height)
+
+            if (new_width, new_height) != (width, height):
+                img_resized = img.resize(
+                    (new_width, new_height), Image.Resampling.BILINEAR
+                )
+            else:
+                img_resized = img
+
+            # 2. 补齐白边 (Padding)
+            pad_width = max(new_width, min_size)
+            pad_height = max(new_height, min_size)
+
+            if pad_width > new_width or pad_height > new_height:
+                # 创建白色背景
+                padded_img = Image.new(
+                    "RGBA" if img_resized.mode in ("RGBA", "P") else "RGB",
+                    (pad_width, pad_height),
+                    (255, 255, 255, 255),
+                )
+                offset_x = (pad_width - new_width) // 2
+                offset_y = (pad_height - new_height) // 2
+                padded_img.paste(img_resized, (offset_x, offset_y))
+                img_resized = padded_img
+
+            # 转换为 RGB（JPEG不支持 RGBA, P 等具有透明度或索引的模式）
+            if img_resized.mode != "RGB":
+                if img_resized.mode == "RGBA" or img_resized.mode == "P":
+                    # Create a white background for transparent images
+                    background = Image.new("RGB", img_resized.size, (255, 255, 255))
+                    if img_resized.mode == "P" and "transparency" in img_resized.info:
+                        img_resized = img_resized.convert("RGBA")
+                    if img_resized.mode == "RGBA":
+                        background.paste(img_resized, mask=img_resized.split()[3])
+                        img_resized = background
+                img_resized = img_resized.convert("RGB")
+
+            # 重新编码为 JPEG（如果只是补齐没有缩放，可以稍微保留高点质量）
+            buffer = io.BytesIO()
+            quality = 60 if needs_downscale else 85
+            img_resized.save(buffer, format="JPEG", quality=quality, optimize=False)
+            new_data = base64.b64encode(buffer.getvalue()).decode("utf-8")
+
+            return f"data:image/jpeg;base64,{new_data}"
+
+        except Exception as e:
+            self.log.warning(f"[Image Process] 处理图片尺寸失败: {e}")
+            return None
+
+    async def _generate_image_description(
+        self, image_url: str, current_model: str
+    ) -> str:
+        """
+        使用小模型生成图片的文本描述
+
+        Args:
+            image_url: 图片 URL（base64 或 http(s)）
+            current_model: 当前使用的模型
+
+        Returns:
+            图片描述文本
+        """
+        # The parameter remains part of AgentRunner's compatibility hook; the
+        # description itself always uses the dedicated vision model below.
+        del current_model
+        try:
+            # 使用 qwen-vl-max（通义千问视觉模型）生成描述
+            # 注意：qwen-vl 系列专门支持视觉输入
+            description_model = "qwen-vl-max"
+
+            # 构建描述请求
+            messages = [
+                {
+                    "role": "user",
+                    "content": [
+                        {"type": "image_url", "image_url": {"url": image_url}},
+                        {
+                            "type": "text",
+                            "text": "请用 1-2 句话简洁描述这张图片的主要内容。",
+                        },
+                    ],
+                }
+            ]
+
+            # 调用 LLM
+            result = await self.llm_call(
+                messages=messages,
+                model=description_model,
+                tools=None,
+                temperature=0.3,
+            )
+
+            description = result.get("content", "").strip()
+            return description if description else "图片内容"
+
+        except Exception as e:
+            self.log.warning(f"[Image Description] 生成描述失败: {e}")
+            return "图片内容"
+
+    def _add_cache_control(
+        self, messages: List[Dict], model: str, enable: bool
+    ) -> List[Dict]:
+        """
+        为支持的模型添加 Prompt Caching 标记
+
+        策略：固定位置 + 延迟缓存
+        1. 如果有未处理的图片（最后一条 assistant 之后的 tool messages 中有图片），跳过缓存
+        2. system message 添加缓存（如果足够长）
+        3. 固定位置缓存点（20, 40, 60, 80），确保每个缓存点间隔 >= 1024 tokens
+        4. 最多使用 4 个缓存点（含 system）
+
+        Args:
+            messages: 原始消息列表
+            model: 模型名称
+            enable: 是否启用缓存
+
+        Returns:
+            添加了 cache_control 的消息列表（深拷贝）
+        """
+        if not enable:
+            return messages
+
+        # 只对 Claude 模型启用
+        if "claude" not in model.lower():
+            return messages
+
+        # 延迟缓存：检查是否有未处理的图片
+        last_assistant_idx = -1
+        for i in range(len(messages) - 1, -1, -1):
+            if messages[i].get("role") == "assistant":
+                last_assistant_idx = i
+                break
+
+        # 检查最后一条 assistant 之后是否有包含图片的 tool messages
+        has_unprocessed_images = False
+        if last_assistant_idx >= 0:
+            for i in range(last_assistant_idx + 1, len(messages)):
+                msg = messages[i]
+                if msg.get("role") == "tool":
+                    content = msg.get("content")
+                    if isinstance(content, list):
+                        has_unprocessed_images = any(
+                            isinstance(block, dict) and block.get("type") == "image_url"
+                            for block in content
+                        )
+                        if has_unprocessed_images:
+                            break
+
+        if has_unprocessed_images:
+            self.log.debug("[Cache] 检测到未处理的图片，延迟缓存建立")
+            return messages
+
+        # 深拷贝避免修改原始数据
+        import copy
+
+        messages = copy.deepcopy(messages)
+
+        # 策略 1: 为 system message 添加缓存
+        system_cached = False
+        for msg in messages:
+            if msg.get("role") == "system":
+                content = msg.get("content", "")
+                if isinstance(content, str) and len(content) > 1000:
+                    msg["content"] = [
+                        {
+                            "type": "text",
+                            "text": content,
+                            "cache_control": {"type": "ephemeral"},
+                        }
+                    ]
+                    system_cached = True
+                    self.log.debug(
+                        f"[Cache] 为 system message 添加缓存标记 (len={len(content)})"
+                    )
+                break
+
+        # 策略 2: 固定位置缓存点
+        CACHE_INTERVAL = 20
+        MAX_POINTS = 3 if system_cached else 4
+        MIN_TOKENS = 1024
+        AVG_TOKENS_PER_MSG = 70
+
+        total_msgs = len(messages)
+        if total_msgs == 0:
+            return messages
+
+        cache_positions = []
+        last_cache_pos = 0
+
+        for i in range(1, MAX_POINTS + 1):
+            target_pos = i * CACHE_INTERVAL - 1  # 19, 39, 59, 79
+
+            if target_pos >= total_msgs:
+                break
+
+            # 从目标位置开始查找合适的 user/assistant 消息
+            for j in range(target_pos, total_msgs):
+                msg = messages[j]
+
+                if msg.get("role") not in ("user", "assistant"):
+                    continue
+
+                content = msg.get("content", "")
+                if not content:
+                    continue
+
+                # 检查 content 是否非空
+                is_valid = False
+                if isinstance(content, str):
+                    is_valid = len(content) > 0
+                elif isinstance(content, list):
+                    is_valid = any(
+                        isinstance(block, dict)
+                        and block.get("type") == "text"
+                        and len(block.get("text", "")) > 0
+                        for block in content
+                    )
+
+                if not is_valid:
+                    continue
+
+                # 检查 token 距离
+                msg_count = j - last_cache_pos
+                estimated_tokens = msg_count * AVG_TOKENS_PER_MSG
+
+                if estimated_tokens >= MIN_TOKENS:
+                    cache_positions.append(j)
+                    last_cache_pos = j
+                    self.log.debug(
+                        f"[Cache] 在位置 {j} 添加缓存点 (估算 {estimated_tokens} tokens)"
+                    )
+                    break
+
+        # 应用缓存标记
+        for idx in cache_positions:
+            msg = messages[idx]
+            content = msg.get("content", "")
+
+            if isinstance(content, str):
+                msg["content"] = [
+                    {
+                        "type": "text",
+                        "text": content,
+                        "cache_control": {"type": "ephemeral"},
+                    }
+                ]
+                self.log.debug(
+                    f"[Cache] 为 message[{idx}] ({msg.get('role')}) 添加缓存标记"
+                )
+            elif isinstance(content, list):
+                # 在最后一个 text block 添加 cache_control
+                for block in reversed(content):
+                    if isinstance(block, dict) and block.get("type") == "text":
+                        block["cache_control"] = {"type": "ephemeral"}
+                        self.log.debug(
+                            f"[Cache] 为 message[{idx}] ({msg.get('role')}) 添加缓存标记"
+                        )
+                        break
+
+        self.log.debug(
+            f"[Cache] 总消息: {total_msgs}, "
+            f"缓存点: {len(cache_positions)} at {cache_positions}"
+        )
+        return messages
+
+
+__all__ = ["RunnerImageRuntime"]

agent/agent/core/runner_prompts.py

@@ -0,0 +1,140 @@
+"""Prompt assembly and task naming for ``AgentRunner``."""
+
+from __future__ import annotations
+
+import logging
+from typing import Any, Awaitable, Callable, Dict, List, Optional
+
+from agent.core.presets import AGENT_PRESETS
+from agent.core.prompts import (
+    DEFAULT_SYSTEM_PREFIX,
+    ROLE_CONTRACTS,
+    TASK_NAME_FALLBACK,
+    TASK_NAME_GENERATION_SYSTEM_PROMPT,
+)
+from agent.orchestration.models import CompletionPolicy
+from agent.skill.models import Skill
+from agent.skill.skill_loader import load_skills_from_dir
+
+
+class RunnerPromptRuntime:
+    """Build prompts while leaving Agent loop state in ``AgentRunner``."""
+
+    def __init__(
+        self,
+        *,
+        logger: logging.Logger,
+    ) -> None:
+        self._log = logger
+
+    async def build_system_prompt(
+        self,
+        config: Any,
+        base_prompt: Optional[str],
+        *,
+        skills_dir: Optional[str],
+        resolve_policy: Callable[[Any], Any],
+        format_skills: Callable[[List[Skill]], str],
+    ) -> Optional[str]:
+        if base_prompt is not None:
+            system_prompt = base_prompt
+        elif config.system_prompt is not None:
+            system_prompt = config.system_prompt
+        else:
+            preset = AGENT_PRESETS.get(config.agent_type)
+            system_prompt = (
+                preset.system_prompt if preset and preset.system_prompt else None
+            )
+
+        skills_filter: Optional[List[str]] = config.skills
+        if skills_filter is None:
+            preset = AGENT_PRESETS.get(config.agent_type)
+            if preset is not None:
+                skills_filter = preset.skills
+
+        all_skills = load_skills_from_dir(skills_dir)
+        if skills_filter is not None:
+            skills = [skill for skill in all_skills if skill.name in skills_filter]
+        else:
+            skills = all_skills
+        skills_text = format_skills(skills) if skills else ""
+
+        if system_prompt:
+            if skills_text:
+                system_prompt += f"\n\n## Skills\n{skills_text}"
+        else:
+            system_prompt = DEFAULT_SYSTEM_PREFIX
+            if skills_text:
+                system_prompt += f"\n\n## Skills\n{skills_text}"
+
+        if config.max_iterations and config.max_iterations > 0:
+            system_prompt += (
+                "\n\n## Execution Constraint\n"
+                f"这是一项有严格步数限制的任务。你最多可以用 {config.max_iterations} "
+                "轮交互来解决问题。\n请务必【边查边写、随时存档】！每当你收集或得出一个"
+                "有价值的独立结果（如收集到一个独立 Case），请立刻调用工具写入或追加"
+                "到结果文件中，绝对不要等到所有任务都做完再最后一次性输出。这样"
+                "即使触达步数上限被强制打断，你已经收集的成果也能安全保留！"
+            )
+
+        if config.memory:
+            try:
+                from agent.core.memory import format_memory_injection, load_memory_files
+
+                files = load_memory_files(config.memory)
+                memory_text = format_memory_injection(files)
+                if memory_text:
+                    system_prompt += f"\n\n{memory_text}"
+            except Exception as exc:
+                self._log.warning("[Memory] 加载记忆失败，跳过注入: %s", exc)
+
+        policy = resolve_policy(config)
+        if policy.completion_policy == CompletionPolicy.EXPLICIT_VALIDATION:
+            contract = ROLE_CONTRACTS.get(policy.role.value)
+            if contract:
+                system_prompt += f"\n\n{contract}"
+        return system_prompt
+
+    async def generate_task_name(
+        self,
+        messages: List[Dict[str, Any]],
+        *,
+        utility_llm_call: Optional[Callable[..., Awaitable[Dict[str, Any]]]],
+    ) -> str:
+        text_parts: List[str] = []
+        for message in messages:
+            content = message.get("content", "")
+            if isinstance(content, str):
+                text_parts.append(content)
+            elif isinstance(content, list):
+                for part in content:
+                    if isinstance(part, dict) and part.get("type") == "text":
+                        text_parts.append(part.get("text", ""))
+        raw_text = " ".join(text_parts).strip()
+        if not raw_text:
+            return TASK_NAME_FALLBACK
+
+        if utility_llm_call:
+            try:
+                result = await utility_llm_call(
+                    messages=[
+                        {
+                            "role": "system",
+                            "content": TASK_NAME_GENERATION_SYSTEM_PROMPT,
+                        },
+                        {"role": "user", "content": raw_text[:2000]},
+                    ],
+                    model="gpt-4o-mini",
+                )
+                title = result.get("content", "").strip()
+                if title and len(title) < 100:
+                    return title
+            except Exception:
+                self._log.debug(
+                    "Utility model failed to generate a task name",
+                    exc_info=True,
+                )
+        return raw_text[:50] + ("..." if len(raw_text) > 50 else "")
+
+
+__all__ = ["RunnerPromptRuntime"]

agent/agent/core/runner_tools.py

@@ -0,0 +1,187 @@
+"""Role-aware tool execution support for :mod:`agent.core.runner`.
+
+This module keeps tool selection, authorization, and protected context
+construction together.  ``AgentRunner`` remains the public facade and delegates
+to this internal component so existing callers and subclass overrides keep the
+same entry points.
+"""
+
+from __future__ import annotations
+
+import logging
+from typing import Any, Callable, Dict, List, Optional
+
+from agent.core.presets import get_preset
+from agent.orchestration.models import AgentRole, CompletionPolicy
+from agent.orchestration.policy import (
+    PLANNER_TOOLS,
+    VALIDATOR_TOOLS,
+    WORKER_TOOLS,
+    ResolvedAgentPolicy,
+)
+from agent.tools import ToolRegistry
+from agent.trace.goal_models import GoalTree
+from agent.trace.models import Trace
+
+
+class RunnerToolRuntime:
+    """Resolve and execute tools without owning the Agent loop."""
+
+    def __init__(
+        self,
+        registry: ToolRegistry,
+        policy: Any,
+        logger: logging.Logger,
+    ) -> None:
+        self._registry = registry
+        self._policy = policy
+        self._log = logger
+
+    def resolve_policy(
+        self,
+        config: Any,
+        trace: Optional[Trace] = None,
+    ) -> ResolvedAgentPolicy:
+        """Resolve role and effective tools from the preset, never host context."""
+        preset = get_preset(config.agent_type)
+        resolved = self._policy.resolve(config, preset, self._registry)
+        if trace and trace.agent_role != resolved.role.value:
+            raise ValueError(
+                "Trace role mismatch: "
+                f"stored={trace.agent_role}, requested={resolved.role.value}"
+            )
+        if (
+            resolved.completion_policy == CompletionPolicy.EXPLICIT_VALIDATION
+            and resolved.role == AgentRole.LEGACY
+        ):
+            raise ValueError(
+                f"Preset '{config.agent_type}' has legacy role and cannot run "
+                "explicit_validation"
+            )
+        return resolved
+
+    def get_run_tool_schemas(
+        self,
+        config: Any,
+        resolved: Optional[ResolvedAgentPolicy] = None,
+        *,
+        get_tool_schemas: Optional[Callable[..., List[Dict[str, Any]]]] = None,
+    ) -> List[Dict[str, Any]]:
+        resolved = resolved or self.resolve_policy(config)
+        if resolved.completion_policy == CompletionPolicy.LEGACY_AUTO:
+            schema_resolver = get_tool_schemas or self.get_tool_schemas
+            schemas = schema_resolver(
+                config.tools,
+                config.tool_groups,
+                config.exclude_tools,
+            )
+            explicit_only = PLANNER_TOOLS | WORKER_TOOLS | VALIDATOR_TOOLS
+            return [
+                schema
+                for schema in schemas
+                if schema.get("function", {}).get("name") not in explicit_only
+            ]
+        return self._registry.get_schemas(sorted(resolved.effective_tools))
+
+    def get_tool_schemas(
+        self,
+        tools: Optional[List[str]] = None,
+        tool_groups: Optional[List[str]] = None,
+        exclude_tools: Optional[List[str]] = None,
+    ) -> List[Dict[str, Any]]:
+        """Return the legacy union of tool groups and explicit tool names."""
+        if tool_groups is not None:
+            tool_names = set(self._registry.get_tool_names(groups=tool_groups))
+        else:
+            tool_names = set(self._registry.get_tool_names())
+        if tools is not None:
+            tool_names.update(tools)
+        if exclude_tools:
+            tool_names.difference_update(exclude_tools)
+        return self._registry.get_schemas(list(tool_names))
+
+    @staticmethod
+    def build_protected_context(
+        config: Any,
+        trace: Trace,
+        goal_tree: Optional[GoalTree],
+        sequence: int,
+        *,
+        runner: Any,
+        trace_store: Any,
+        task_coordinator: Any,
+        side_branch: Optional[Dict[str, Any]] = None,
+    ) -> Dict[str, Any]:
+        """Merge host context first so framework identity always wins."""
+        context = dict(config.context or {})
+        trace_context = trace.context or {}
+        context.update(
+            {
+                "store": trace_store,
+                "trace_id": trace.trace_id,
+                "root_trace_id": trace_context.get("root_trace_id", trace.trace_id),
+                "goal_id": goal_tree.current_id if goal_tree else None,
+                "runner": runner,
+                "goal_tree": goal_tree,
+                "knowledge_config": config.knowledge,
+                "sequence": sequence,
+                "side_branch": side_branch,
+                "role": trace.agent_role,
+                "completion_policy": CompletionPolicy(config.completion_policy).value,
+                "task_id": trace_context.get("task_id"),
+                "spec_version": trace_context.get("spec_version"),
+                "attempt_id": trace_context.get("attempt_id"),
+                "snapshot_id": trace_context.get("snapshot_id"),
+                "validation_id": trace_context.get("validation_id"),
+                "coordinator": task_coordinator,
+            }
+        )
+        return context
+
+    async def execute_authorized_tool(
+        self,
+        tool_name: str,
+        tool_args: Dict[str, Any],
+        tool_call_id: str,
+        config: Any,
+        trace: Trace,
+        goal_tree: Optional[GoalTree],
+        sequence: int,
+        *,
+        resolve_policy: Callable[[Any, Optional[Trace]], ResolvedAgentPolicy],
+        build_context: Callable[..., Dict[str, Any]],
+        side_branch: Optional[Dict[str, Any]] = None,
+    ) -> Any:
+        resolved = resolve_policy(config, trace)
+        if resolved.completion_policy == CompletionPolicy.EXPLICIT_VALIDATION:
+            authorization = self._policy.authorize(
+                resolved.role,
+                tool_name,
+                resolved,
+            )
+            if not authorization.allowed:
+                self._log.warning(
+                    "Rejected unauthorized tool call: %s",
+                    authorization.reason,
+                )
+                return {
+                    "text": f"Error: {authorization.reason}",
+                    "error": "unauthorized_tool",
+                }
+        context = build_context(
+            config,
+            trace,
+            goal_tree,
+            sequence,
+            side_branch=side_branch,
+        )
+        context["tool_call_id"] = tool_call_id
+        return await self._registry.execute(
+            tool_name,
+            tool_args,
+            uid=config.uid or "",
+            context=context,
+        )
+
+
+__all__ = ["RunnerToolRuntime"]

agent/agent/tools/builtin/knowledge.py

@@ -7,13 +7,11 @@
 import os
 import json
 import logging
-import subprocess
 import uuid
 import httpx
-from dataclasses import dataclass
 from typing import List, Dict, Optional, Any
 from agent.tools import tool, ToolResult, ToolContext
-from agent.core.prompts import build_reflect_prompt, COMPLETION_REFLECT_PROMPT
+from agent.core.knowledge_config import KnowledgeConfig as KnowledgeConfig
 
 logger = logging.getLogger(__name__)
 
@@ -21,76 +19,6 @@ logger = logging.getLogger(__name__)
 KNOWHUB_API = os.getenv("KNOWHUB_API", "http://localhost:8000").rstrip("/")
 
 
-# ===== 知识管理配置 =====
-
-@dataclass
-class KnowledgeConfig:
-    """知识提取与注入的配置"""
-
-    # 压缩时提取（消息量超阈值触发压缩时，在 Level 1 过滤前用完整 history 反思）
-    enable_extraction: bool = True         # 是否在压缩触发时提取知识
-    reflect_prompt: str = ""               # 自定义反思 prompt；空则使用默认，见 agent/core/prompts/knowledge.py:REFLECT_PROMPT
-
-    # agent运行完成后提取（不代表任务完成，agent 可能中途退出等待人工评估）
-    enable_completion_extraction: bool = True      # 是否在运行完成后提取知识
-    completion_reflect_prompt: str = ""            # 自定义复盘 prompt；空则使用默认，见 agent/core/prompts/knowledge.py:COMPLETION_REFLECT_PROMPT
-
-    # 提取-审核-提交两阶段开关（见 agent/docs/memory.md 第三节）
-    reflect_auto_commit: bool = False
-    # False（默认）: reflection 仅写 cognition_log: type="extraction_pending"，
-    #               人工通过 CLI（agent/cli/extraction_review.py）review + commit 才进 KnowHub
-    # True         : reflection 直接 upload_knowledge（旧行为），适合无人值守的 example
-
-    # 知识注入（agent切换当前工作的goal时，自动注入相关知识）
-    enable_injection: bool = True          # 是否在 focus goal 时自动注入相关知识
-
-    # 默认字段（保存/搜索时自动注入）
-    owner: str = ""                            # 所有者（空则尝试从 git config user.email 获取，再空则用 agent:{agent_id}）
-    default_tags: Optional[Dict[str, str]] = None      # 默认 tags（会与工具调用参数合并）
-    default_scopes: Optional[List[str]] = None         # 默认 scopes（空则用 ["org:cybertogether"]）
-    default_search_types: Optional[List[str]] = None   # 默认搜索类型过滤
-    default_search_owner: str = ""                     # 默认搜索 owner 过滤（空则不过滤，支持多个owner用逗号分隔，如 "user1@example.com,user2@example.com"）
-
-    def get_reflect_prompt(self) -> str:
-        """压缩时反思 prompt"""
-        return self.reflect_prompt if self.reflect_prompt else build_reflect_prompt()
-
-    def get_completion_reflect_prompt(self) -> str:
-        """任务完成后复盘 prompt"""
-        return self.completion_reflect_prompt if self.completion_reflect_prompt else COMPLETION_REFLECT_PROMPT
-
-    @property
-    def resolved_owner(self) -> str:
-        """解析后的 owner（优先级：配置 > git email > 'agent'）
-
-        供 inject_params key path 使用：knowledge_config.resolved_owner
-        """
-        if self.owner:
-            return self.owner
-
-        # 尝试从 git config 获取
-        try:
-            result = subprocess.run(
-                ["git", "config", "user.email"],
-                capture_output=True,
-                text=True,
-                timeout=2,
-            )
-            if result.returncode == 0 and result.stdout.strip():
-                return result.stdout.strip()
-        except Exception:
-            pass
-
-        return "agent"
-
-    def get_owner(self, agent_id: str = "agent") -> str:
-        """获取 owner（优先级：配置 > git email > agent:{agent_id}）"""
-        owner = self.resolved_owner
-        if owner == "agent" and agent_id != "agent":
-            return f"agent:{agent_id}"
-        return owner
-
-
 @tool(groups=["knowledge_internal"], hidden_params=["context"])
 async def knowledge_search(
     query: str,
@@ -670,8 +598,7 @@ async def resource_save(
         async with httpx.AsyncClient(timeout=30.0) as client:
             response = await client.post(f"{KNOWHUB_API}/api/resource", json=payload)
             response.raise_for_status()
-            data = response.json()
-
+            response.json()  # Preserve response-shape validation without retaining unused data.
         return ToolResult(
             title="✅ 资源已保存",
             output=f"资源 ID: {resource_id}\n类型: {content_type}\n标题: {title}",
@@ -758,7 +685,8 @@ async def tool_search(
     """
     try:
         params = {"q": query, "top_k": top_k}
-        if status: params["status"] = status
+        if status:
+            params["status"] = status
         async with httpx.AsyncClient(timeout=30.0) as client:
             res = await client.get(f"{KNOWHUB_API}/api/tool/search", params=params)
             res.raise_for_status()
@@ -778,7 +706,8 @@ async def tool_list(
     """列出工具列表 (Tool)"""
     try:
         params = {"limit": limit, "offset": offset}
-        if status: params["status"] = status
+        if status:
+            params["status"] = status
         async with httpx.AsyncClient(timeout=30.0) as client:
             res = await client.get(f"{KNOWHUB_API}/api/tool", params=params)
             res.raise_for_status()
@@ -860,7 +789,8 @@ async def requirement_list(
     """列出需求列表 (Requirement)"""
     try:
         params = {"limit": limit, "offset": offset}
-        if status: params["status"] = status
+        if status:
+            params["status"] = status
         async with httpx.AsyncClient(timeout=30.0) as client:
             res = await client.get(f"{KNOWHUB_API}/api/requirement", params=params)
             res.raise_for_status()

agent/tests/test_runner_components.py

@@ -0,0 +1,166 @@
+import pytest
+
+from agent import AgentRunner as PublicAgentRunner
+from agent import RunConfig as PublicRunConfig
+from agent.core.runner import AgentRunner, RunConfig
+from agent.tools.registry import ToolRegistry
+from agent.trace.models import Trace
+
+
+def test_runner_public_imports_remain_stable():
+    assert PublicAgentRunner is AgentRunner
+    assert PublicRunConfig is RunConfig
+
+
+@pytest.mark.asyncio
+async def test_prompt_runtime_preserves_host_prompt_and_iteration_constraint(tmp_path):
+    runner = AgentRunner(skills_dir=str(tmp_path))
+
+    prompt = await runner._build_system_prompt(
+        RunConfig(max_iterations=7),
+        base_prompt="host contract",
+    )
+
+    assert prompt.startswith("host contract\n\n## Execution Constraint")
+    assert "最多可以用 7 轮交互" in prompt
+
+
+@pytest.mark.asyncio
+async def test_task_name_runtime_uses_current_utility_model_callback():
+    runner = AgentRunner()
+
+    async def utility_llm_call(**_kwargs):
+        return {"content": "stable title"}
+
+    # Preserve the historical ability to configure this dependency after
+    # construction; the internal component must not capture a stale value.
+    runner.utility_llm_call = utility_llm_call
+
+    assert (
+        await runner._generate_task_name([{"role": "user", "content": "task"}])
+        == "stable title"
+    )
+
+
+@pytest.mark.asyncio
+async def test_task_name_runtime_keeps_text_fallback_on_utility_failure():
+    async def failing_utility(**_kwargs):
+        raise RuntimeError("unavailable")
+
+    runner = AgentRunner(utility_llm_call=failing_utility)
+    text = "x" * 60
+
+    assert await runner._generate_task_name([{"role": "user", "content": text}]) == (
+        "x" * 50 + "..."
+    )
+
+
+@pytest.mark.asyncio
+async def test_tool_runtime_keeps_runner_override_hooks() -> None:
+    registry = ToolRegistry()
+
+    async def sample() -> str:
+        return "ok"
+
+    registry.register(
+        sample,
+        schema={
+            "type": "function",
+            "function": {
+                "name": "sample",
+                "description": "sample",
+                "parameters": {"type": "object", "properties": {}},
+            },
+        },
+    )
+
+    class ExtensibleRunner(AgentRunner):
+        policy_calls = 0
+        context_calls = 0
+        schema_calls = 0
+
+        def _resolve_run_policy(self, config, trace=None):
+            self.policy_calls += 1
+            return super()._resolve_run_policy(config, trace)
+
+        def _build_protected_tool_context(
+            self,
+            config,
+            trace,
+            goal_tree,
+            sequence,
+            side_branch=None,
+        ):
+            self.context_calls += 1
+            return super()._build_protected_tool_context(
+                config,
+                trace,
+                goal_tree,
+                sequence,
+                side_branch=side_branch,
+            )
+
+        def _get_tool_schemas(
+            self,
+            tools=None,
+            tool_groups=None,
+            exclude_tools=None,
+        ):
+            self.schema_calls += 1
+            return super()._get_tool_schemas(
+                tools,
+                tool_groups,
+                exclude_tools,
+            )
+
+    runner = ExtensibleRunner(tool_registry=registry)
+    runner._get_run_tool_schemas(RunConfig(tools=["sample"], tool_groups=[]))
+    result = await runner._execute_authorized_tool(
+        "sample",
+        {},
+        "call-1",
+        RunConfig(tools=["sample"], tool_groups=[]),
+        Trace(trace_id="legacy", mode="agent", agent_role="legacy"),
+        None,
+        1,
+    )
+
+    assert result == "ok"
+    assert runner.policy_calls == 2
+    assert runner.context_calls == 1
+    assert runner.schema_calls == 1
+
+
+@pytest.mark.asyncio
+async def test_image_runtime_keeps_runner_override_hooks() -> None:
+    class ExtensibleRunner(AgentRunner):
+        image_calls = 0
+
+        async def _process_image_size(
+            self,
+            base64_url,
+            max_size=512,
+            min_size=11,
+        ):
+            self.image_calls += 1
+            return "data:image/jpeg;base64,processed"
+
+    runner = ExtensibleRunner()
+    messages = [
+        {
+            "role": "tool",
+            "content": [
+                {
+                    "type": "image_url",
+                    "image_url": {"url": "data:image/png;base64,source"},
+                }
+            ],
+        },
+        {"role": "assistant", "content": "observed"},
+    ]
+
+    optimized = await runner._optimize_images(messages, "gpt-4o")
+
+    assert runner.image_calls == 1
+    assert optimized[0]["content"][0]["image_url"]["url"].endswith("processed")
+    assert messages[0]["content"][0]["image_url"]["url"].endswith("source")