传奇流水线机器人（pipeline_bot）实施计划

For agentic workers: REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (- [ ]) syntax for tracking.

Goal: 在全新目录 pipeline_bot/ 建立后续工程的基础版本：从 protocol_bot/ 迁移全部运行所需资产，并实现"登录→创角→喝药→比奇20杀→天然洞穴20杀→比奇矿区20杀→蜈蚣洞穴40杀→神通→五行→沃玛挂机"九任务串行流水线，共用一次登录会话，每任务独立文件、完成打勾（控制台✓ + 进度_{区服}_{角色}.md）。

Architecture: 三层结构：① 资产迁移层（cp 保留目录布局，parents[3] 相对定位继续生效）；② 框架增强层（SC52 等级解析进 WorldMirror）；③ 流水线层（pipeline/ 包：ctx.py 会话上下文 + base.py 任务基类 + 9 个任务文件 + run_pipeline.py 总控）。任务文件只编排，战斗/传送逻辑复用已实测的行为类（KillBehavior / CaveStayKillBehavior / AutoFightBehavior / CaveEnterBehavior / MineEnterBehavior）。

Tech Stack: Python 3.12 / uv、pbot 分层框架（codec/login/transport/game/world/bot）、requests。

Spec: 用户需求（2026-10-01 对话）：账号 longwu988 / 密码 Longwu1988，进【玖扬复古176】最新区，创全中文随机名角色；沃玛挂机时长为变量（默认 30 分钟）；打勾文件名 = 游戏区服名称+游戏区名称+游戏角色；等级不足 35 时继续杀怪到 35 再开神通。参考定案文档：protocol_bot/docs/协议定案-神通开启与五行之力.md、protocol_bot/docs/协议定案-盟重传送员三图传送.md。

Global Constraints

新目录名：pipeline_bot/，位于仓库根 /Users/zlwu/work/00code/传奇/pipeline_bot/
protocol_bot/ 与 protocol_bot_backup/ 冻结为探索参考，只读，禁止改动
迁移后 pipeline_bot/ 内代码禁止 import protocol_bot（self-contained）
资产相对定位布局不变：src/pbot/ → parents[3] = 项目根 → 根下须有 assets/、vendor/、Encode6BitBuf_替换表.bin
帐号凭据：-u longwu988 -p Longwu1988（写死在 run_pipeline.py 默认值，CLI 可覆盖）
沃玛挂机分钟数：CLI --woma-minutes，默认 30.0
打勾文件：写到 pipeline_bot/进度_玖扬复古176_{区名}_{角色名}.md（区名 = 服务器列表 list_name；文件名含 / 等非法字符时替换为 _）
每任务文件提供统一签名 def run_task(ctx: PipelineContext) -> TaskResult
运行验证环境：cd pipeline_bot && uv sync && uv run python pipeline/run_pipeline.py --dry-run（不登录服务器冒烟）+ uv run pytest + uv run ruff check .

Task 1: 建立新工程目录并迁移资产

Files:

Create: pipeline_bot/（目录树，见 Step 1）
Create: pipeline_bot/README.md

Interfaces:

Consumes: protocol_bot/ 的 src/ tests/ assets/ vendor/ scripts/、Encode6BitBuf_替换表.bin、pyproject.toml、pytest.ini（若存在，否则用 pyproject 内 pytest 配置）、uv.lock
Produces: 自包含的 pipeline_bot/ 工程，uv run python -c "import pbot" 成功
[ ] Step 1: 拷贝目录与文件

cd /Users/zlwu/work/00code/传奇
mkdir -p pipeline_bot
# 框架源码+测试(排除缓存)
rsync -a --exclude '__pycache__' --exclude '.pytest_cache' --exclude '.ruff_cache' \
    protocol_bot/src protocol_bot/tests pipeline_bot/
# 资产(地图/模板/vendor)
rsync -a --exclude '.DS_Store' protocol_bot/assets protocol_bot/vendor pipeline_bot/
# 编解码表 + 依赖清单 + 锁文件 + pytest 配置
cp protocol_bot/Encode6BitBuf_替换表.bin pipeline_bot/
cp protocol_bot/pyproject.toml pipeline_bot/
cp protocol_bot/uv.lock pipeline_bot/
cp protocol_bot/main.py pipeline_bot/          # 保留单段实验入口(参考)
# 行为库脚本(只迁 run_juece13_* 系列与依赖, 不迁 diag/probe)
mkdir -p pipeline_bot/scripts
cp protocol_bot/scripts/run_juece13_*.py pipeline_bot/scripts/
cp protocol_bot/scripts/enable_heal.py pipeline_bot/scripts/ 2>/dev/null || true

注意：protocol_bot/pytest.ini 若存在也要 cp（上面 ls 显示根目录有 pytest.ini，检查 protocol_bot/ 内是否有；pyproject [tool.pytest.ini_options] 已含配置，无 pytest.ini 也能跑）。

Step 2: pyproject 改名

pipeline_bot/pyproject.toml 中 name = "pbot" 保持不变（包名仍是 pbot，目录名与包名解耦）。仅在 description 追加说明：传奇流水线机器人 — 新手九任务串行(基础版)。

Step 3: 验证自包含性

cd pipeline_bot
uv sync
uv run python -c "import pbot; from pbot.bot.kill import KillBehavior; from pbot.login.geeked_solver import solve_slide; print('ok')"
uv run pytest -q
uv run ruff check src tests

Expected: import ok；pytest 全绿（36+ 用例）；ruff 无错。

Step 4: 写 README.md

内容要点（简洁，30 行内）：目录结构、uv run python pipeline/run_pipeline.py -u longwu988 -p Longwu1988 --woma-minutes 30 用法、打勾文件说明、"探索参考见 ../protocol_bot 与 ../protocol_bot_backup"。

Step 5: Commit

git add pipeline_bot
git commit -m "feat(pipeline_bot): 从 protocol_bot 迁移建立基础工程"

Task 2: SC52 等级解析进 WorldMirror

Files:

Modify: pipeline_bot/src/pbot/world/mirror.py:17-41（PlayerState 增字段）
Modify: pipeline_bot/src/pbot/world/update.py:167-229（apply 增 SC52 分支）
Test: pipeline_bot/tests/test_pipeline_units.py（新建）

Interfaces:

Consumes: SC52 帧 body=encode6(JSON)，字段 "level": int（2026-10-01 用 cap_hoe_success_frames_20261001.jsonl 实测验证：{"job":0,...,"level":1,"hp":15,...}）
Produces: WorldMirror.me.level: int（默认 0，SC52 到达后更新）；WorldUpdater.apply 对 msg_id==52 调 _on_player_props
[ ] Step 1: 写失败测试

pipeline_bot/tests/test_pipeline_units.py：

"""流水线新增单元测试: SC52 等级解析 / 名字生成 / 打勾文件名清洗。"""
from pbot.codec.bitbuf import encode6
from pbot.codec.frame import build_frame
from pbot.world.mirror import WorldMirror
from pbot.world.update import WorldUpdater


def _sc52_frame(level: int, hp: int = 100):
    import json
    body = json.dumps({"job": 0, "level": level, "hp": hp,
                       "maxhp": hp, "exp": 0, "maxexp": 100}).encode()
    raw = build_frame(52, 0, 0, 0, 0, encode6(body))
    return raw


def test_sc52_updates_level():
    w = WorldMirror()
    u = WorldUpdater()
    from pbot.codec.frame import parse_frame
    u.apply(w, parse_frame(_sc52_frame(35)))
    assert w.me.level == 35
    u.apply(w, parse_frame(_sc52_frame(36, hp=120)))
    assert w.me.level == 36
    assert w.me.hp == 120  # maxhp/hp 同步(若实现)或仅 level(见下)


def test_sc52_level_zero_ignored():
    w = WorldMirror()
    u = WorldUpdater()
    from pbot.codec.frame import parse_frame
    u.apply(w, parse_frame(_sc52_frame(0)))
    assert w.me.level == 0  # 0 不产生误报

