fix+refactor: 安全漏洞修复 & 架构师审查改进

.github/workflows/pylint.yml

@@ -17,7 +17,7 @@ jobs:
     - name: Install dependencies
       run: |
         python -m pip install --upgrade pip
-        pip install pylint
+        pip install pylint tomli
     - name: Analysing the code with pylint
       run: |
-        pylint $(git ls-files '*.py')
+        pylint $(git ls-files '*.py') --fail-under=8.0

agents/events.py

@@ -18,6 +18,11 @@
 from pathlib import Path
 from typing import Any
 
+# On Linux a single write() on a file opened with O_APPEND is atomic up to
+# PIPE_BUF (4096 B) for regular files. All events are well under that limit,
+# so this gives us safe cross-process concurrent appends without a file lock.
+_APPEND_FLAGS = os.O_WRONLY | os.O_APPEND | os.O_CREAT
+
 BASE_DIR = Path(__file__).resolve().parent.parent
 VOL_HOST = Path(os.getenv("VOL_HOST", BASE_DIR / "vol"))
 EVENTS_DIR = VOL_HOST / "logs"
@@ -41,8 +46,12 @@ def emit(kind: str, **payload: Any) -> None:
     event = {"seq": _next_seq(), "ts": time.time(), "kind": kind, **payload}
     try:
         EVENTS_DIR.mkdir(parents=True, exist_ok=True)
-        with EVENTS_FILE.open("a", encoding="utf-8") as f:
-            f.write(json.dumps(event, ensure_ascii=False) + "\n")
+        line = (json.dumps(event, ensure_ascii=False) + "\n").encode("utf-8")
+        fd = os.open(str(EVENTS_FILE), _APPEND_FLAGS, 0o644)
+        try:
+            os.write(fd, line)
+        finally:
+            os.close(fd)
     except Exception as exc:  # noqa: BLE001
         # Events must never break the pipeline. Swallow and print once.
         print(f"  [events] emit 失败 ({kind}): {exc}")
@@ -56,10 +65,14 @@ def emit(kind: str, **payload: Any) -> None:
 
 
 def reset() -> None:
-    """Truncate the events file — call at pipeline start for a fresh run."""
+    """Truncate the events file — call at pipeline start for a fresh run.
+
+    Only safe to call when the pipeline subprocess is not running.
+    """
     try:
         EVENTS_DIR.mkdir(parents=True, exist_ok=True)
-        EVENTS_FILE.write_text("", encoding="utf-8")
+        fd = os.open(str(EVENTS_FILE), os.O_WRONLY | os.O_TRUNC | os.O_CREAT, 0o644)
+        os.close(fd)
         global _seq
         with _lock:
             _seq = 0

agents/extensions/registry.py

@@ -79,6 +79,7 @@ class ExtensionRegistry:
     skills: list[SkillEntry] = field(default_factory=list)
     event_handlers: list[EventEntry] = field(default_factory=list)
     mcp: MCPClient | None = None
+    _loaded: bool = field(default=False, repr=False)
 
     def emit_event(self, kind: str, payload: dict) -> None:
         """Dispatch an event to all subscribed handlers. Never raises."""
@@ -191,6 +192,9 @@ def _load_plugin_module(name: str) -> list[Plugin]:
 def load_all() -> ExtensionRegistry:
     """Load plugins + MCP + local skills. Idempotent — safe to call twice."""
     registry = get_registry()
+    if registry._loaded:
+        return registry
+    registry._loaded = True
     # 1. Plugins
     plugins_cfg = _load_toml(PLUGINS_CONFIG)
     enabled = plugins_cfg.get("enabled")

agents/model_router.py

@@ -38,8 +38,9 @@
 from __future__ import annotations
 
 import os
-import tomllib
 from pathlib import Path
+
+from agents.utils import load_toml
 #采用相对路径
 CURRENT_DIR = Path(__file__).parent.parent   # project root (E:\mathmodel)
 DEFAULT_PATH = CURRENT_DIR / "config" / "model_routes.toml"
@@ -63,17 +64,8 @@
 # 这个模块负责根据任务类型（如不同阶段的建模任务）动态选择和配置使用的语言模型。
 # 通过读取一个 TOML 格式的配置文件（路径可通过环境变量 MODEL_ROUTES_FILE 指定），它定义了不同任务对应的模型列表、超时时间和
 def _load_routes() -> dict:
-    if _ROUTES_PATH.exists():
-        try:
-            # Use utf-8-sig to tolerate BOM written by some editors/shells.
-            text = _ROUTES_PATH.read_text(encoding="utf-8-sig")
-            #Byte Order mask是win系统保存文件时候细化在开头偷偷夹带私货的几个特殊字符，utf-8-sig编码可以自动识别并去除这些字符，避免解析错误。
-            data = tomllib.loads(text)
-            if isinstance(data, dict):# 确保解析结果是一个字典
-                return data
-        except Exception:
-            return _FALLBACK
-    return _FALLBACK
+    data = load_toml(_ROUTES_PATH)
+    return data if isinstance(data, dict) and data else _FALLBACK
 
 # 提供了三个主要函数：get_task_route 根据任务类型获取对应的路由配置；
 

agents/utils.py

@@ -7,6 +7,24 @@
 from pathlib import Path
 
 
+def load_toml(path: Path) -> dict:
+    """Parse a TOML file; returns {} on missing file or parse error.
+
+    Uses tomllib (Python ≥3.11 stdlib) with tomli as fallback for older runtimes.
+    """
+    if not path.exists():
+        return {}
+    text = path.read_text(encoding="utf-8-sig")  # utf-8-sig strips Windows BOM
+    try:
+        try:
+            import tomllib  # type: ignore[import]  # Python 3.11+
+        except ModuleNotFoundError:
+            import tomli as tomllib  # type: ignore[import,no-redef]
+        return tomllib.loads(text)
+    except Exception:  # noqa: BLE001
+        return {}
+
+
 def parse_json(raw: str) -> dict:
     """Parse JSON content from raw LLM output."""
     text = raw.strip()

sandbox/loop.py

@@ -13,6 +13,16 @@
 from sandbox.healer import heal
 
 CONTEXT_PATH = Path(os.getenv("CONTEXT_STORE", "context_store/context.json"))
+_PIPELINE_CFG_PATH = Path(os.getenv("PIPELINE_CFG_PATH", Path(__file__).parent.parent / "config" / "pipeline.json"))
+
+
+def _load_max_iter() -> int:
+    """Read max_heal_iterations from pipeline.json, fall back to env var."""
+    try:
+        cfg = json.loads(_PIPELINE_CFG_PATH.read_text(encoding="utf-8"))
+        return int(cfg.get("max_heal_iterations", os.getenv("MAX_HEAL_ITERATIONS", "5")))
+    except Exception:
+        return int(os.getenv("MAX_HEAL_ITERATIONS", "5"))
 
 
 def _run_script(script_host_path: str) -> tuple[int, str, str]:
@@ -41,7 +51,7 @@ def _update_ctx(updates: dict) -> None:
 def execute_with_healing(script_name: str) -> dict:
     """Run script with auto-healing loop."""
     script_host_path = str(vol_host() / "scripts" / script_name)
-    max_iter = int(os.getenv("MAX_HEAL_ITERATIONS", "5"))
+    max_iter = _load_max_iter()
 
     _update_ctx({"status": "running", "current_script": script_host_path, "iterations": 0})