# corecoder/llm.py:105-127def chat(    self,    messages: list[dict],    tools: list[dict] | None = None,    on_token=None,) -> LLMResponse:    params: dict = {        "model": self.model,        "messages": messages,        "stream": True,           # 关键：始终使用流式        **self.extra,    }    if tools:        params["tools"] = tools    # stream_options 是 OpenAI 的扩展字段，不是所有 provider 都支持    try:        params["stream_options"] = {"include_usage": True}        stream = self._call_with_retry(params)    except Exception:        params.pop("stream_options", None)        stream = self._call_with_retry(params)

# corecoder/llm.py:129-162content_parts: list[str] = []tc_map: dict[int, dict] = {}  # index -> {id, name, arguments_str}prompt_tok = 0completion_tok = 0for chunk in stream:    # token 用量在最后一个 chunk 中    if chunk.usage:        prompt_tok = chunk.usage.prompt_tokens        completion_tok = chunk.usage.completion_tokens    if not chunk.choices:        continue    delta = chunk.choices[0].delta    # 累积文本内容    if delta.content:        content_parts.append(delta.content)        if on_token:            on_token(delta.content)    # 实时回调给 CLI 层做流式显示    # 累积 tool_calls —— 这是最精妙的部分    if delta.tool_calls:        for tc_delta in delta.tool_calls:            idx = tc_delta.index       # 用 index 区分不同的 tool_call            if idx not in tc_map:                tc_map[idx] = {"id": "", "name": "", "args": ""}            if tc_delta.id:                tc_map[idx]["id"] = tc_delta.id            if tc_delta.function:                if tc_delta.function.name:                    tc_map[idx]["name"] = tc_delta.function.name                if tc_delta.function.arguments:                    tc_map[idx]["args"] += tc_delta.function.arguments  # 注意：这里是 += 拼接

chunk 1: tc_delta.id = "call_abc", tc_delta.function.name = "edit_file"chunk 2: tc_delta.function.arguments = '{"file_path": "main'chunk 3: tc_delta.function.arguments = '.py", "old_string": "from ut'chunk 4: tc_delta.function.arguments = 'ils import halper", "new_string"...'chunk 5: tc_delta.function.arguments = ': "from utils import helper"}'

# corecoder/llm.py:164-172parsed: list[ToolCall] = []for idx insorted(tc_map):    raw = tc_map[idx]    try:        args = json.loads(raw["args"])    except (json.JSONDecodeError, KeyError):        args = {}    # 解析失败就传空参数，让工具自己报错    parsed.append(ToolCall(id=raw["id"], name=raw["name"], arguments=args))

# corecoder/llm.py:184-199def _call_with_retry(self, params: dict, max_retries: int = 3):    """Retry on transient errors with exponential backoff."""    for attempt in range(max_retries):        try:            return self.client.chat.completions.create(**params)        except (RateLimitError, APITimeoutError, APIConnectionError) as e:            if attempt == max_retries - 1:                raise               # 最后一次还失败就不忍了，直接抛异常            wait = 2 ** attempt     # 1秒 → 2秒 → 4秒            time.sleep(wait)        except APIError as e:            # 5xx = 服务端错误，重试；4xx = 客户端错误，不重试            if e.status_code and e.status_code >= 500 and attempt < max_retries - 1:                time.sleep(2 ** attempt)            else:                raise

# corecoder/llm.py:48-76 (定价表)_PRICING = {    # OpenAI    "gpt-5.4": (2.5, 15),          # (input $/M tokens, output $/M tokens)    "gpt-4o": (2.5, 10),    "gpt-4o-mini": (0.15, 0.6),    # DeepSeek    "deepseek-chat": (0.27, 1.10),    # Anthropic Claude    "claude-opus-4-6": (5, 25),    # Alibaba Qwen    "qwen-max": (0.78, 3.9),    # Moonshot Kimi    "kimi-k2.5": (0.6, 3),    # ... 共 15+ 个模型}