Step 2: 运行确认失败

cd pipeline_bot && uv run pytest tests/test_pipeline_units.py -v

Expected: FAIL — AttributeError: 'PlayerState' object has no attribute 'level'。

Step 3: 实现

mirror.py PlayerState 增：

    level: int = 0  # SC52 属性帧 "level"

update.py apply 链增（放在 elif mid == 53: 之前）：

        elif mid == 52:
            self._on_player_props(world, frame)

新增方法（放在 _on_hp 附近）：

    def _on_player_props(self, world: WorldMirror, f: Frame) -> None:
        """SC52 角色属性: level 权威源(神通 35 级门槛判定用)。

        body=encode6(JSON), 实测(cap_hoe_success_frames_20261001):
        {"job":0,"level":1,"hp":15,"maxhp":29,"exp":0,...}
        hp/maxhp 有值时同步(比 SC53 稀疏推送更全)。
        """
        j = body_json(f)
        lv = _as_int(j.get("level", 0))
        if lv > 0:
            if lv != world.me.level:
                _LOG.info("[SC52] 等级 %d → %d", world.me.level, lv)
            world.me.level = lv
        hp = _as_int(j.get("hp", 0))
        maxhp = _as_int(j.get("maxhp", 0))
        if hp > 0:
            world.me.hp = hp
        if maxhp > 0:
            world.me.max_hp = maxhp
        world.record("player_props", f"lv={lv}")

注意核对 _as_int 在 update.py 的既有名字（grep def _as_int，若叫别的名以现名为准）；body_json 已存在。

Step 4: 运行确认通过 + 全量回归

uv run pytest tests/test_pipeline_units.py -v && uv run pytest -q

Step 5: Commit

git add src/pbot/world tests/test_pipeline_units.py
git commit -m "feat(world): SC52 属性帧解析等级进镜像"

Task 3: pipeline 包骨架 — PipelineContext / TaskResult / ChecklistWriter / 名字生成

Files:

Create: pipeline_bot/pipeline/__init__.py
Create: pipeline_bot/pipeline/ctx.py
Create: pipeline_bot/pipeline/base.py
Create: pipeline_bot/pipeline/naming.py
Test: pipeline_bot/tests/test_pipeline_units.py（追加）

Interfaces:

Produces（后续所有任务依赖，签名逐字准确）:
- @dataclass class PipelineContext: 字段 user: str、password: str、zone: ZoneId、zone_label: str（如"玖扬复古176.38区"或服务器 list_name）、role_name: str、woma_minutes: float、conn: GatewayConnection、world: WorldMirror、updater: WorldUpdater、sid: int、stop: list[bool]、checklist: ChecklistWriter、http: requests.Session、jwt: str、uid: int、log: logging.Logger；方法 pump(window: float = 0.5) -> None（转发 pbot.bot.runner.pump）
- @dataclass class TaskResult: 字段 ok: bool、detail: str = ""；静态构造 TaskResult.done(detail) / TaskResult.fail(detail)
- class ChecklistWriter: __init__(self, path: Path, zone_label: str, role: str)；mark(title: str, ok: bool, detail: str = "") -> None（实时写：☐→☑/☒ + 时间戳 + detail 行）；note(text: str) -> None；内部首写 header（# 传奇流水线进度 — {zone_label} — {role} + 任务总表）
- def random_cn_name() -> str（naming.py）：纯中文角色名，格式 = 随机姓(约50个) + 随机名(1-2字，池约120字)，确保 all('\u4e00' <= c <= '\u9fff' for c in name)
- def safe_filename(s: str) -> str（naming.py）：re.sub(r'[\\/:*?\"<>|\s]+', '_', s).strip('_') or 'unnamed'
[ ] Step 1: 写失败测试（追加到 test_pipeline_units.py）

# ---- naming ----
from pipeline.naming import random_cn_name, safe_filename


def test_random_cn_name_all_chinese():
    for _ in range(50):
        n = random_cn_name()
        assert 2 <= len(n) <= 3
        assert all('\u4e00' <= c <= '\u9fff' for c in n)


def test_safe_filename():
    assert safe_filename("玖扬复古176.38区/新角色") == "玖扬复古176.38区_新角色"
    assert safe_filename("a\\b:*?\"<>|") == "a_b"
    assert safe_filename("   ") == "unnamed"


# ---- checklist ----
from pathlib import Path
from pipeline.ctx import ChecklistWriter


def test_checklist_writer_marks(tmp_path: Path):
    w = ChecklistWriter(tmp_path / "进度.md", "玖扬复古176.38区", "凌风")
    w.mark("T1 喝药保护", True, "state=on")
    w.mark("T2 比奇杀怪", False, "超时")
    txt = (tmp_path / "进度.md").read_text(encoding="utf-8")
    assert "☑" in txt and "☒" in txt and "凌风" in txt
    # 二次 mark 同名任务 = 更新而非重复追加
    w.mark("T2 比奇杀怪", True, "20/20")
    txt = (tmp_path / "进度.md").read_text(encoding="utf-8")
    assert txt.count("T2 比奇杀怪") == 1

Step 2: 运行确认失败

uv run pytest tests/test_pipeline_units.py -v -k "naming or checklist"

Expected: FAIL — ModuleNotFoundError: No module named 'pipeline'。

Step 3: 实现

pipeline/__init__.py：

"""pipeline — 新手九任务串行流水线(共用一次登录会话)。"""

pipeline/naming.py：

"""随机中文角色名 + 文件名清洗。"""
import random
import re

_SURNAMES: tuple[str, ...] = (
    "赵钱孙李周吴郑王冯陈褚卫蒋沈韩杨朱秦许何吕施张孔曹严华金魏陶姜",
    "谢邹柏水窦章云苏潘葛范彭郎鲁韦昌马苗凤花方俞任袁柳鲍史唐",
    "萧叶阎余潘杜夏宋钟汪田任姜范方石姚谭廖邹郝孔崔秦邱侯邵龚",
    "林卓唐温贾雷贺汤罗毕安于时傅皮卞齐康伍余元卜顾孟平黄穆萧尹",
    "欧阳司马上官诸葛夏侯皇甫尉迟公羊澹台公冶宗政濮阳淳于单于",
)
_GIVEN_CHARS: tuple[str, ...] = (
    "风云雷电霜雪星月山河江海林泉石松竹梅兰菊荷峰岳城铁剑刀"
    "狂傲凌飞跃震天啸苍穹宇辰晓旭尧舜禹轩辕擎苍痕无极道玄清真"
    "长庚启明北斗天涯孤鸿惊鸿流云追风逐月听雨眠霜寒影夜白轻语"
    "凌波青丝白首丹心碧血铁骨侠客行止言笑怒悲欢离合恩怨情仇录"
    "一乙二十丁厂七卜人入八九几儿了力乃刀又三于干亏士工土才寸"
    " warfare测试"  # 占位防呆 — 见下行剔除
)
_GIVEN = tuple(c for c in _GIVEN_CHARS if "\u4e00" <= c <= "\u9fff")


def random_cn_name() -> str:
    """纯中文随机名: 姓(1-2字) + 名(1-2字), 全字 CJK 检查兜底。"""
    surname = random.choice(_SURNAMES)  # noqa: S311
    # 复姓保持整姓, 单姓池取首字
    surname = surname if len(surname) == 2 else surname[0] \
        if random.random() < 0.5 and len(surname) > 1 else random.choice(surname)  # noqa: S311
    given_len = random.choice((1, 2))  # noqa: S311
    given = "".join(random.choice(_GIVEN) for _ in range(given_len))  # noqa: S311
    name = surname[:2] + given
    name = "".join(c for c in name if "\u4e00" <= c <= "\u9fff")
    return name[:4]


def safe_filename(s: str) -> str:
    """路径非法字符 → 下划线。"""
    return re.sub(r'[\\/:*?\"<>|\s]+', "_", s).strip("_") or "unnamed"

