实战四：用 Graph 重构大型项目

from typing import TypedDict, NotRequired

class RefactorState(TypedDict):
    modules: list[str]                 # 从 inventory 来
    typing_baseline: dict[str,int]     # 模名 -> 错误数
    plan_md: str                       # oracle 产出的 plan
    approved: bool                     # user 是否同意
    cur_module: str | None             # current node 因子
    completed: list[str]
    failures: list[tuple[str,str,str]] # (module,file,why)
    final_report: str

async def node_refactor_one_module(state: RefactorState) -> RefactorState:
    module_path = state["cur_module"]
    if module_path in state["completed"]:
        return state

    logger.info(f"着手中: {module_path}")

    files_under_module = list(Path(module_path).rglob("*.py"))
    # Splitting 秀: 单 Atmospheric File ≥ 250 lines
    for f in files_under_module:
        if num_loc(f) >= 250:
            split_plan = await oracle_decide_split(f)
            # oracle 帮我们决定 split plan
        # 否则单文件作为 unit of work

    # 用 sub build 同时 refactor 多文件
    parallel_prs = asyncio.gather(*[
        OpenCode_subagent(subagent_type="build",
            prompt=f"""
[CONTEXT]
project root: {project_root}
module: {module_path}
target file: {f.path}

[GOAL]
为该文件加完整 type annot, 拆长函数 (< 80 行 / fn)

[DOWNSTREAM]
verify_node 会跑 basedpyright, 失败要这文件回退到 pass

[REQUIRED TOOLS]
read, edit, lsp

[MUST DO]
1. read 文件
2. read 相关 importers 看是否有 API 依赖
3. 加 type, 不改行为(except for shrink)
4. 运行 ruff (black + isort + unused import)
5. 输出 1-3 行 diff summary

[MUST NOT DO]
- 不跨模块改 file
- 不替代 inline impl
- 不引入新依赖
""",
        )
        for f in files_under_module
    ])

    await parallel_prs
    state["completed"].append(module_path)
    return state

async def node_verify_mod_graph(state: RefactorState) -> RefactorState:
    mod = state["cur_module"]
    # 1 类型校验
    type_result = subprocess.run(["basedpyright", mod],
                                  capture_output=True, text=True)
    if type_result.returncode != 0:
        state["failures"].append((mod, type_result.stdout[:200]))
        return replace(state, cur_module=None)  # 路由回 remediate

    # 2 unit test
    test_result = subprocess.run(["pytest", module_to_test_dir(mod), "-x", "-q"],
                                  capture_output=True, text=True)
    if test_result.returncode != 0:
        state["failures"].append((mod, test_result.stdout[:200]))
        return replace(state, cur_module=None)

    return state

def route_after_verify(state: RefactorState) -> str:
    if any(f[0] == state["cur_module"] for f in state["failures"]):
        return "remediate"
    return "next"

async def node_plan(state: RefactorState) -> RefactorState:
    inventory_dump = serialize_dict_diff(state, only_top_keys=True)
    plan_oracle = await OpenCode_subagent(
        subagent_type="oracle",
        prompt=f"""
[CONTEXT]
这是项目 inventory + typing baseline + module import graph:

{inventory_dump}

[GOAL]
设计一份 30 天重构 plan spec.md

[CRITICAL DECISIONS]
1. 哪个模块先重构? (lowest coupling-first)
2. 三个 orchestrator (main, cli, web_server) 拆成新 lib 名什么?
3. 哪个模块跳过(因 legacy 太久 / 未来替换)? (限 ≥ 1 个)
4. test framework 用什么? (pytest / pytest-xdist 建议为什么异?)

[OUTPUT]
markdown spec.md with header:
# Plan
## Rationale
## Phase 1 (Week 1)
## Phase 2 (Week 2-3)
## Phase 3 (Week 4)
## Risks
""",
    )
    state["plan_md"] = plan_oracle
    return state

async def wait_for_user_approval(state: RefactorState) -> RefactorState:
    spec = state["plan_md"]
    Path("spec.md").write_text(spec)
    print("Spec written to spec.md.  Reply 'go' to continue.")
    answer = input_presented_to_user()
    if answer.strip().lower() != "go":
        # 让用户修改 plan_md
        ...
    state["approved"] = True
    return state

def route_after_verify(state: RefactorState) -> str:
    failed_on_mod = [f for f in state["failures"]
                      if f[0] == state["cur_module"]]
    if len(failed_on_mod) >= 3:
        print(f"在 {state['cur_module']} 失败 ≥ 3 次, 报告而退出")
        return "escalate"
    if failed_on_mod:
        return "remediate"
    return "next"

def safety_check(state: RefactorState) -> None:
    # every "completed" mod must pass verify
    for mod in state["completed"]:
        if not module_passes_verify(mod):
            raise RuntimeError(f"completed mod {mod} failed verify")
    # 全测试一次性终末
    if state["final_report"]:
        full_test = subprocess.run(["pytest", "-x", "-q"], ...)
        if full_test.returncode != 0:
            raise RuntimeError("full test suite failed post-refactor")

async def node_fuse_orchestrator(state: RefactorState) -> RefactorState:
    # 1. 把三个 orchestrator 的公共逻辑 extract 到 `lib/<project>_common/`
    # 2. orchestrator 只保留 CLI / WS / main 入口 + plugins
    extract_actions = await OpenCode_subagent(
        subagent_type="explore",
        prompt=("[GOAL] 列出 main.py / cli.py / web_server.py 共享的"
                " helper functions, ≤ 30 个, 按 import 关系排序"),)
    # 然后让 build 一一 extract phase
    extracted = await OpenCode_subagent(subagent_type="build",
        prompt=f"Based on:\n{extract_actions}\n\n做 extract + update imports")
    state["final_report"] = f"Extracted: {extracted[:100]}"
    return state

async def test_graph_runs_on_tiny_project(tmp_path):
    # mock excludes ok state
    state = init_minimal_state(tmp_path)
    runner = compile_graph(graph_spec).compile()
    final = await runner.ainvoke(state)
    assert "orchestrator extraction" in final["final_report"]
    assert 95 > len(final["completed"]) > 0

async def test_remediation_loop():
    state = init_with_one_fail_mod(tmp_path)
    runner = compile_graph().compile()
    final = await runner.ainvoke(state)
    assert final["failures"] == []