# corecoder/llm.py:93-103@propertydef estimated_cost(self) -> float | None:    pricing = _PRICING.get(self.model)    if not pricing:        return None       # 未知模型返回 None，不会瞎猜    input_rate, output_rate = pricing    return (        self.total_prompt_tokens * input_rate / 1_000_000        + self.total_completion_tokens * output_rate / 1_000_000    )

# corecoder/tools/base.py（完整代码，仅 28 行）from abc import ABC, abstractmethodclass Tool(ABC):    """Minimal tool interface. Subclass this to add new capabilities."""    name: str           # 工具名称，如 "bash"、"edit_file"    description: str    # 给 LLM 看的自然语言描述    parameters: dict    # JSON Schema，定义工具接受的参数    @abstractmethod    def execute(self, **kwargs) -> str:        """Run the tool and return a text result."""        ...    def schema(self) -> dict:        """OpenAI function-calling schema."""        return {            "type": "function",            "function": {                "name": self.name,                "description": self.description,                "parameters": self.parameters,            },        }

# corecoder/tools/__init__.py（完整代码）from .bash import BashToolfrom .read import ReadFileToolfrom .write import WriteFileToolfrom .edit import EditFileToolfrom .glob_tool import GlobToolfrom .grep import GrepToolfrom .agent import AgentToolALL_TOOLS = [    BashTool(),    ReadFileTool(),    WriteFileTool(),    EditFileTool(),    GlobTool(),    GrepTool(),    AgentTool(),]def get_tool(name: str):    """Look up a tool by name."""    for t in ALL_TOOLS:        if t.name == name:            return t    return None

# corecoder/tools/edit.py:44-73def execute(self, file_path: str, old_string: str, new_string: str) -> str:    try:        p = Path(file_path).expanduser().resolve()        if not p.exists():            return f"Error: {file_path} not found"        content = p.read_text()        occurrences = content.count(old_string)        if occurrences == 0:            # 没找到 → 报错 + 显示文件开头（帮助 LLM 纠正）            preview = content[:500] + ("..." if len(content) > 500 else "")            return (                f"Error: old_string not found in {file_path}.\n"                f"File starts with:\n{preview}"            )        if occurrences > 1:            # 找到多个 → 报错 + 提示加更多上下文            return (                f"Error: old_string appears {occurrences} times in {file_path}. "                f"Include more surrounding lines to make it unique."            )        # 恰好找到 1 次 → 替换        new_content = content.replace(old_string, new_string, 1)        p.write_text(new_content)        _changed_files.add(str(p))    # 记录修改过的文件        # 生成 unified diff 供用户审查        diff = _unified_diff(content, new_content, str(p))        return f"Edited {file_path}\n{diff}"    except Exception as e:        return f"Error: {e}"

# corecoder/tools/edit.py:76-89def _unified_diff(old: str, new: str, filename: str, context: int = 3) -> str:    """Generate a compact unified diff between old and new file content."""    old_lines = old.splitlines(keepends=True)    new_lines = new.splitlines(keepends=True)    diff = difflib.unified_diff(        old_lines, new_lines,        fromfile=f"a/{filename}", tofile=f"b/{filename}",        n=context,    )    result = "".join(diff)    # 过大的 diff 截断，避免撑爆上下文    if len(result) > 3000:        result = result[:2500] + "\n... (diff truncated)\n"    return result

# corecoder/tools/bash.py:19-29_DANGEROUS_PATTERNS = [    (r"\brm\s+(-\w*)?-r\w*\s+(/|~|\$HOME)", "recursive delete on home/root"),    (r"\brm\s+(-\w*)?-rf\s", "force recursive delete"),    (r"\bmkfs\b", "format filesystem"),    (r"\bdd\s+.*of=/dev/", "raw disk write"),    (r">\s*/dev/sd[a-z]", "overwrite block device"),    (r"\bchmod\s+(-R\s+)?777\s+/", "chmod 777 on root"),    (r":\(\)\s*\{.*:\|:.*\}", "fork bomb"),    (r"\bcurl\b.*\|\s*(sudo\s+)?bash", "pipe curl to bash"),    (r"\bwget\b.*\|\s*(sudo\s+)?bash", "pipe wget to bash"),]