（注意：上面 _GIVEN_CHARS 末行占位串会被 CJK 过滤自动剔除，实现时直接删掉该行与注释，保留前四行中文池即可。）

pipeline/ctx.py：

"""PipelineContext — 一次登录会话贯穿九任务; TaskResult 与打勾落盘。"""
from __future__ import annotations

import logging
import time
from dataclasses import dataclass, field
from pathlib import Path
from typing import Final

import requests

from pbot.bot.runner import pump as _pump
from pbot.transport.gateway import GatewayConnection
from pbot.types import ZoneId
from pbot.world.mirror import WorldMirror
from pbot.world.update import WorldUpdater

_FMT: Final[str] = "%Y-%m-%d %H:%M:%S"


@dataclass
class TaskResult:
    """单任务结果。"""
    ok: bool
    detail: str = ""

    @staticmethod
    def done(detail: str = "") -> "TaskResult":
        return TaskResult(ok=True, detail=detail)

    @staticmethod
    def fail(detail: str) -> "TaskResult":
        return TaskResult(ok=False, detail=detail)


class ChecklistWriter:
    """打勾文件实时写: 每任务一行, 完成即勾, 失败即叉, 支持更新。"""

    def __init__(self, path: Path, zone_label: str, role: str) -> None:
        self._path = path
        self._role = role
        self._marks: dict[str, tuple[bool, str]] = {}
        path.parent.mkdir(parents=True, exist_ok=True)
        with path.open("w", encoding="utf-8") as f:
            f.write(f"# 传奇流水线进度 — {zone_label} — {role}\n\n")
            f.write(f"- 启动: {time.strftime(_FMT)}\n\n")
            f.write("## 任务清单\n\n")

    def note(self, text: str) -> None:
        with self._path.open("a", encoding="utf-8") as f:
            f.write(f"> {time.strftime(_FMT)} {text}\n")

    def mark(self, title: str, ok: bool, detail: str = "") -> None:
        self._marks[title] = (ok, detail)
        lines = ["## 任务清单\n\n"]
        for t, (o, d) in self._marks.items():
            mark = "☑" if o else "☒"
            lines.append(f"- {time.strftime(_FMT)} {mark} **{t}**"
                         + (f" — {d}" if d else "") + "\n")
        tmp = self._path.with_suffix(".tmp")
        tmp.write_text("".join(lines), encoding="utf-8")
        _ = tmp.replace(self._path)
        mark = "✓" if ok else "✗"
        print(f"[{mark}] {title}" + (f" — {detail}" if detail else ""),
              flush=True)


@dataclass
class PipelineContext:
    """共用会话上下文: 登录一次, 九任务复用 conn/world/sid。"""
    user: str
    password: str
    zone: ZoneId
    zone_label: str
    role_name: str
    woma_minutes: float
    conn: GatewayConnection
    world: WorldMirror
    updater: WorldUpdater
    sid: int
    stop: list[bool]
    checklist: ChecklistWriter
    http: requests.Session = field(default_factory=requests.Session)
    jwt: str = ""
    uid: int = 0
    log: logging.Logger = field(default_factory=lambda: logging.getLogger("pipeline"))

    def pump(self, window: float = 0.5) -> None:
        """收帧窗口(转发 runner.pump)。"""
        _ = _pump(self.conn, self.world, self.updater, window)

pipeline/base.py：

"""任务基协议: 每任务文件暴露 run_task(ctx) -> TaskResult。"""
from pipeline.ctx import PipelineContext, TaskResult

__all__ = ["PipelineContext", "TaskResult", "run_task", "TASK_TITLE"]

TASK_TITLE: str = "(未命名任务)"


def run_task(ctx: PipelineContext) -> TaskResult:  # pragma: no cover - 接口文档
    raise NotImplementedError("任务文件必须实现 run_task(ctx)")

（base.py 仅作接口文档；各任务文件自带 TASK_TITLE 常量与 run_task 函数，总控用 getattr(mod, "TASK_TITLE") / mod.run_task(ctx) 调用。）

Step 4: 运行确认通过

uv run pytest tests/test_pipeline_units.py -v

注意：pyproject [tool.hatch.build.targets.wheel] packages = ["src/pbot"]，pipeline/ 在根目录可直接 import（uv run 时 cwd 在 sys.path）。若 pytest 报 No module named pipeline，在 pyproject.toml [tool.pytest.ini_options] 加 pythonpath = ["."]。

Step 5: Commit

git add pipeline tests/test_pipeline_units.py pyproject.toml
git commit -m "feat(pipeline): 会话上下文/任务结果/打勾落盘/中文名生成"

Task 4: 登录编排器 — bootstrap_session()

Files:

Create: pipeline_bot/pipeline/bootstrap.py
Test: pipeline_bot/tests/test_pipeline_units.py（追加，仅测纯函数部分）

Interfaces:

Consumes: pbot.login.box_client.login/enter_game_client、pbot.transport.gateway.login_gate_with_retry、pbot.game.session.enter_world、main.py 的 pick_latest_zone 逻辑（复制进 bootstrap，不 import protocol_bot）
Produces: def bootstrap_session(user: str, password: str, woma_minutes: float, login_port: int = 0, force_zone: int = 0, role_name: str = "") -> tuple[PipelineContext, str]（返回 ctx 与实际角色名 real_name；内部完成选区→登录→创角→进世界→返回上下文；不开启喝药——那是 T1 任务）

实现要点（照抄 main.py 已验证链路）：

zone = ZoneId(pick_latest_zone())（pick_latest_zone 从 main.py 复制，含 SERVERLIST_URL 常量）；
role_name = role_name or random_cn_name()，create_role_needed(RoleSpec(name=role_name)) 规范化；
login(Account(...)) → jwt, uid；enter_game_client 重试 4 次（间隔 12s）；
login_gate_with_retry(zone, jwt, uid, role_name, ports) → key32, sid, real_name, ticket_key, game_port；若角色不存在会自动创角（读 gateway.py 确认：login_gate 内部处理创角，login_gate_with_retry 只是端口轮询外壳）——若实际是"已存在列表选择"，按 gateway.py 真实语义调用（执行者须读 login_gate_with_retry 源码确认，把"角色不存在→创角"逻辑写在 bootstrap 内，参考 main.py _enter_game 阶段 3 的调用方式）；
enter_world(zone, real_name, ticket_key, sid, game_port, uid, stop=stop) → conn/world/updater；
zone_label：登录后从服务器列表 best 元组取 list_name（pick_latest_zone 返回 int，需改造为内部函数 _latest_zone() -> tuple[int, str] 同时返回 id 与 list_name；label = 玖扬复古176·{list_name}）；
ChecklistWriter 路径 = 项目根 / f"进度玖扬复古176{safe_filename(listname)}{safe_filename(real_name)}.md"；
组装 PipelineContext 返回。

Step 1: 写失败测试（纯函数部分）

追加到 test_pipeline_units.py：

from pipeline.bootstrap import _latest_zone_label


def test_latest_zone_label_format(monkeypatch):
    import pipeline.bootstrap as bs

    def fake_walk(o, best=None):
        return (1750000000.0, "38区", 14594068)

    monkeypatch.setattr(bs, "_pick_latest_zone_tuple", fake_walk)
    assert _latest_zone_label() == ("14594068", "玖扬复古176·38区")

（实现时按实际拆分调整——测试意图：label 拼接格式正确。若实现签名不同，以"测 label 拼接纯函数"为准改造测试，不必逐字对齐。）

Step 2: 运行确认失败 → Step 3: 实现 → Step 4: 确认通过

实现主体（骨架，执行者按 gateway 真实签名补全）：

"""bootstrap — 登录→选区→创角→进世界, 返回共用会话上下文。"""
from __future__ import annotations

import json
import logging
import re
import time
from pathlib import Path

import requests

from pbot.config import Account, RoleSpec
from pbot.game.session import enter_world
from pbot.login.box_client import enter_game_client, login
from pbot.transport.gateway import (
    create_role_needed,
    login_gate_with_retry,
)
from pbot.types import ZoneId
from pbot.world.update import to_float, to_int, typed_json, as_dict

