后台执行栈

前 14 章看的都是"用户输入 → Agent 答"的同步流程。但生产 Agent 里有一大半工作不在这条路上: 夜里自动跑数据 + 推送报告;白天并发跑几百个 prompt 生成训练轨迹;跑完后压缩到能塞进 SFT; 从开源仓库挑 SWE-Bench 任务做评测。 这一章拆 Hermes 的 5 个后台子系统:Cron、cronjob 工具、Batch Runner、Trajectory Compressor、Mini SWE Runner。 代码量加起来 7600 行——但每一行都解决一个"夜里没人盯着也得稳"的工程问题。

15.1三种"不在 turn 里跑"的场景

共同特点:没有用户看着。这意味着所有失败都得"自己救",所有产出都得记盘,所有资源都得有 cap。

flowchart TB
  subgraph Foreground["前 14 章 — 同步 Agent"]
    U["用户消息"] --> Loop["run_conversation"]
    Loop --> R["回复"]
  end
  subgraph Background["本章 — 后台执行栈"]
    direction TB
    Sched["Cron Scheduler
(tick 每 60s)"] --> Job["定时 Agent"]
    LLM["LLM 自己排课
(cronjob 工具)"] --> Job
    Batch["batch_runner
(多进程 Pool)"] --> Traj["轨迹文件"]
    Traj --> Comp["trajectory_compressor
(LLM 总结中段)"]
    Comp --> SFT["SFT 训练集"]
    SWE["mini_swe_runner"] --> Eval["pass/fail 报告"]
  end
  Background -.-> KeyPoints["共同关切:
失败自救 · 持久化 · 资源 cap"]:::accent
  classDef accent fill:#f5ede0,stroke:#8b1538,color:#8b1538

15.2Cron 调度:jobs.py + scheduler.py

15.2.1 Job 持久化结构

Cron 的"job 数据库"是一个 JSON 文件 ~/.hermes/cron/jobs.json。粗暴但有效。看 job schema(来自 create_job()):

cron/jobs.py:654 (节选)
job = {
    "id": job_id,                       # UUID hex[:12], FS-safe
    "name": name or label_source[:50],
    "prompt": prompt_text,               # 给 Agent 的指令
    "skills": normalized_skills,         # 要加载的 skill
    "model": normalized_model,           # 单 job 模型覆盖(可空)
    "provider": normalized_provider,
    "base_url": normalized_base_url,
    "script": normalized_script,         # 预脚本路径 ~/.hermes/scripts/
    "no_agent": normalized_no_agent,     # True = 跳过 LLM,直接跑脚本
    "context_from": context_from,       # 链式上游 job IDs
    "schedule": parsed_schedule,         # {kind, minutes, expr, run_at}
    "repeat": {"times": repeat, "completed": 0},
    "enabled": True,
    "state": "scheduled",             # scheduled|paused|completed|error
    "next_run_at": compute_next_run(parsed_schedule),
    "last_run_at": None,
    "deliver": deliver,                  # local|origin|telegram:xxx
    "origin": origin,                    # {platform, chat_id, thread_id}
    "enabled_toolsets": normalized_toolsets,
    "workdir": normalized_workdir,       # 项目根(AGENTS.md/CLAUDE.md 注入)
    "profile": normalized_profile,       # Hermes profile 隔离
}

持久化用 tempfile + atomic rename + fsync,owner-only 权限(0600):

cron/jobs.py:455-465
def save_jobs(jobs):
    ensure_dirs()
    fd, tmp_path = tempfile.mkstemp(dir=str(JOBS_FILE.parent),
                                     suffix='.tmp', prefix='.jobs_')
    try:
        with os.fdopen(fd, 'w', encoding='utf-8') as f:
            json.dump({"jobs": jobs, "updated_at": _hermes_now().isoformat()}, f)
            f.flush()
            os.fsync(f.fileno())            # 强制落盘
        atomic_replace(tmp_path, JOBS_FILE)   # 原子 rename
        _secure_file(JOBS_FILE)               # chmod 0600

15.2.2 调度解析:四种格式统一

cron/jobs.py:209 (parse_schedule)
"30m"                    → 30 分钟后一次   (kind="once", run_at=now+30m)
"every 30m"              → 每 30 分钟重复  (kind="interval", minutes=30)
"0 9 * * *"              → 标准 cron 表达式 (kind="cron", expr="0 9 * * *")
"every 2h"               → 每 2 小时       (kind="interval", minutes=120)
"2026-02-03T14:00:00"    → 一次性绝对时刻   (kind="once", run_at=ISO_str)

"interval" 和 "every Xm" 是同一回事的两种说法——Hermes 都接受,内部统一成 minutes。 "cron" 用 croniter 库解析。所有格式都汇成 dict,后续 compute_next_run() 统一处理。

15.2.3 tick 循环 + 文件锁

Cron 没有独立 daemon ——它在每次 Hermes Gateway 进程的事件循环里每 60 秒检查一次。 但如果用户同时跑了 CLI 和 Gateway,两边都想 tick 怎么办? 文件锁:

cron/scheduler.py:1843 (tick 入口)
def tick(verbose=True, adapters=None, loop=None) -> int:
    """检查并跑所有到期 job。用文件锁,只允许一个 tick 同时跑。"""
    lock_dir, lock_file = _get_lock_paths()
    lock_dir.mkdir(parents=True, exist_ok=True)

    lock_fd = None
    try:
        lock_fd = open(lock_file, "w", encoding="utf-8")
        if fcntl:
            fcntl.flock(lock_fd, fcntl.LOCK_EX | fcntl.LOCK_NB)   # 非阻塞
        elif msvcrt:
            msvcrt.locking(lock_fd.fileno(), msvcrt.LK_NBLCK, 1)
    except (OSError, IOError):
        logger.debug("Tick skipped — 另一实例持锁")
        if lock_fd is not None:
            lock_fd.close()
        return 0

    try:
        due_jobs = get_due_jobs()
        ...

设计要点:

• 非阻塞锁(LOCK_NB)——抢不到立刻返回,不等。下一轮 60 秒后再试。
• 跨平台—— Unix 用 fcntl.flock,Windows 用 msvcrt.locking。
• 锁文件路径用 get_hermes_home() 解析,respect 当前 profile。
• 先 advance_next_run 再执行(第 1885-1888 行):在锁内先把所有到期 job 的 next_run_at 推到下一轮,然后才开始执行。这样即使执行慢、下一个 tick 看到时,也不会重复触发。这是至多一次语义的保证。

15.2.4 catchup window:120s–2h clamp

问题:Hermes 重启,中间错过了 6 小时的"每 5 分钟一次"任务。要补跑 72 次吗?显然不要。

Hermes 的策略:用 grace window 决定补跑还是 fast-forward。

cron/jobs.py:344-373
def _compute_grace_seconds(schedule: dict) -> int:
    """算出一个 job 最多可以晚多少秒还允许 catchup,
    超出就 fast-forward 到下一次。

    用周期的一半,clamp 在 [120s, 7200s] 区间。
    每日 job → 12h//2 = 6h → 被 clamp 到 2h
    每小时 job → 30 分钟
    每 5 分钟 job → 2.5 分钟 → 被 clamp 到 120s
    """
    MIN_GRACE = 120
    MAX_GRACE = 7200   # 2 hours

    kind = schedule.get("kind")

    if kind == "interval":
        period_seconds = schedule.get("minutes", 1) * 60
        grace = period_seconds // 2
        return max(MIN_GRACE, min(grace, MAX_GRACE))

    if kind == "cron" and HAS_CRONITER:
        try:
            now = _hermes_now()
            cron = croniter(schedule["expr"], now)
            first = cron.get_next(datetime)
            second = cron.get_next(datetime)
            period_seconds = int((second - first).total_seconds())
            grace = period_seconds // 2
            return max(MIN_GRACE, min(grace, MAX_GRACE))
        except Exception:
            pass

    return MIN_GRACE

在 _get_due_jobs_locked() 里应用:

cron/jobs.py:1065-1085
if kind in {"cron", "interval"} and (now - next_run_dt).total_seconds() > grace:
    # 错过 grace window — 这是"陈旧"任务,跳过本次,推到下一次
    new_next = compute_next_run(schedule, now.isoformat())
    if new_next:
        logger.info(
            "Job '%s' missed scheduled time (%s, grace=%ds). "
            "Fast-forwarding to next run: %s",
            job.get("name", job["id"]), next_run, grace, new_next,
        )
        for rj in raw_jobs:
            if rj["id"] == job["id"]:
                rj["next_run_at"] = new_next
                needs_save = True
                break
        continue   # 跳过本次

15.2.5 Inactivity Timeout:基于"活动",不是 wall-clock

看 AGENTS.md 里说"3 分钟硬中断"。但读源码发现真实实现是 600 秒 inactivity-based 超时:

cron/scheduler.py:1644-1730 (节选)
# Job 可以跑几小时,只要它在调工具或收 stream tokens;
# 但如果 600 秒没活动(API 卡住或工具死循环),就杀。
# 默认 600s;HERMES_CRON_TIMEOUT 覆盖;0 = 无限。
_cron_timeout = 600.0
_cron_inactivity_limit = _cron_timeout if _cron_timeout > 0 else None
_POLL_INTERVAL = 5.0
_cron_pool = concurrent.futures.ThreadPoolExecutor(max_workers=1)

_cron_context = contextvars.copy_context()
_cron_future = _cron_pool.submit(_cron_context.run, agent.run_conversation, prompt)
_inactivity_timeout = False

try:
    while True:
        done, _ = concurrent.futures.wait({_cron_future}, timeout=_POLL_INTERVAL)
        if done:
            result = _cron_future.result()
            break
        # 还在跑 — 检查 inactivity
        _idle_secs = 0.0
        if hasattr(agent, "get_activity_summary"):
            _act = agent.get_activity_summary()
            _idle_secs = _act.get("seconds_since_activity", 0.0)
        if _idle_secs >= _cron_inactivity_limit:
            _inactivity_timeout = True
            break

if _inactivity_timeout:
    _activity = agent.get_activity_summary()
    logger.error(
        "Job '%s' idle %.0fs (limit %.0fs) | last=%s | iter=%s/%s | tool=%s",
        job_name, _activity["seconds_since_activity"], _cron_inactivity_limit,
        _activity["last_activity_desc"], _activity["api_call_count"],
        _activity["max_iterations"], _activity.get("current_tool") or "none",
    )
    agent.interrupt("Cron job timed out (inactivity)")
    raise TimeoutError(...)

15.2.6 Cron-spawned AIAgent 的差异

看 cron/scheduler.py:1611-1642 构造 cron 用的 AIAgent:

cron/scheduler.py:1611-1642 (关键字段)
agent = AIAgent(
    model=model,
    max_iterations=max_iterations,
    enabled_toolsets=_resolve_cron_enabled_toolsets(job, _cfg),
    disabled_toolsets=_resolve_cron_disabled_toolsets(_cfg),
    quiet_mode=True,                          # 不打印 banner
    skip_context_files=not bool(_job_workdir),  # 无 workdir 就不自动找 AGENTS.md
    load_soul_identity=True,                   # 但 SOUL.md 总是加载
    skip_memory=True,                          # 关键!不污染主 memory
    platform="cron",                            # 平台标签
    session_id=_cron_session_id,                  # 独立 session
    session_db=_session_db,
)

三个非显然的设计:

1. skip_memory=True:不 load Mem0/Supermemory 等持久 memory。 理由——cron job 是"自动跑",输出不该污染人类用户的 memory layer,反过来 memory 里的 上下文也不该影响 job 的客观执行。
2. disabled_toolsets 默认含 ["cronjob", "messaging", "clarify"]。 cron job 不能创建新 cron job(防递归),不能 send_message(模糊责任),不能 clarify(无人答)。
3. platform="cron":工具集筛选时可以针对 cron 应用特殊限制 (例如某些工具只在 cli 模式有意义)。

15.3cronjob 工具:让 Agent 自己排课

这是 LLM-facing 的工具。Agent 在跑过程中可以决定"这事 1 小时后再做"。

15.3.1 9 个 actions

tools/cronjob_tools.py (schema 节选)
"action": {
    "enum": [
        "create", "list", "show", "update", "remove",
        "pause", "resume", "run", "run_now", "trigger",
    ]
}

所有 cron 操作收敛在一个工具里——schema 里的 description 把所有动词解释清楚。 LLM 选 action,余下参数按需带。比给 10 个独立工具紧凑。

15.3.2 Prompt Injection 扫描——16 类 pattern

因为 cron job 的 prompt 是"将来某个时刻自动执行"的,一次创建潜在跑 N 次。 所以创建时必须扫:

tools/cronjob_tools.py:43-140 (扫描器)
_CRON_THREAT_PATTERNS = [
    (r'ignore\s+(?:\w+\s+)*(?:previous|all|above|prior)\s+(?:\w+\s+)*instructions', "prompt_injection"),
    (r'do\s+not\s+tell\s+the\s+user', "deception_hide"),
    (r'system\s+prompt\s+override', "sys_prompt_override"),
    (r'disregard\s+(your|all|any)\s+(instructions|rules|guidelines)', "disregard_rules"),
    (r'cat\s+[^\n]*(\.env|credentials|\.netrc|\.pgpass)', "read_secrets"),
    (r'authorized_keys', "ssh_backdoor"),
    (r'/etc/sudoers|visudo', "sudoers_mod"),
    (r'rm\s+-rf\s+/', "destructive_root_rm"),
]
_CRON_SECRET_VAR_RE = r'\$\{?\w*(?:KEY|TOKEN|SECRET|PASSWORD|CREDENTIAL|API)\w*\}?'
_CRON_EXFIL_COMMAND_PATTERNS = [
    (rf'curl\s+[^\n]*https?://[^\s"\'`]*{_CRON_SECRET_VAR_RE}', "exfil_curl_url"),
    (rf'wget\s+[^\n]*https?://[^\s"\'`]*{_CRON_SECRET_VAR_RE}', "exfil_wget_url"),
    ...
]

def _scan_cron_prompt(prompt: str) -> str:
    # GitHub API token 是合法的(自带 skill 用)
    github_auth_header = re.search(r'curl\s+...token\s+...api.github.com', prompt, re.IGNORECASE)
    prompt_to_scan = prompt.replace(...) if github_auth_header else prompt

    # 检查不可见 Unicode
    for char in _CRON_INVISIBLE_CHARS:
        if char in _strip_legitimate_emoji_zwj(prompt_to_scan):
            return f"Blocked: invisible unicode U+{ord(char):04X}"

    # 威胁模式
    for pattern, pid in _CRON_THREAT_PATTERNS:
        if re.search(pattern, prompt_to_scan, re.IGNORECASE):
            return f"Blocked: threat pattern '{pid}'"

    # 数据泄露模式
    for pattern, pid in _CRON_EXFIL_COMMAND_PATTERNS:
        if re.search(pattern, prompt_to_scan, re.IGNORECASE):
            return f"Blocked: threat pattern '{pid}'"

    return ""

三类威胁:

1. Prompt override:"ignore previous instructions" 等典型 indirect injection 范式。
2. 系统破坏:rm -rf /、改 sudoers、改 authorized_keys。
3. 数据泄露:curl + 凭证 env var → 外部 URL(典型 token exfiltration)。 但 GitHub API token + api.github.com 是合法 skill 用例,白名单。

这是第 7 章 Capability Security 的纵深——即使工具能调,prompt 本身也得过扫描。

15.3.3 no_agent=True:Watchdog 模式

有些任务根本不需要 LLM——比如"每 5 分钟检查 GPU 温度,>80 度报警"。no_agent=True:

cronjob(
    action="create",
    schedule="every 5m",
    no_agent=True,
    script="gpu_watchdog.sh",         # 路径相对 ~/.hermes/scripts/
    deliver="origin",                # 输出回到我手机
)

"delivery 语义" 写得很清楚:

• stdout 非空 → 原样作为消息发给 origin
• stdout 空 → 完全静默(用户什么也不看见)——典型 watchdog "没事就别说话"模式
• 退出码 ≠ 0 → 发错误警报(防止 watchdog 静默失败)

15.4Batch Runner:并行轨迹生成

批量 prompt → 批量 trajectories(给 SFT 用)。1321 行,用 multiprocessing.Pool:

batch_runner.py (核心)
from multiprocessing import Pool, Lock

_WORKER_CONFIG = {}    # 给 worker 进程的全局配置

def _process_single_prompt(prompt_index, prompt_data, batch_num, config):
    """每个 worker 进程处理一条 prompt"""
    prompt = prompt_data["prompt"]
    task_id = f"task_{prompt_index}"

    # Per-prompt container image override
    container_image = prompt_data.get("image") or prompt_data.get("docker_image")
    if container_image:
        env_type = os.getenv("TERMINAL_ENV", "local")
        if env_type == "docker":
            # 提前验证镜像可用,免得花了 token 再发现镜像不在
            probe = subprocess.run(["docker", "image", "inspect", container_image],
                                   capture_output=True, timeout=10)
            if probe.returncode != 0:
                pull = subprocess.run(["docker", "pull", container_image],
                                      capture_output=True, timeout=600)
                if pull.returncode != 0:
                    return {"success": False, "error": f"Docker image not available: ..."}

关键设计:

1. multiprocessing 而非 threading:每个进程独立 Python GIL,真并行。LLM API call 是 IO bound 但 token 处理是 CPU bound。
2. 每 prompt 可指定 docker image:不同任务可用不同环境(Python 3.11、Node 20、自定义)。Hermes 内部还支持 Modal/Singularity/Daytona 多种 sandbox 后端,通过 register_task_env_overrides() 注入。
3. 提前 docker inspect 验镜像:不在镜像可用前花 token——这是节省费用的关键防御。

15.4.1 Tool Stats 提取

训练数据需要标注"每条轨迹用了哪些工具,各调几次,成功率多少"。Hermes 用 5 步状态机提取:

batch_runner.py:125-205
def _extract_tool_stats(messages) -> Dict[str, Dict[str, int]]:
    tool_stats = {}
    tool_calls_map = {}   # tool_call_id → tool_name

    for msg in messages:
        if msg["role"] == "assistant" and msg.get("tool_calls"):
            for tool_call in msg["tool_calls"]:
                tool_name = tool_call["function"]["name"]
                tool_call_id = tool_call["id"]
                if tool_name not in tool_stats:
                    tool_stats[tool_name] = {"count": 0, "success": 0, "failure": 0}
                tool_stats[tool_name]["count"] += 1
                tool_calls_map[tool_call_id] = tool_name

        elif msg["role"] == "tool":
            # 解析 tool result 判成功/失败
            content_json = json.loads(msg["content"]) if isinstance(msg["content"], str) else ...
            is_success = True
            if isinstance(content_json, dict) and content_json.get("error") is not None:
                is_success = False
            if tool_call_id in tool_calls_map:
                tool_name = tool_calls_map[tool_call_id]
                tool_stats[tool_name]["success" if is_success else "failure"] += 1

    return tool_stats

注意"成功"的判定标准:工具返回 JSON 且 error 字段为 None 才算成功。如果 error: null 或 error 字段不存在都算成功——这是故意宽松的,因为很多工具不显式标 success。

15.4.2 输出格式

{
    "success": True,
    "prompt_index": i,
    "trajectory": [{from, value}, ...],   # ShareGPT 格式
    "tool_stats": {tool_name: {count, success, failure}},
    "reasoning_stats": {turns_with_reasoning, ...},
    "completed": True,                    # 模型自然结束?
    "partial": False,                     # 被中断?
    "api_calls": 12,                       # 模型调用次数
    "toolsets_used": ["web", "file", ...],
    "metadata": {batch_num, timestamp, model}
}

每条 prompt 一个独立 result dict,可以分批写盘、可以恢复。--resume 标志通过对比已有 batch 输出文件和 dataset 决定从哪里继续。

15.5Trajectory Compressor:用 LLM 压 LLM

大 model 跑出来的 trajectory 动辄 30K-100K tokens。但 SFT 训练时,大多 LLM 训练框架要求每条样本 <= 16K tokens。怎么压?

方案:保护头尾,LLM 总结中段。这就是 trajectory_compressor.py。

15.5.1 配置

trajectory_compressor.py:82-179
@dataclass
class CompressionConfig:
    # Tokenizer
    tokenizer_name: str = "moonshotai/Kimi-K2-Thinking"

    # 压缩目标
    target_max_tokens: int = 15250      # 压完上限
    summary_target_tokens: int = 750    # 总结约多长

    # 保护层
    protect_first_system: bool = True   # 首 system
    protect_first_human: bool = True    # 首 user 输入
    protect_first_gpt: bool = True      # 首 assistant 响应
    protect_first_tool: bool = True     # 首 tool 结果
    protect_last_n_turns: int = 4      # 最后 4 轮

    # 总结用 OpenRouter Gemini Flash(便宜)
    summarization_model: str = "google/gemini-3-flash-preview"
    api_key_env: str = "OPENROUTER_API_KEY"
    temperature: float = 0.3

    add_summary_notice: bool = True
    summary_notice_text: str = "\n\nSome of your previous tool responses may be summarized to preserve context."

15.5.2 算法

flowchart TB
  In["输入轨迹
(50 turns, 32K tokens)"] --> Count["数 token 数"]
  Count --> Cmp{"总 token
≤ 15.25K?"}
  Cmp -->|是| Skip["不压"]
  Cmp -->|否| Find["定位可压区间
(头尾保护之间)"]
  Find --> Calc["算需省 tokens
= 总数 - target + summary_size"]
  Calc --> Accum["从前往后累加 turns
直到累计 ≥ 需省"]
  Accum --> Sum["送 Gemini Flash 总结"]
  Sum --> Build["重建:头 + 总结 + 尾"]
  Build --> Out["压缩后轨迹
(~15K tokens)"]

trajectory_compressor.py:709-827 (核心)
def compress_trajectory(self, trajectory):
    metrics = TrajectoryMetrics()
    metrics.original_turns = len(trajectory)

    turn_tokens = self.count_turn_tokens(trajectory)
    total_tokens = sum(turn_tokens)
    metrics.original_tokens = total_tokens

    if total_tokens <= self.config.target_max_tokens:
        metrics.skipped_under_target = True
        return trajectory, metrics

    # 找保护区:头部 4 个 + 尾部 4 个
    protected, compress_start, compress_end = self._find_protected_indices(trajectory)

    # 算要省多少 tokens
    tokens_to_save = total_tokens - self.config.target_max_tokens
    target_tokens_to_compress = tokens_to_save + self.config.summary_target_tokens

    # 从前往后累加,累到够省即止
    accumulated_tokens = 0
    compress_until = compress_start
    for i in range(compress_start, compress_end):
        accumulated_tokens += turn_tokens[i]
        compress_until = i + 1
        if accumulated_tokens >= target_tokens_to_compress:
            break

    # 提取要压的内容
    content_to_summarize = self._extract_turn_content_for_summary(
        trajectory, compress_start, compress_until)

    # 用 LLM 总结(这里是真调 API)
    summary = self._generate_summary(content_to_summarize, metrics)

    # 重建:头 + 总结 + 尾
    compressed = []
    for i in range(compress_start):
        turn = trajectory[i].copy()
        if turn.get("from") == "system" and self.config.add_summary_notice:
            turn["value"] = turn["value"] + self.config.summary_notice_text
        compressed.append(turn)

    compressed.append({"from": "human", "value": summary})

    for i in range(compress_until, len(trajectory)):
        compressed.append(trajectory[i].copy())

    return compressed, metrics

设计精髓:

• 保护头尾,牺牲中段。模型读训练数据时,任务和结论都在两端,中间过程概括就够。
• "先头 + 总结 + 尾"重建。总结被 cast 成 from: "human",作为合成的用户提示插入。
• System message 末尾追加 notice。让模型知道"中段有省略",避免训完模型以为某些信息真的不存在。
• 用便宜 model 压(Gemini Flash)。压缩 pipeline 不该比训练贵。

15.6Mini SWE Runner:Hermes 内置 SWE-Bench

SWE-Bench 评测需要在隔离环境里跑、捕获 patch、跑测试、记 pass/fail。mini_swe_runner.py(735 行)是 Hermes 自己的简化实现。

15.6.1 任务格式

# 单任务模式
python mini_swe_runner.py --task "Create a hello world Python script" --env local

# 批量模式(JSONL)
python mini_swe_runner.py --tasks tasks.jsonl --env docker

# tasks.jsonl 每行:
{"task": "...", "env": "docker", "image": "python:3.11-slim"}

15.6.2 环境抽象

mini_swe_runner.py:121-150
def create_environment(env_type="local", image="python:3.11-slim",
                       cwd="/tmp", timeout=60, **kwargs):
    if env_type == "local":
        from tools.environments.local import LocalEnvironment
        return LocalEnvironment(cwd=cwd, timeout=timeout)
    elif env_type == "docker":
        from tools.environments.docker import DockerEnvironment
        return DockerEnvironment(image=image, cwd=cwd, timeout=timeout, **kwargs)
    elif env_type == "modal":
        from tools.environments.modal import ModalEnvironment
        ...

三种沙箱后端:local(开发用)、docker(评测主力)、modal(serverless)。 每个后端都提供 execute(cmd) + cleanup() 接口——典型的策略模式。

15.6.3 完成标记

怎么知道 Agent 觉得任务完成了?用哨兵字符串:

mini_swe_runner.py:97
# When task is complete, output: echo "MINI_SWE_AGENT_FINAL_OUTPUT" followed by your result

Agent 跑 echo "MINI_SWE_AGENT_FINAL_OUTPUT\n<result>" 时,runner 监听 terminal stdout, 看到这个 token 就知道结束。这种"约定优于实现"的方法比让 Agent 调一个 finish_task 工具简单很多—— 不用 schema,prompt 里说一句就行。

15.7设计原则提炼

5 个子系统看下来,提炼出后台执行的 6 条工程原则:

1. 持久化用最简结构。Cron job 用 JSON + tempfile + atomic rename + fsync + chmod 0600, 不上 SQLite。规模小、可读、可手改、可 grep。
2. 锁但别等。Cron tick 用 LOCK_NB 非阻塞——抢不到立刻让出,下一轮再来。 避免雪崩。
3. Catchup 要有"放弃"机制。Grace window 决定"补"还是"跳"。重启后不该补 1000 次"每 5 分钟"。
4. 超时基于活动,不是 wall-clock。慢 LLM call 是正常的;没有活动才是异常。 get_activity_summary() 是核心 instrumentation。
5. 子任务的 toolset 默认窄。Cron job 不能创建 cron job(防递归),不能 send_message(模糊责任), 不能 clarify(无人答)。能力受限优于约束 prompt(第 7 章主题)。
6. 能在花 token 之前验证的,先验证。Docker pull 失败比花了一百次 LLM 调用才发现镜像没拉成本低 1000 倍。