# corecoder/tools/bash.py:82-87iflen(out) > 15_000:    out = (        out[:6000]        + f"\n\n... truncated ({len(out)} chars total) ...\n\n"        + out[-3000:]    )

# corecoder/tools/bash.py:103-116def _update_cwd(command: str, current_cwd: str):    """Track directory changes from cd commands."""    global _cwd    parts = command.split("&&")    for part in parts:        part = part.strip()        if part.startswith("cd "):            target = part[3:].strip().strip("'\"")            if target:                new_dir = os.path.normpath(                    os.path.join(current_cwd, os.path.expanduser(target))                )                if os.path.isdir(new_dir):                    _cwd = new_dir

# corecoder/tools/read.py:32-53def execute(self, file_path: str, offset: int = 1, limit: int = 2000) -> str:    try:        p = Path(file_path).expanduser().resolve()        if not p.exists():            return f"Error: {file_path} not found"        if not p.is_file():            return f"Error: {file_path} is a directory, not a file"        text = p.read_text(errors="replace")        lines = text.splitlines()        total = len(lines)        start = max(0, offset - 1)       # 1-based → 0-based        chunk = lines[start : start + limit]        numbered = [f"{start + i + 1}\t{ln}" for i, ln in enumerate(chunk)]        result = "\n".join(numbered)        if total > start + limit:            result += f"\n... ({total} lines total, showing {start+1}-{start+len(chunk)})"        return result or "(empty file)"    except Exception as e:        return f"Error: {e}"

# corecoder/tools/write.py（完整代码，仅 39 行）from pathlib import Pathfrom .base import Toolfrom .edit import _changed_filesclass WriteFileTool(Tool):    name = "write_file"    description = (        "Create a new file or completely overwrite an existing one. "        "For small edits to existing files, prefer edit_file instead."    )    parameters = {        "type": "object",        "properties": {            "file_path": {"type": "string", "description": "Path for the file"},            "content": {"type": "string", "description": "Full file content to write"},        },        "required": ["file_path", "content"],    }    def execute(self, file_path: str, content: str) -> str:        try:            p = Path(file_path).expanduser().resolve()            p.parent.mkdir(parents=True, exist_ok=True)   # 自动创建嵌套目录            p.write_text(content)            _changed_files.add(str(p))                     # 共享修改记录            n_lines = content.count("\n") + (1 if content and not content.endswith("\n") else 0)            return f"Wrote {n_lines} lines to {file_path}"        except Exception as e:            return f"Error: {e}"

# corecoder/tools/glob_tool.py:28-47（核心逻辑）def execute(self, pattern: str, path: str = ".") -> str:    base = Path(path).expanduser().resolve()    if not base.is_dir():        return f"Error: {path} is not a directory"    hits = list(base.glob(pattern))    # 按修改时间排序，最新的在前    hits.sort(key=lambda p: p.stat().st_mtime if p.exists() else 0, reverse=True)    total = len(hits)    shown = hits[:100]             # 最多返回 100 条    ...

# corecoder/tools/grep.py:8_SKIP_DIRS = {".git", "node_modules", "__pycache__", ".venv", "venv", ".tox", "dist", "build"}

# corecoder/tools/agent.py:36-58def execute(self, task: str) -> str:    if self._parent_agent is None:        return "Error: agent tool not initialized (no parent agent)"    from ..agent import Agent    # 延迟 import，避免循环依赖    parent = self._parent_agent    sub = Agent(        llm=parent.llm,                                          # ① 共享 LLM        tools=[t for t in parent.tools if t.name != "agent"],     # ② 排除自身        max_context_tokens=parent.context.max_tokens,        max_rounds=20,                                           # ③ 最多 20 轮    )    try:        result = sub.chat(task)        # 结果太长时截断，防止撑爆父级上下文        if len(result) > 5000:            result = result[:4500] + "\n... (sub-agent output truncated)"        return f"[Sub-agent completed]\n{result}"    except Exception as e:        return f"Sub-agent error: {e}"