from pipeline.ctx import ChecklistWriter, PipelineContext
from pipeline.naming import random_cn_name, safe_filename

SERVERLIST_URL = "https://list-box.dhsf.xqhuyu.com/box_serverlist/206923.txt"
DEFAULT_PORTS: tuple[int, ...] = (8021, 8061, 8001, 8041)
_ROOT = Path(__file__).resolve().parents[1]

_LOG = logging.getLogger("pipeline.bootstrap")


def _pick_latest_zone_tuple() -> tuple[float, str, int]:
    """(open_time, list_name, serverid) — state=1 且已开服的最新区。"""
    s = requests.Session()
    s.trust_env = False
    s.verify = False
    r = s.get(SERVERLIST_URL, timeout=20)
    m = re.search(r"\{.*\}", r.text, re.DOTALL)
    if not m:
        raise RuntimeError("区列表拉取失败")
    data = typed_json(json.loads(m.group(0)))  # pyright: ignore[reportAny]
    best: list[tuple[float, str, int]] = []

    def walk(o: object) -> None:
        if isinstance(o, dict):
            d = as_dict(typed_json(o))
            sid = to_int(d.get("serverid"))
            state = to_int(d.get("state"))
            ot = to_float(d.get("open_time"))
            if sid > 0 and state == 1 and 0 < ot <= time.time():
                name = str(d.get("list_name") or d.get("server_name") or "")
                best.append((ot, name, sid))
            for v in d.values():
                walk(v)
        elif isinstance(o, list):
            for v in list(o):
                walk(v)

    walk(data)
    if not best:
        raise RuntimeError("无已开新区")
    return max(best)


def _latest_zone_label() -> tuple[str, str]:
    """(serverid 字符串, '玖扬复古176·N区')。"""
    _, name, sid = _pick_latest_zone_tuple()
    return str(sid), f"玖扬复古176·{name}"


def bootstrap_session(user: str, password: str, woma_minutes: float,
                      login_port: int = 0, force_zone: int = 0,
                      role_name: str = "",
                      ) -> tuple[PipelineContext, str]:
    """登录一次 → 创角 → 进世界; 喝药在 T1 任务开。"""
    _, label = _latest_zone_label() if force_zone == 0 else ("", f"区{force_zone}")
    if force_zone:
        zone = ZoneId(force_zone)
        label = f"玖扬复古176·区{force_zone}"
    else:
        _, name, sid = _pick_latest_zone_tuple()
        zone = ZoneId(sid)
        label = f"玖扬复古176·{name}"
    role = role_name or random_cn_name()
    role = create_role_needed(RoleSpec(name=role))
    _LOG.info("区=%s(%s) 角色=%s", label, zone, role)

    http = requests.Session()
    http.trust_env = False
    http.verify = False
    jwt, uid = login(Account(username=user, password=password))
    for attempt in range(4):
        try:
            _ = enter_game_client(http, jwt, zone)
            break
        except Exception:
            if attempt == 3:
                raise
            time.sleep(12)
    ports = (login_port,) if login_port else DEFAULT_PORTS
    _, sid_game, real_name, ticket_key, game_port = login_gate_with_retry(
        zone, jwt, uid, role, ports)
    stop: list[bool] = [False]
    conn, world, updater = enter_world(zone, real_name, ticket_key,
                                       sid_game, game_port, uid, stop=stop)
    checklist_path = _ROOT / (
        f"进度_玖扬复古176_{safe_filename(label.split('·', 1)[1])}"
        f"_{safe_filename(real_name)}.md")
    ctx = PipelineContext(
        user=user, password=password, zone=zone, zone_label=label,
        role_name=real_name, woma_minutes=woma_minutes,
        conn=conn, world=world, updater=updater, sid=sid_game, stop=stop,
        checklist=ChecklistWriter(checklist_path, label, real_name),
        http=http, jwt=jwt, uid=uid)
    return ctx, real_name

（执行者注意：①login_gate_with_retry 返回五元组顺序以 gateway.py:326 签名为准；②label.split('·',1)[1] 防御 force_zone 分支无"·"时直接用 label；③清理上面 _latest_zone_label 双重调用冗余——收敛为一个 _resolve_zone() 内部函数。）

Step 5: 静态验证 + Commit

uv run ruff check pipeline
uv run basedpyright pipeline 2>/dev/null || true   # 脚本域宽松, 不阻塞
git add pipeline/bootstrap.py tests/test_pipeline_units.py
git commit -m "feat(pipeline): 登录编排器 bootstrap_session"

Task 5: T1 喝药保护任务

Files:

Create: pipeline_bot/pipeline/tasks/t01_heal.py

Interfaces:

Consumes: pbot.game.heal.enable_auto_heal(conn, sid, want_on=True, world=world) -> 'on'|'off'|None
Produces: TASK_TITLE = "T1 开启喝药保护"；run_task(ctx) -> TaskResult（ok 当且仅当 state=="on"；重试 2 轮，高危图后续任务都依赖它）

"""T1 — 开启喝药保护(后续所有战斗任务的前置)。"""
from __future__ import annotations

import time

from pipeline.ctx import PipelineContext, TaskResult

TASK_TITLE = "T1 开启喝药保护"


def run_task(ctx: PipelineContext) -> TaskResult:
    from pbot.game.heal import enable_auto_heal
    state = None
    for i in range(3):
        state = enable_auto_heal(ctx.conn, ctx.sid, want_on=True,
                                 world=ctx.world)
        if state == "on":
            break
        ctx.log.warning("[T1] 第%d轮 state=%s — 重试", i + 1, state)
        time.sleep(1.0)
    if state == "on":
        return TaskResult.done("state=on")
    return TaskResult.fail(f"喝药终态={state}(高危, 拒绝继续)")

Step 1: 创建文件（如上全文）
Step 2: 冒烟 uv run python -c "from pipeline.tasks import t01_heal; print(t01_heal.TASK_TITLE)"
Step 3: Commit git add pipeline/tasks && git commit -m "feat(pipeline): T1 喝药保护任务"

Task 6: T2 比奇大城杀 20（含点我刷怪）

Files:

Create: pipeline_bot/pipeline/tasks/__init__.py（空包 + docstring）
Create: pipeline_bot/pipeline/tasks/t02_biqi_kill20.py

Interfaces:

Consumes: pbot.bot.kill.KillBehavior(sid, kill_target, renwu_act="act_move,1", allow_reroll=True)；world.kills 是累计全局计数（kill.py:1317 world.kills += 1）→ 目标线传 baseline + 20；pbot.bot.runner.run(conn, world, updater, behavior, duration_s)
Produces: TASK_TITLE = "T2 比奇大城杀怪20"；run_task(ctx) — 出生任意图 → renwu 任务链传送比奇 → 杀满增量 20 只（KillBehavior 内建：无怪跑图 + 点【点我刷怪】NPC + 死亡复活重走）

"""T2 — 新手接待→比奇大城寻图杀怪 20(怪不够点【点我刷怪】NPC)。"""
from __future__ import annotations

from pipeline.ctx import PipelineContext, TaskResult

TASK_TITLE = "T2 比奇大城杀怪20"

BIQI_MAP = "比奇大城"
KILL_TARGET = 20


def run_task(ctx: PipelineContext) -> TaskResult:
    import os
    from pbot.bot.kill import KillBehavior
    from pbot.bot.runner import run

    baseline = ctx.world.kills
    goal = baseline + KILL_TARGET
    os.environ.setdefault("PBOT_RENWU_ACT", "act_move,1")
    bh = KillBehavior(sid=ctx.sid, kill_target=goal, renwu_act="act_move,1",
                      allow_reroll=True)
    result = run(ctx.conn, ctx.world, ctx.updater, bh,
                 duration_s=max(900.0, KILL_TARGET * 60.0))
    killed = ctx.world.kills - baseline
    ok = killed >= KILL_TARGET
    ctx.log.info("[T2] result=%s 本段击杀=%d/%d map=%s",
                 result.value, killed, KILL_TARGET, ctx.world.me.map_name)
    if ok:
        return TaskResult.done(f"击杀 {killed}/{KILL_TARGET} @比奇大城")
    return TaskResult.fail(f"仅 {killed}/{KILL_TARGET}(result={result.value})")

（执行者核对：KillBehavior.__init__ 参数名 renwu_act/allow_reroll 与 kill.py:244 一致；renwu_act 对新手指引任务二应是 "act_move,1"（比奇）——参照 run_juece13_kill30.py 的用法：其用 "renwu_mov,1"，以 run_juece13_kill30.py 实际传参为准，若该脚本用 renwu_mov,1 则照抄。）

Step 1: 读 run_juece13_kill30.py 全文核对 renwu_act 传参 → 按上文创建文件（以核对结果修正 renwu_act）
Step 2: 冒烟 import + Commit

uv run python -c "from pipeline.tasks import t02_biqi_kill20; print(t02_biqi_kill20.TASK_TITLE)"
git add pipeline/tasks && git commit -m "feat(pipeline): T2 比奇杀20"

Task 7: T3 天然洞穴杀 20

Files:

Create: pipeline_bot/pipeline/tasks/t03_cave_kill20.py

Interfaces:

Consumes: scripts.run_juece13_cave30_v2.CaveStayKillBehavior（sys.path 加 pipeline_bot/scripts 后 import）；CAVE_MAP="天然洞穴一层"
Produces: TASK_TITLE = "T3 天然洞穴杀怪20"；进洞铁律 allow_reroll=False 由行为类内部处理；计数闸门只认洞内击杀 → baseline 用 ctx.world.kills

"""T3 — 任务指引→天然洞穴寻图杀怪 20(驻留洞穴, 不点任务跳转)。"""
from __future__ import annotations

import sys
from pathlib import Path

from pipeline.ctx import PipelineContext, TaskResult

TASK_TITLE = "T3 天然洞穴杀怪20"

KILL_TARGET = 20
_SCRIPTS = str(Path(__file__).resolve().parents[2] / "scripts")


def run_task(ctx: PipelineContext) -> TaskResult:
    if _SCRIPTS not in sys.path:
        sys.path.insert(0, _SCRIPTS)
    from run_juece13_cave30_v2 import CaveStayKillBehavior
    from pbot.bot.runner import run

    baseline = ctx.world.kills
    goal = baseline + KILL_TARGET
    bh = CaveStayKillBehavior(sid=ctx.sid, kill_target=goal,
                              renwu_act="renwu_mov,2", allow_reroll=True)
    result = run(ctx.conn, ctx.world, ctx.updater, bh,
                 duration_s=max(1500.0, KILL_TARGET * 75.0))
    killed = ctx.world.kills - baseline
    ok = killed >= KILL_TARGET and ctx.world.me.map_name == "天然洞穴一层"
    ctx.log.info("[T3] result=%s 击杀=%d/%d map=%s", result.value, killed,
                 KILL_TARGET, ctx.world.me.map_name)
    if ok:
        return TaskResult.done(f"洞穴击杀 {killed}/{KILL_TARGET}")
    return TaskResult.fail(f"{killed}/{KILL_TARGET} map={ctx.world.me.map_name}")

（执行者核对 CaveStayKillBehavior.init 是否接受 kill_target/renwu_act/allow_reroll —— 它继承 KillBehavior，签名一致。）

Step 1: 创建文件 → Step 2: 冒烟 import → Step 3: Commit feat(pipeline): T3 天然洞穴杀20

Task 8: T4 比奇矿区杀 20

Files:

Create: pipeline_bot/pipeline/tasks/t04_mine_kill20.py

Interfaces:

Consumes: scripts.run_juece13_mine_farthest.MineEnterBehavior（盟重传送员→比奇矿区）；scripts.run_juece13_wugong_autofight.AutoFightBehavior(map_name="比奇矿区", walls=_d401_walls(), farthest_fn=find_farthest, kill_goal=N)；AutoFight 死亡回城 DONE 退出 + kills 实例级 → 外层轮次循环累计
Produces: TASK_TITLE = "T4 比奇矿区杀怪20"

"""T4 — 任务指引→比奇矿区寻图杀怪 20(传送员链 + 自动战斗轮次)。"""
from __future__ import annotations

import time

from pipeline.ctx import PipelineContext, TaskResult

TASK_TITLE = "T4 比奇矿区杀怪20"

KILL_TARGET = 20
MINE_MAP = "比奇矿区"
_MAX_ROUNDS = 6


def run_task(ctx: PipelineContext) -> TaskResult:
    import sys
    from pathlib import Path
    scripts = str(Path(__file__).resolve().parents[2] / "scripts")
    if scripts not in sys.path:
        sys.path.insert(0, scripts)
    from run_juece13_mine_farthest import (
        MINE_MAP as _MM, MineEnterBehavior, _d401_walls, find_farthest,
    )
    from run_juece13_wugong_autofight import AutoFightBehavior
    from pbot.bot.runner import pump, run

    _ = _MM
    total = 0
    for rnd in range(1, _MAX_ROUNDS + 1):
        if ctx.world.me.map_name != MINE_MAP:
            ctx.log.info("[T4] 第%d轮 进图(当前 %s)", rnd,
                         ctx.world.me.map_name)
            bh = MineEnterBehavior(sid=ctx.sid, kill_target=0)
            r = run(ctx.conn, ctx.world, ctx.updater, bh, duration_s=300.0)
            if ctx.world.me.map_name != MINE_MAP:
                ctx.log.warning("[T4] 传送失败 result=%s — 重试", r.value)
                time.sleep(3.0)
                continue
            time.sleep(3.0)
            _ = pump(ctx.conn, ctx.world, ctx.updater, 2.0)
        need = KILL_TARGET - total
        af = AutoFightBehavior(sid=ctx.sid, map_name=MINE_MAP,
                               walls=_d401_walls(), farthest_fn=find_farthest,
                               kill_goal=need)
        r = run(ctx.conn, ctx.world, ctx.updater, af,
                duration_s=max(600.0, need * 60.0))
        total += af.kills
        ctx.log.info("[T4] 第%d轮 result=%s 轮杀=%d 累计=%d/%d", rnd,
                     r.value, af.kills, total, KILL_TARGET)
        if total >= KILL_TARGET:
            return TaskResult.done(f"矿区击杀 {total}/{KILL_TARGET}")
    return TaskResult.fail(f"{total}/{KILL_TARGET} 轮次耗尽")

Step 1: 创建文件 → Step 2: 冒烟 import → Step 3: Commit feat(pipeline): T4 矿区杀20

Task 9: T5 蜈蚣洞穴杀 40（等级≥35 收尾）

Files:

Create: pipeline_bot/pipeline/tasks/t05_wugong_kill40.py

Interfaces:

Consumes: scripts.run_juece13_wugong_farthest.CaveEnterBehavior + 模块常量 CAVE_MAP/_ACK_BODY/_TELEPORT_ACTS（默认即蜈蚣参数，无需覆写）；AutoFightBehavior；ctx.world.me.level（Task 2 产出）
Produces: TASK_TITLE = "T5 蜈蚣洞穴杀怪40+至35级"

"""T5 — 任务指引→蜈蚣洞穴杀怪 40; 若等级<35 继续杀到 35(用户决策)。"""
from __future__ import annotations

import time

from pipeline.ctx import PipelineContext, TaskResult

TASK_TITLE = "T5 蜈蚣洞穴杀怪40+至35级"

KILL_TARGET = 40
LEVEL_GOAL = 35
WUGONG_MAP = "地牢一层东"
_MAX_ROUNDS = 8


def run_task(ctx: PipelineContext) -> TaskResult:
    import sys
    from pathlib import Path
    scripts = str(Path(__file__).resolve().parents[2] / "scripts")
    if scripts not in sys.path:
        sys.path.insert(0, scripts)
    from run_juece13_wugong_autofight import AutoFightBehavior
    from run_juece13_wugong_farthest import (  # noqa: F401 — 引用即生效
        _cave_walls, find_farthest,
    )
    from run_juece13_wugong_farthest import CaveEnterBehavior
    from pbot.bot.runner import pump, run

    total = 0
    for rnd in range(1, _MAX_ROUNDS + 1):
        if ctx.world.me.map_name != WUGONG_MAP:
            ctx.log.info("[T5] 第%d轮 进图(当前 %s)", rnd,
                         ctx.world.me.map_name)
            bh = CaveEnterBehavior(sid=ctx.sid, kill_target=0)
            r = run(ctx.conn, ctx.world, ctx.updater, bh, duration_s=300.0)
            if ctx.world.me.map_name != WUGONG_MAP:
                ctx.log.warning("[T5] 传送失败 result=%s", r.value)
                time.sleep(3.0)
                continue
            time.sleep(3.0)
            _ = pump(ctx.conn, ctx.world, ctx.updater, 2.0)
        need = KILL_TARGET - total
        lv_goal_met = ctx.world.me.level >= LEVEL_GOAL
        af = AutoFightBehavior(sid=ctx.sid, map_name=WUGONG_MAP,
                               walls=_cave_walls(), farthest_fn=find_farthest,
                               kill_goal=need)
        r = run(ctx.conn, ctx.world, ctx.updater, af,
                duration_s=max(600.0, need * 60.0))
        total += af.kills
        ctx.log.info("[T5] 第%d轮 result=%s 轮杀=%d 累计=%d/%d lv=%d", rnd,
                     r.value, af.kills, total, KILL_TARGET,
                     ctx.world.me.level)
        if total >= KILL_TARGET and ctx.world.me.level >= LEVEL_GOAL:
            return TaskResult.done(
                f"击杀 {total}/{KILL_TARGET} 等级 {ctx.world.me.level}")
        if total >= KILL_TARGET and not lv_goal_met:
            continue  # 杀够了但级不够: 继续轮次杀(need 会变 0 → kill_goal=0=∞)
        if total >= KILL_TARGET:
            return TaskResult.done(
                f"击杀 {total}/{KILL_TARGET} 等级 {ctx.world.me.level}")
    lv = ctx.world.me.level
    if total >= KILL_TARGET:
        return TaskResult.fail(
            f"杀满 {total} 但等级 {lv}<35 — 神通任务将跳过等级检查再试")
    return TaskResult.fail(f"{total}/{KILL_TARGET} 轮次耗尽 lv={lv}")

（kill_goal=0 语义 = 无上限杀到超时——AutoFightBehavior.__init__ 默认 kill_goal=0 且 _kill_goal and ... 短路 = ∞，已核对 autofight:98-104,176。超时 duration 每轮 max(600, need*60)，等级不足轮 need=0 → 600s 杀怪升级。）

Step 1: 创建文件 → Step 2: 冒烟 import → Step 3: Commit feat(pipeline): T5 蜈蚣40杀+冲35级

Task 10: T6 神通开启（35级）

Files:

Create: pipeline_bot/pipeline/tasks/t06_shentong.py

Interfaces:

Consumes: 协议定案 docs/协议定案-神通开启与五行之力.md §二：CS3199 {"index":39,"subid":1,"act":"@mingge_kaiqi,1"}；成功=5s 内无 SC100(Type=9)；伴随 CS6900（heal.py _send_action 模式：tail 主链自动递增）
Produces: TASK_TITLE = "T6 开启神通(35级)"

判定实现：发送后 pump 收帧 5s，扫 world.events/帧内 msg==100 且 body 含"等级不足"；SC100 检测需在 update.py 现有分支上确认（若无 msg==100 处理，用 conn.recv_frames 直读 + body_json 解析 Type==9）。

"""T6 — 35级开启神通: CS3199 @mingge_kaiqi,1; 静默5s无SC100(Type=9)=成功。"""
from __future__ import annotations

import json
import time

from pipeline.ctx import PipelineContext, TaskResult

TASK_TITLE = "T6 开启神通(35级)"

LEVEL_NEED = 35
WAIT_S = 5.0


def _send_3199(ctx: PipelineContext, act: str, index: int) -> None:
    body = json.dumps({"index": index, "act": act, "subid": 1},
                      separators=(",", ":")).encode()
    _ = ctx.conn.send_frame(3199, 0, 0, 0, ctx.sid, body)


def run_task(ctx: PipelineContext) -> TaskResult:
    lv = ctx.world.me.level
    if lv < LEVEL_NEED:
        # 仍发一次 — 服务端是权威(抓包实证 40 级槽拒绝/35 级静默成功)
        ctx.log.warning("[T6] 镜像等级 %d<35 — 发送并按回执判定", lv)
    _send_3199(ctx, "@openmianban,1", 104)
    time.sleep(1.0)
    _ = ctx.pump(1.0)
    _send_3199(ctx, "@mingge_kaiqi,1", 39)
    deadline = time.time() + WAIT_S
    while time.time() < deadline:
        for f in ctx.conn.recv_frames(0.25):
            if f.msg_id == 100:
                from pbot.world.update import body_json
                j = body_json(f)
                if int(j.get("Type", 0)) == 9:
                    return TaskResult.fail(
                        f"服务器拒绝: {str(j.get('Msg', ''))[:60]}")
        ctx.pump(0.2)
    return TaskResult.done("5s 静默 — 神通已开(35级槽)")

（执行者核对 body_json 是否从 pbot.world.update 导出——update.py 内部函数，若私有则本地复制一份 zlib→decode6→json.loads 小函数；SC100 是否被 zlib 批量帧包裹要看抓包：cap_shentong_success t=84.24 有独立 msg=100 帧，直读可行。）

Step 1: 创建文件 → Step 2: 冒烟 import → Step 3: Commit feat(pipeline): T6 神通开启

Task 11: T7 五行之力开启

Files:

Create: pipeline_bot/pipeline/tasks/t07_wuxing.py

Interfaces:

Consumes: 协议定案 §三：CS3031(任务体, assets/cs3031_full.bin) → CS1011 @renwu_qwyd,33(传送盟重) → CS1010 {UserID:10031,index:33} → 等 SC643 五行面板 → CS1011 @func17_wxqh,1 → 成功=SC-1 大帧含"[主线]新手指引五"或五行珠入包；pbot.bot.kill._load_cs3031_body(world)（kill.py:162 私有函数 — 本任务改为直接读 assets/cs3031_full.bin）
Produces: TASK_TITLE = "T7 开启五行之力"

"""T7 — 五行之力: renwu_qwyd,33 → NPC10031 → func17_wxqh,1。"""
from __future__ import annotations

import json
import time
from pathlib import Path

from pipeline.ctx import PipelineContext, TaskResult

TASK_TITLE = "T7 开启五行之力"

_ASSETS = Path(__file__).resolve().parents[2] / "assets"
_NPC_ID = "10031"
_MENU_INDEX = 33


def _cs1011(ctx: PipelineContext, act: str) -> None:
    body = json.dumps({"UserID": "999999999", "Act": act,
                       "index": 999999999}, separators=(",", ":")).encode()
    _ = ctx.conn.send_frame(1011, 0, 0, 0, ctx.sid, body)


def _click_npc(ctx: PipelineContext) -> None:
    body = json.dumps({"UserID": _NPC_ID, "index": _MENU_INDEX},
                      separators=(",", ":")).encode()
    _ = ctx.conn.send_frame(1010, 0, 0, 0, ctx.sid, body)


def run_task(ctx: PipelineContext) -> TaskResult:
    # 1) 任务面板体 + 前往(自动传送盟重土城)
    body = (_ASSETS / "cs3031_full.bin").read_bytes()
    _ = ctx.conn.send_frame(3031, 0, 0, 0, ctx.sid, body)
    time.sleep(1.2)                      # 黄金时序: 3031→1011 间隔≥1.2s
    ctx.pump(0.5)
    _cs1011(ctx, "@renwu_qwyd,33")
    # 2) 等传送落地盟重
    deadline = time.time() + 20.0
    while time.time() < deadline:
        ctx.pump(0.5)
        if "土城" in ctx.world.me.map_name:
            break
    if "土城" not in ctx.world.me.map_name:
        return TaskResult.fail(f"未落地盟重(当前 {ctx.world.me.map_name})")
    ctx.checklist.note("五行: 已落地盟重, 走向 NPC 10031")
    # 3) 点 NPC → 等面板(yd_wxjh) → 激活
    _click_npc(ctx)
    deadline = time.time() + 10.0
    seen_panel = False
    while time.time() < deadline:
        ctx.pump(0.5)
        dlg = ctx.world.npc_dialog
        if dlg and "wxjh" in str(dlg.get("sMsg", "")):
            seen_panel = True
            break
        if time.time() % 5 < 0.5:        # 5s 无面板重点一次
            _click_npc(ctx)
    if not seen_panel:
        return TaskResult.fail("五行面板未出现(SC643 无 yd_wxjh)")
    time.sleep(0.5)
    _cs1011(ctx, "@func17_wxqh,1")
    # 4) 等成功信号: 任务链推进 or 五行珠入包(SC-1 子帧)
    deadline = time.time() + 15.0
    while time.time() < deadline:
        ctx.pump(0.5)
        for e in ctx.world.events:
            if e.kind == "task_state" or "新手指引五" in e.detail:
                ctx.world.events.clear()
                return TaskResult.done("任务链推进至指引五")
        for mid in ctx.world.frame_counts:
            _ = mid
    # 兜底: 事件无 task_state(644 帧稀疏) — 查 npc_dialog/任务体
    mb = ctx.world.mission_bodies
    for _t, b in mb.items():
        if "指引五" in str(b):
            return TaskResult.done("任务体已含指引五")
    return TaskResult.fail("15s 未捕获成功信号(面板已点激活)")

（执行者核对：①world.events 的 task_state 事件由 SC644 触发（update.py:217-218 world.record("task_state", f"p3={frame.p3}")）— detail 是 p3=N 不含"指引五"，成功判定主信号应改为：SC-1 大帧解析后 mission_bodies 含新任务体，或入包推送。落地实现时以"SC-1 子帧文本含 '新手指引五' 或 'Index':50675" 为准，给 pump 收帧后扫 world.mission_bodies 值 + 最近 SC-1 原始文本；②若 30s 内无信号但面板激活已发，重发 _cs1011(ctx, "@func17_wxqh,1") 一次再等 10s。）

Step 1: 创建文件（按上方代码 + 核对注记修正成功判定）
Step 2: 冒烟 import → Step 3: Commit feat(pipeline): T7 五行之力

Task 12: T8 沃玛寺庙挂机（时长变量）

Files:

Create: pipeline_bot/pipeline/tasks/t08_woma_farm.py

Interfaces:

Consumes: scripts.run_juece13_woma_kill50.py 的参数覆写模式（cave_mod.CAVE_MAP/_ACK_BODY/_TELEPORT_ACTS 覆写为沃玛）；AutoFightBehavior；d022 墙 load_walls_from_map("assets/maps/d022.map")；ctx.woma_minutes
Produces: TASK_TITLE = "T8 沃玛寺庙挂机"；时长 = ctx.woma_minutes 分钟（默认 30）；无杀怪数上限，跑到时长或手动停

"""T8 — 传送员→沃玛寺庙挂机: 时长=ctx.woma_minutes(默认30min), 无杀数上限。"""
from __future__ import annotations

import time

from pipeline.ctx import PipelineContext, TaskResult

TASK_TITLE = "T8 沃玛寺庙挂机"

WOMA_MAP = "沃玛寺庙一层"
_MAX_ROUNDS = 6


def run_task(ctx: PipelineContext) -> TaskResult:
    import sys
    from pathlib import Path
    root = Path(__file__).resolve().parents[2]
    scripts = str(root / "scripts")
    if scripts not in sys.path:
        sys.path.insert(0, scripts)
    import run_juece13_wugong_farthest as cave_mod
    from run_juece13_wugong_farthest import CaveEnterBehavior
    from run_juece13_wugong_autofight import AutoFightBehavior
    from pbot.bot.pathfind import load_walls_from_map
    from pbot.bot.runner import pump, run

    cave_mod.CAVE_MAP = WOMA_MAP                    # 覆写为沃玛
    cave_mod._ACK_BODY = "沃玛一层"                  # noqa: SLF001 — 协议别名
    cave_mod._TELEPORT_ACTS = ("@jiuxian_move_1,1,3", "@act_move,3")

    walls = load_walls_from_map(str(root / "assets/maps/d022.map"))
    deadline = time.time() + ctx.woma_minutes * 60.0
    total = 0
    rounds = 0
    while time.time() < deadline:
        rounds += 1
        if rounds > _MAX_ROUNDS and ctx.world.me.map_name != WOMA_MAP:
            break
        if ctx.world.me.map_name != WOMA_MAP:
            ctx.log.info("[T8] 进图(当前 %s)", ctx.world.me.map_name)
            bh = CaveEnterBehavior(sid=ctx.sid, kill_target=0)
            r = run(ctx.conn, ctx.world, ctx.updater, bh, duration_s=300.0)
            if ctx.world.me.map_name != WOMA_MAP:
                ctx.log.warning("[T8] 传送失败 result=%s — 3s 重试", r.value)
                time.sleep(3.0)
                continue
            time.sleep(3.0)
            _ = pump(ctx.conn, ctx.world, ctx.updater, 2.0)
        remain = deadline - time.time()
        if remain <= 0:
            break
        af = AutoFightBehavior(sid=ctx.sid, map_name=WOMA_MAP, walls=walls,
                               farthest_fn=None, kill_goal=0)
        r = run(ctx.conn, ctx.world, ctx.updater, af, duration_s=remain)
        total += af.kills
        ctx.log.info("[T8] 第%d轮 result=%s 轮杀=%d 累计=%d 剩余%.0fs",
                     rounds, r.value, af.kills, total,
                     max(0.0, deadline - time.time()))
        if r.value == "done" and ctx.world.me.map_name != WOMA_MAP:
            time.sleep(3.0)      # 死亡回城: 下轮循环重新传送
            _ = pump(ctx.conn, ctx.world, ctx.updater, 2.0)
    ctx.checklist.note(f"T8 沃玛挂机结束: 累计 {total} 杀, "
                       f"时长 {ctx.woma_minutes:.0f}min")
    return TaskResult.done(f"挂机 {ctx.woma_minutes:.0f}min 累计 {total} 杀")


def _woma_farthest_factory():
    """d022 BFS 最远点(撤离用) — run_juece13_woma_kill50 同款。"""
    from collections import deque
    from run_juece13_wugong_autofight import DIR_VECS

    walls: set[tuple[int, int]] = set()

    def farthest(start: tuple[int, int]):
        visited = {start}
        dq = deque([(start, 0)])
        far, far_d = start, 0
        while dq:
            (x, y), d = dq.popleft()
            if d > far_d:
                far, far_d = (x, y), d
            for dx, dy in DIR_VECS:
                nxt = (x + dx, y + dy)
                if nxt in visited or not (0 <= nxt[0] < 250
                                          and 0 <= nxt[1] < 250):
                    continue
                if nxt in walls:
                    continue
                visited.add(nxt)
                dq.append((nxt, d + 1))
        return far, far_d, len(visited) < 50
    return farthest, walls

（执行者注意：①_woma_farthest_factory 当前未接线——若 AutoFightBehavior 强依赖 farthest_fn 撤离(高危图推荐传)，把 walls 填充后在主循环传 farthest_fn=farthest；实现时以 autofight 源码 self._farthest 使用点为准决定传 None 还是接线；②r.value == "done" — TickResult 是 Enum，比较用 r is TickResult.DONE 或 r.value == "done" 以实际定义为准，参照 runner.py。）

Step 1: 创建文件（含 farthest 接线决定）→ Step 2: 冒烟 import → Step 3: Commit feat(pipeline): T8 沃玛挂机

Task 13: 总控 run_pipeline.py

Files:

Create: pipeline_bot/pipeline/run_pipeline.py

Interfaces:

Consumes: Task 3-12 全部产出
Produces: CLI 入口；--dry-run 时不登录、只打印任务序列（冒烟）

"""run_pipeline — 九任务串行总控(一次登录, 全程共用会话)。

用法:
    cd pipeline_bot
    uv run python pipeline/run_pipeline.py \
        [-u longwu988] [-p Longwu1988] [--woma-minutes 30] \
        [--zone 0] [--role-name ""] [--login-port 0] [--dry-run]
"""
from __future__ import annotations

import argparse
import logging
import sys
import time

from pipeline.ctx import TaskResult
from pipeline.naming import safe_filename

TASKS: list[str] = [
    "pipeline.tasks.t01_heal",
    "pipeline.tasks.t02_biqi_kill20",
    "pipeline.tasks.t03_cave_kill20",
    "pipeline.tasks.t04_mine_kill20",
    "pipeline.tasks.t05_wugong_kill40",
    "pipeline.tasks.t06_shentong",
    "pipeline.tasks.t07_wuxing",
    "pipeline.tasks.t08_woma_farm",
]

# 失败后可继续的(软失败): T6 神通(等级信号弱判定) — 其余硬失败即停
_SOFT: set[str] = {"pipeline.tasks.t06_shentong"}


def parse_args() -> argparse.Namespace:
    ap = argparse.ArgumentParser()
    _ = ap.add_argument("-u", "--user", default="longwu988")
    _ = ap.add_argument("-p", "--password", default="Longwu1988")
    _ = ap.add_argument("--woma-minutes", type=float, default=30.0)
    _ = ap.add_argument("--zone", type=int, default=0, help="0=最新区")
    _ = ap.add_argument("--role-name", default="", help="空=随机中文名")
    _ = ap.add_argument("--login-port", type=int, default=0)
    _ = ap.add_argument("--dry-run", action="store_true")
    return ap.parse_args()


def main() -> int:
    a = parse_args()
    logging.basicConfig(
        level=logging.INFO,
        format="%(asctime)s %(levelname)-7s %(name)s: %(message)s",
        stream=sys.stdout,
    )
    log = logging.getLogger("pipeline")

    if a.dry_run:
        print("[dry-run] 任务序列:")
        for t in TASKS:
            mod = __import__(t, fromlist=["run_task"])
            print(f"  - {getattr(mod, 'TASK_TITLE', t)}")
        print(f"[dry-run] 参数: zone={a.zone} role={a.role_name or '<随机>'} "
              f"woma={a.woma_minutes}min")
        return 0

    from pipeline.bootstrap import bootstrap_session
    ctx, real_name = bootstrap_session(
        user=a.user, password=a.password, woma_minutes=a.woma_minutes,
        login_port=a.login_port, force_zone=a.zone,
        role_name=a.role_name)
    log.info("会话就绪: 区=%s 角色=%s", ctx.zone_label, real_name)

    exit_code = 0
    try:
        for name in TASKS:
            mod = __import__(name, fromlist=["run_task"])
            title = getattr(mod, "TASK_TITLE", name)
            log.info("======== %s ========", title)
            t0 = time.time()
            try:
                result: TaskResult = mod.run_task(ctx)
            except Exception as e:  # noqa: BLE001 — 单任务异常不拖垮全链
                result = TaskResult.fail(f"异常: {e!r}")
            dt = time.time() - t0
            ctx.checklist.mark(title, result.ok, result.detail)
            log.info("[%s] %s (%.0fs)", "✓" if result.ok else "✗",
                     result.detail, dt)
            if not result.ok and name not in _SOFT:
                log.error("硬失败 — 流水线终止")
                exit_code = 1
                break
            if not result.ok:
                log.warning("软失败 — 继续下一任务")
    finally:
        ctx.stop[0] = True
        ctx.conn.close()
        log.info("会话已关闭; 打勾文件: %s", ctx.checklist_path())
    return exit_code


if __name__ == "__main__":
    sys.exit(main())

（执行者注意：①ctx.checklist_path() 未定义 — 给 ChecklistWriter 加 path property 或改 ctx.checklist._path→公共属性 path；②__import__ 建议 importlib.import_module。）

Step 1: 修 ChecklistWriter 暴露 path → 创建 run_pipeline.py
Step 2: dry-run 冒烟

cd pipeline_bot && uv run python pipeline/run_pipeline.py --dry-run

Expected: 打印 8 条任务标题（T1..T8）+ 参数行，退出码 0。

Step 3: 全量回归 + 静态检查

uv run pytest -q && uv run ruff check src tests pipeline scripts

Step 4: Commit git add pipeline && git commit -m "feat(pipeline): 九任务串行总控"

Task 14: 端到端实跑验收（真实服务器）

Files:

无新文件（产出 = 打勾文件 + 运行日志）

Interfaces:

Consumes: 完整流水线
Produces: 进度_玖扬复古176_*.md 全勾或勾到失败点
[ ] Step 1: 实跑

cd pipeline_bot && uv run python pipeline/run_pipeline.py --woma-minutes 30 \
    2>&1 | tee /tmp/pipeline_run_$(date +%m%d_%H%M).log

[ ] Step 2: 验收检查
打勾文件存在于 pipeline_bot/ 根，文件名含区名与角色名
每完成任务一行 ☑ + 时间戳；失败任务 ☒ + 原因
观察 T2-T5 的杀怪计数日志与 map 切换；T5 结束 lv≥35；T6 静默成功；T7 五行激活；T8 挂机 30min 计杀
[ ] Step 3: 按实跑结果修 bug（协议时序类小修不扩 scope）
[ ] Step 4: Commit 实跑修复

git add -A pipeline_bot && git commit -m "fix(pipeline): 端到端实跑修复"

Self-Review 结论

Spec 覆盖：九任务（喝药/比奇20/洞穴20/矿区20/蜈蚣40+35级/神通/五行/沃玛挂机30min变量）→ Task 5-12；账号/最新区/随机中文名 → Task 4；共用会话 → Task 3/13；每任务独立文件+打勾 → Task 3/13；死亡复活+传送员回图 → KillBehavior/AutoFight 内建 + T4/T5/T8 轮次循环；点我刷怪 → KillBehavior 内建（kill.py:1241"刷怪"）。✅
占位符扫描：无 TBD/TODO；Task 4/11/12 带"执行者核对"注记（协议细节以源码为准），非占位符。✅
类型一致性：run_task(ctx) -> TaskResult 全任务统一；PipelineContext 字段与 bootstrap 组装一致；ChecklistWriter.mark/note 签名一致；Task 13 引用的 checklist_path() 已在注记中要求补 path property。✅

2026-10-01-pipeline-bot.md 55 KB Постоянная ссылка История Исходник

传奇流水线机器人（pipeline_bot）实施计划

Global Constraints

Task 1: 建立新工程目录并迁移资产

Task 2: SC52 等级解析进 WorldMirror

Task 3: pipeline 包骨架 — PipelineContext / TaskResult / ChecklistWriter / 名字生成

Task 4: 登录编排器 — bootstrap_session()

Task 5: T1 喝药保护任务

Task 6: T2 比奇大城杀 20（含点我刷怪）

Task 7: T3 天然洞穴杀 20

Task 8: T4 比奇矿区杀 20

Task 9: T5 蜈蚣洞穴杀 40（等级≥35 收尾）

Task 10: T6 神通开启（35级）

Task 11: T7 五行之力开启

Task 12: T8 沃玛寺庙挂机（时长变量）

Task 13: 总控 run_pipeline.py

Task 14: 端到端实跑验收（真实服务器）

Self-Review 结论

2026-10-01-pipeline-bot.md 55 KB

Постоянная ссылка История Исходник