Agent/examples/process_pipeline/script/search_eval/procedure-dsl/spec-back-1/tools/wf-patch.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
wf-patch.py — workflow.json 的安全批量字段设置器.

为什么有这个工具:
  workflow.json 由各 phase **直接 Write 骨架 + 逐字段填充** 演化. 但「给几十个 IO
  逐个加 anchor」「给每个 step 填 effect/action/type」这类批量字段赋值, 用 Edit
  一处一处改太碎, 手写整段 JSON 又极易踩转义 / 控制字符坑 (把文件搞坏).
  wf-patch 卡在中间: **你只负责语义决策 (path=value), 工具负责安全落盘 + 合法性校验**.

  - 安全 IO: 工具自己 json.load -> 改 -> json.dump(ensure_ascii=False), 你永远不手写 JSON.
  - 写入即校验 (fail-fast): 每条赋值立刻对照字典树 / type_registry / anchor 格式校验,
    **任何一条非法 -> 报具体哪条错, 整批不写** (不产出悄悄错的文件). lint 仍做全局兜底.

用法:
    # 单条 / 多条 --set (path=value, 只在第一个 '=' 处切, value 可含 '=' 和空格)
    python spec/tools/wf-patch.py --workflow outputs/case-N/workflow.json \
        --set 'p1.s1.inputs[0].anchor=← s0.主角图' \
        --set 'p1.s2.effect=主体生成' \
        --set 'p1.s2.action=生成/图像生成/文生图'

    # 或一次性喂一份 patch 清单 (适合 1.3 加 anchor / 2A 填字段这种几十处批量)
    python spec/tools/wf-patch.py --workflow outputs/case-N/workflow.json --patch _scratch/anchors.json
    # anchors.json = [{"path": "p1.s1.inputs[0].anchor", "value": "← s0.x"}, ...]

    # 只校验不写
    python spec/tools/wf-patch.py --workflow ... --set '...' --dry-run

    # 删字段 (取代手 Edit 删; 字段不存在则幂等跳过)
    python spec/tools/wf-patch.py --workflow ... --unset 'p1.declarations.inputs[0].inferred'

    # 只校验不写
    python spec/tools/wf-patch.py --workflow ... --set '...' --dry-run

路径语法 (proc/step 主用 id (p1/s1), 下标 procedures[N].steps[M] 也接受; inputs/outputs 用 [i]; 嵌套步 id 带点 s2.1 也支持):
    p1.s2.effect                      step 标量字段 (effect/substance/form/via/action/directive/kind/intent/group)
    p1.s1.inputs[0].anchor            IO 字段 (anchor/type/value/id)
    p1.s2.focus                       step 的 focus 数组 (逗号分隔: focus=via,action,out-type-0)
    p1.purpose                        procedure 头部字段 (name/purpose/category/platform/author)
    p1.declarations.inputs[0].desc    declarations 内任意字段 (通用下钻)
    source.url                        case-level 原帖信息 (platform/author/date/url/title/excerpt)
    p1.type_registry.场景图.extends    注册 case-specific 类型 (会自动建 type_registry 段)

value 特殊取值:
    __null__   -> JSON null (用于 substance/form/url 可空)

仍用 Write / Edit 的只剩 (尽量别碰生 JSON):
    - workflow.json 骨架的首次创建 (Phase 1.2 从 template Write)
    - instruction (列表套列表, 手动 Edit; 透传 directive 用 --resolve-passthrough)
    改字段/删字段/改 source 现在都走本工具, 不要再 Read→Edit 改 workflow.json (会反复重读、烧 token).

自动修引号 (load 时兜底):
    workflow.json 由模型直 Write, 偶尔把中文引号写成未转义的 ASCII " → JSON 崩.
    本工具 load 失败时会自动把这类误引号修成「」再 parse; 修成功则继续 patch, 并把
    修复随本次写回落盘 (--dry-run 不写). 修不回才按 exit 2 报错. 不用再手写 _scratch 修复脚本.

退出码:
    0  全部校验通过并写入 (--dry-run 时为校验通过)
    1  有校验失败 (整批未写) / 路径解析失败
    2  CLI 参数错误 / 文件不存在 / JSON 损坏 (且自动修引号也救不回)
"""
from __future__ import annotations

import argparse
import json
import re
import sys
from pathlib import Path

# spec/tools/wf-patch.py -> procedure-dsl/
DSL_ROOT = Path(__file__).resolve().parent.parent.parent
TAX_DIR = DSL_ROOT / 'spec' / 'taxonomy'

# Windows 控制台 UTF-8
for _s in (sys.stdout, sys.stderr):
    if hasattr(_s, 'reconfigure'):
        try:
            _s.reconfigure(encoding='utf-8', errors='replace')
        except Exception:
            pass

KIND_VOCAB = {'step', 'block', 'nested'}

# value/directive 里的「引用占位」文案 — 这些是 anchor 的活, value 应填数据本身.
# 命中即视为「未真正回填」(--resolve-passthrough 会尝试填, lint 会报警).
META_REF = re.compile(r'[（(]?\s*同\s*s[\d]|见\s*s[\d]|←\s*s[\d]|同上')


# ===========================================================================
# 自动修引号: 模型直 Write workflow.json 时常把中文引号写成 ASCII " (未转义) → JSON 崩.
# 仅在 json.loads 失败时兜底调用 (合法文件零开销). 判别: 串内一个 ASCII " 之后第一个
# 非空白字符 ∈ {,:}]} 或 EOF → 真·字符串定界符 (保留); 否则是误写的内容引号 → 换直角
# 引号「」(串内交替 开「/闭」). 逻辑独立内置于本文件 (不 import 任何外部模块);
# scratch/repair_workflow_quotes.py 是同款独立实现, 二者无依赖关系. 改完必须能 parse 才用.
# ===========================================================================

_STRUCT_AFTER = set(',:}]')


def repair_ascii_quotes(raw: str):
    """→ (修后文本, 改动的内容引号数). 纯走字符, 不依赖能否 parse."""
    out, i, n = [], 0, len(raw)
    in_str = esc = False
    open_q = True
    changes = 0
    while i < n:
        c = raw[i]
        if not in_str:
            out.append(c)
            if c == '"':
                in_str, esc, open_q = True, False, True
            i += 1
            continue
        if esc:
            out.append(c); esc = False; i += 1; continue
        if c == '\\':
            out.append(c); esc = True; i += 1; continue
        if c == '"':
            j = i + 1
            while j < n and raw[j] in ' \t\r\n':
                j += 1
            nxt = raw[j] if j < n else ''
            if nxt == '' or nxt in _STRUCT_AFTER:
                out.append(c); in_str = False          # 真·结束符
            else:
                out.append('「' if open_q else '」')     # 误写的内容引号
                open_q = not open_q
                changes += 1
            i += 1
            continue
        out.append(c); i += 1
    return ''.join(out), changes


class PathError(Exception):
    """路径无法解析到 workflow.json 里的目标位置."""


# ===========================================================================
# 字典树加载: leaf 集 + {leaf: 全路径} + 全叶路径集 (与 lint 同款叶子派生)
# ===========================================================================

def _load_tree(name: str):
    """读 spec/taxonomy/{name}.json. 返回 (leaves:set, leaf2path:dict, control:list)."""
    f = TAX_DIR / f'{name}.json'
    if not f.exists():
        return set(), {}, []
    d = json.loads(f.read_text(encoding='utf-8'))
    leaf2path: dict[str, str] = {}

    def walk(node: dict, prefix: list[str]):
        nm = node.get('分类名称')
        if not nm:
            return
        p = prefix + [nm]
        kids = node.get('子分类') or []
        if not kids:                       # 无子分类 = 叶子
            leaf2path[nm] = '/'.join(p)
        for c in kids:
            walk(c, p)

    for top in d.get('最终分类树') or []:
        walk(top, [])
    leaves = set(d.get('$leaves') or leaf2path.keys())
    return leaves, leaf2path, (d.get('$control') or [])


EFFECT_LEAVES, EFFECT_PATHS, _ = _load_tree('effect')
ACTION_LEAVES, ACTION_PATHS, ACTION_CONTROL = _load_tree('action')
TYPE_LEAVES, TYPE_PATHS, _ = _load_tree('type')

def _closest(name: str, leaves) -> str:
    """给个最接近的叶子名做提示 (子串/前缀朴素匹配, 仅供报错文案)."""
    cands = [lf for lf in leaves if name and (name in lf or lf in name)]
    return (' 最接近: ' + '/'.join(cands[:3])) if cands else ''


# ===========================================================================
# 字段校验 -> (ok, normalized_value, err_msg)
# ===========================================================================

def validate_field(field: str, value, proc: dict, pending_types: set[str] = None):
    # null 哨兵 (substance/form/url 可空)
    if value == '__null__':
        if field in ('substance', 'form', 'url'):
            return True, None, ''
        return False, value, f'__null__ 只对 substance/form/url 有意义, {field} 不可为 null'

    # focus 是数组: 逗号分隔 → list ('via,action,out-type-0'); 空串 → []
    if field == 'focus':
        items = [t.strip() for t in str(value).split(',') if t.strip()]
        return True, items, ''

    if field == 'effect':
        if value in EFFECT_LEAVES:
            return True, value, ''
        # 给了全路径 -> 归一到叶名 (schema 存叶名)
        for leaf, path in EFFECT_PATHS.items():
            if value == path:
                return True, leaf, ''
        # 容错: 路径里有某段是合法叶名 → 取最末的那个
        segs = [s.strip() for s in str(value).split('/') if s.strip()]
        for seg in reversed(segs):
            if seg in EFFECT_LEAVES:
                return True, seg, ''
        return False, value, f'effect={value!r} 不是 effect.json 叶子(存叶名).{_closest(segs[-1] if segs else str(value), EFFECT_LEAVES)}'

    if field == 'action':
        # action 存全路径; 给叶名自动展开, 给全叶路径原样接受
        if value in ACTION_PATHS:                 # 是叶名
            return True, ACTION_PATHS[value], ''
        if value in ACTION_PATHS.values():        # 是合法叶路径
            return True, value, ''
        # 容错: 路径里有某段是合法叶名(多写了段 / 前缀错, 如 生成/元素生成/文生图)→ 取最末的合法叶, 纠正到其全路径
        segs = [s.strip() for s in str(value).split('/') if s.strip()]
        for seg in reversed(segs):
            if seg in ACTION_PATHS:
                return True, ACTION_PATHS[seg], ''
        return False, value, (f'action={value!r} 不是合法动作叶子/叶路径 '
                              f'(对到 action.json 的叶子, 如 元素生成、提取/化学提取/反推).{_closest(segs[-1] if segs else str(value), ACTION_LEAVES)}')

    if field == 'type':
        # type 是自由标签: Phase 1 随便起个描述性词即可, 不校验是否叶子/注册。
        # 「对到 type.json 标准叶子 / 注册 type_registry 挂靠」是 Phase 2 归类的活;
        # 最终是否合法由 lint-case (Check 1: 揪未注册的 case-specific type) + render schema 兜底。
        if isinstance(value, str) and value.strip():
            return True, value.strip(), ''
        return False, value, 'type 不能为空'

    if field == 'extends':                        # type_registry entry 的 extends 必须桥到 stdlib 叶子
        if value in TYPE_LEAVES:
            return True, value, ''
        return False, value, f'type_registry extends={value!r} 必须是 type.json 叶子.{_closest(value, TYPE_LEAVES)}'

    if field in ('substance', 'form'):
        # 自由提炼的元素点 — 不查词表、不校验 (spec/tools.md §1; phase2-normalize.md §4).
        # 字符串原样存 (如 "人物、卧室场景"), 数组逐项 strip 后存. 旧的 taxonomy-lookup 校验已废弃.
        if isinstance(value, str):
            return True, value.strip(), ''
        if isinstance(value, list):
            return True, [str(p).strip() for p in value if str(p).strip()], ''
        return False, value, f'{field} 必须是字符串或数组'

    if field == 'anchor':
        if re.match(r'^\s*(←|→)', str(value)):
            return True, value, ''
        return False, value, f'anchor={value!r} 须以 ← (输入引用) 或 → (输出去向) 开头'

    if field == 'kind':
        if value in KIND_VOCAB:
            return True, value, ''
        return False, value, f'kind={value!r} 不在 {sorted(KIND_VOCAB)}'

    # 自由文本字段 (name/value/intent/via/purpose/category/platform/author/desc/group...)
    return True, value, ''


# ===========================================================================
# 路径解析 -> (parent_container, key, proc, field_name)
# ===========================================================================

_SEG = re.compile(r'^([^\[]+)(?:\[(\d+)\])?$')


def _split_seg(seg: str):
    m = _SEG.match(seg)
    if not m:
        raise PathError(f'非法路径段 {seg!r}')
    return m.group(1), (int(m.group(2)) if m.group(2) is not None else None)


def _descend(container, segs, create=False):
    """沿 segs 走进 container, 返回 (parent, last_key). create=True 时自动建中间节点.

    segs 每段可带 [i] 下标. last_key 是 dict 键 (str) 或列表下标 (int);
    设置即 parent[last_key]=value, 删除即 del parent[last_key].
    用于 source.* / declarations.* 等通用路径 (proc/step 的 id 寻址不走这里).
    """
    cur = container
    for i, seg in enumerate(segs):
        name, idx = _split_seg(seg)
        last = (i == len(segs) - 1)

        if idx is None:
            if last:
                if not isinstance(cur, dict):
                    raise PathError(f'{name!r} 的父级不是对象')
                return cur, name
            if not isinstance(cur, dict):
                raise PathError(f'路径段 {name!r} 的父级不是对象')
            if name not in cur:
                if not create:
                    raise PathError(f'路径段 {name!r} 不存在, 无法下钻')
                cur[name] = {}
            cur = cur[name]
        else:
            if not isinstance(cur, dict):
                raise PathError(f'路径段 {name!r} 的父级不是对象')
            if name not in cur:
                if not create:
                    raise PathError(f'路径段 {name!r} 不存在, 无法下钻')
                cur[name] = []
            lst = cur[name]
            if not isinstance(lst, list):
                raise PathError(f'{name} 不是列表')
            if idx >= len(lst):
                if not create:
                    raise PathError(f'{name}[{idx}] 越界或非列表')
                while idx >= len(lst):
                    lst.append({})
            if last:
                return lst, idx
            cur = lst[idx]
    raise PathError('路径为空')


# ── --create 自动建结构用的骨架 ─────────────────────────────────────────────
_STEP_SCALARS = {'effect', 'substance', 'form', 'via', 'action', 'directive', 'kind', 'intent', 'group'}


def _new_procedure(pid: str) -> dict:
    return {
        'id': pid, 'name': '', 'purpose': '', 'category': '', 'platform': '', 'author': '',
        'declarations': {'inputs': [], 'resources': [], 'returns': {}},
        'steps': [],
    }


def _new_step(sid: str) -> dict:
    return {'id': sid, 'kind': 'step', 'via': '', 'inputs': [], 'outputs': []}


def _new_io(is_output: bool, sid: str, idx: int) -> dict:
    if is_output:
        return {'id': f'{sid}o{idx + 1}', 'type': '', 'value': '', 'anchor': ''}
    return {'type': '', 'value': '', 'anchor': ''}


def _split_step_path(remainder: str):
    """create 模式下 step 不存在、无法前缀匹配时, 从路径切出 (sid, fsegs).

    规则: 出现 inputs[i]/outputs[i] 段 → 其前为 sid; 否则末段须是已知 step 标量字段, 其前为 sid.
    sid 可含点 (嵌套步 s2.1)。切不出返回 (None, None) — 不创建, 避免误建。
    """
    segs = remainder.split('.')
    for j, s in enumerate(segs):
        nm, _ = _split_seg(s)
        if nm in ('inputs', 'outputs'):
            return ('.'.join(segs[:j]) or None), segs[j:]
    if segs[-1] in _STEP_SCALARS:
        return ('.'.join(segs[:-1]) or None), [segs[-1]]
    return None, None


def locate(data: dict, path: str, create: bool = False):
    """把 path 解析到目标. 返回 (parent, key, proc, field_name).

    设置即 parent[key] = value. proc 给校验提供 type_registry 上下文.
    proc / step 主用 id 寻址 (p1/s1), 也接受下标 procedures[N]/steps[M]; inputs/outputs 用 [i] 下标.
    step id 可能带点 (嵌套步 s2.1) — 用最长前缀匹配消歧 (s2.1 优先于 s2).
    create=True (构建模式): 缺失的 procedure / step / IO 元素 / 中间结构自动创建, 新建打印到 stderr.
    """
    if '.' not in path:
        raise PathError(f'路径太短 {path!r}, 至少 <proc>.<字段> 或 source.<字段>')

    proc_id, remainder = path.split('.', 1)

    # --- source.* 分支 (case-level 原帖信息, 无 proc 上下文) ---
    if proc_id == 'source':
        src = data.setdefault('source', {})
        parent, key = _descend(src, remainder.split('.'), create=create)
        return parent, key, None, (key if isinstance(key, str) else '')

    procs = data.setdefault('procedures', [])
    # 接受两种 proc 寻址: id (p1) 或下标别名 procedures[N] (映射到第 N 个工序; 弱模型爱用这种)
    m_idx = re.match(r'^procedures\[(\d+)\]$', proc_id)
    if m_idx:
        idx = int(m_idx.group(1))
        if idx < len(procs):
            proc = procs[idx]
        elif create:
            proc = _new_procedure(f'p{idx + 1}')
            procs.append(proc)
            print(f'[wf-patch] + 新建 procedure p{idx + 1} (来自 procedures[{idx}])', file=sys.stderr)
        else:
            raise PathError(f'procedures[{idx}] 越界 (现有 {len(procs)} 个工序)')
    else:
        proc = next((p for p in procs if p.get('id') == proc_id), None)
        if proc is None:
            if not create:
                ids = [p.get('id') for p in procs]
                raise PathError(f'找不到 procedure id={proc_id!r} (现有: {ids})')
            proc = _new_procedure(proc_id)
            procs.append(proc)
            print(f'[wf-patch] + 新建 procedure {proc_id}', file=sys.stderr)

    # --- type_registry 分支 (允许自动建段/条目) ---
    if remainder == 'type_registry' or remainder.startswith('type_registry.'):
        parts = remainder.split('.')
        if len(parts) == 3:
            reg = proc.setdefault('type_registry', {})
            entry = reg.setdefault(parts[1], {})
            return entry, parts[2], proc, parts[2]
        raise PathError('type_registry 路径形如 p1.type_registry.<类型名>.<extends|desc>')

    # --- step 分支: 支持 id 寻址 (p1.s1.effect) 和下标寻址 (procedures[0].steps[0].effect) ---
    matched, field_part = None, None
    m_step = re.match(r'^steps\[(\d+)\]\.(.+)$', remainder)   # 下标寻址 steps[N].<字段>
    if m_step:
        sidx = int(m_step.group(1))
        steps = proc.setdefault('steps', [])
        if sidx < len(steps):
            matched = steps[sidx]
        elif create:
            matched = _new_step(f's{sidx + 1}')
            steps.append(matched)
            print(f'[wf-patch] + 新建 step (procedures.steps[{sidx}] → id s{sidx + 1})', file=sys.stderr)
        else:
            raise PathError(f'steps[{sidx}] 越界 (该工序现有 {len(steps)} 步)')
        field_part = m_step.group(2)
    else:                                                     # id 寻址: 最长前缀匹配现有 step id
        for s in (proc.get('steps') or []):
            sid = s.get('id')
            if not sid:
                continue
            if remainder == sid:
                raise PathError(f'step 路径要带字段, 形如 {proc_id}.{sid}.effect')
            if remainder.startswith(sid + '.') and (matched is None or len(sid) > len(matched['id'])):
                matched = s
        if matched is None and create:
            sid_new, _ = _split_step_path(remainder)
            if sid_new:
                matched = _new_step(sid_new)
                proc.setdefault('steps', []).append(matched)
                print(f'[wf-patch] + 新建 step {proc_id}.{sid_new}', file=sys.stderr)
        if matched is not None:
            field_part = remainder[len(matched['id']) + 1:]   # 'sid.' 之后
    if matched is not None:
        sid = matched.get('id', '?')
        fsegs = field_part.split('.')
        name2, idx2 = _split_seg(fsegs[0])
        if name2 in ('inputs', 'outputs'):
            if idx2 is None:
                raise PathError(f'{name2} 要带下标, 形如 {name2}[0]')
            lst = matched.get(name2)
            if not isinstance(lst, list):
                if create:
                    matched[name2] = lst = []
                else:
                    raise PathError(f'{proc_id}.{sid}.{name2}[{idx2}] 越界 (该 step 有 0 个 {name2})')
            if idx2 >= len(lst):
                if not create:
                    raise PathError(f'{proc_id}.{sid}.{name2}[{idx2}] 越界 (该 step 有 {len(lst)} 个 {name2})')
                while idx2 >= len(lst):
                    lst.append(_new_io(name2 == 'outputs', sid, len(lst)))
            if len(fsegs) != 2:
                raise PathError(f'IO 路径形如 {proc_id}.{sid}.{name2}[{idx2}].anchor')
            return lst[idx2], fsegs[1], proc, fsegs[1]
        else:
            if len(fsegs) != 1:
                raise PathError(f'step 标量字段形如 {proc_id}.{sid}.{name2}')
            return matched, name2, proc, name2

    # --- proc 内其余路径: 头部字段 / declarations.* / return_row.* 等, 走通用下钻 ---
    parent, key = _descend(proc, remainder.split('.'), create=create)
    return parent, key, proc, (key if isinstance(key, str) else '')


# ===========================================================================
# 透传回填: anchor 为纯 ← sN.varname 的 IO, 从源 output 抄 value (逐字回填)
# ===========================================================================

def _is_fillable(value) -> bool:
    """该 value 算「还没真正回填」吗 — 空 / 占位符 / 引用文案."""
    if value in (None, '', '-'):
        return True
    return bool(META_REF.search(str(value)))


def _passthrough_id(anchor):
    """anchor 为 `← <output-id>` (可带 [i] 容器索引) → 返回 output id; 否则 None.

    `← 工序输入` / `← s6 (链)` 等非纯 id 引用返回 None (无法确定唯一源 value).
    """
    m = re.match(r'^\s*←\s*([^\s\[(（]+)', str(anchor or ''))
    if not m:
        return None
    return m.group(1) or None


def resolve_passthrough(data: dict):
    """把 anchor 为纯透传 (← <output-id>)、value 仍空或占位的 input, 用源 output 的 value 逐字填上.

    迭代到不动点 (处理链式透传). 返回 (filled_msgs, warn_msgs).
    """
    out_index = {}          # output id -> output item (读 value)
    for p in data.get('procedures') or []:
        for s in p.get('steps') or []:
            for o in s.get('outputs') or []:
                if isinstance(o, dict) and o.get('id'):
                    out_index[o['id']] = o

    def _src_value(rid):
        src = out_index.get(rid)
        if src is None or _is_fillable(src.get('value')):
            return None
        return src['value']

    filled: list[str] = []
    changed, rounds = True, 0
    while changed and rounds < 20:
        changed, rounds = False, rounds + 1
        for p in data.get('procedures') or []:
            for s in p.get('steps') or []:
                for idx, io in enumerate(s.get('inputs') or []):
                    if not isinstance(io, dict) or not _is_fillable(io.get('value')):
                        continue
                    rid = _passthrough_id(io.get('anchor'))
                    val = _src_value(rid) if rid else None
                    if val is None:
                        continue
                    io['value'] = val
                    filled.append(
                        f"{p.get('id')}.{s.get('id')}.inputs[{idx}].value "
                        f"← 复制自 {rid} ({len(str(val))} 字)"
                    )
                    changed = True

    # 仍填不动的透传 (源 id 找不到) → warn
    warns: list[str] = []
    for p in data.get('procedures') or []:
        for s in p.get('steps') or []:
            for idx, io in enumerate(s.get('inputs') or []):
                if not isinstance(io, dict) or not _is_fillable(io.get('value')):
                    continue
                rid = _passthrough_id(io.get('anchor'))
                if rid and out_index.get(rid) is None:
                    warns.append(
                        f"{p.get('id')}.{s.get('id')}.inputs[{idx}] anchor 指向 "
                        f"{rid} 但找不到该 output id (检查 anchor / output id)"
                    )
    return filled, warns


# ===========================================================================
# 应用
# ===========================================================================

def load_patches(args) -> list[tuple[str, str]]:
    """汇总 --set、--patch 与 --set-file 成 [(path, value), ...]."""
    def _norm(v):
        if isinstance(v, str):
            # 将中文全角双角/单引号自动归一化为标准半角引号，更利于 AI 生图引擎和 Prompt 语法识别
            v = v.replace('“', '"').replace('”', '"').replace('‘', "'").replace('’', "'")
        return v

    def _dq(x: str) -> str:
        # 剥掉成对外层引号. cmd.exe 不剥单引号 → agent 写的 --set 'p.f=v' / p.f='v'
        # 会把单引号原样传进来, 导致 path 变成 'p1 这种 (找不到 procedure). 这里兜底剥掉.
        x = x.strip()
        if len(x) >= 2 and x[0] == x[-1] and x[0] in ('"', "'"):
            x = x[1:-1].strip()
        return x

    out: list[tuple[str, str]] = []
    for s in args.set or []:
        s = _dq(s)                           # 整体外层引号 (cmd 没剥的 'path=value')
        if '=' not in s:
            raise SystemExit(f'wf-patch: --set 缺 "=" : {s!r} (形如 path=value)')
        path, value = s.split('=', 1)        # 只切第一个 '='
        out.append((_dq(path), _norm(_dq(value))))

    # 🟢 新增：从外部文件读取值注入
    for sf in getattr(args, 'set_file', None) or []:
        if '=' not in sf:
            raise SystemExit(f'wf-patch: --set-file 缺 "=" : {sf!r} (形如 path=file_path)')
        path, fpath_str = sf.split('=', 1)
        fpath = Path(fpath_str.strip())
        if not fpath.exists():
            raise SystemExit(f'wf-patch: --set-file 指定的文件不存在: {fpath_str}')
        try:
            value = fpath.read_text(encoding='utf-8')
        except Exception as e:
            raise SystemExit(f'wf-patch: 无法读取 --set-file 指定的文件 {fpath_str}: {e}')
        out.append((path.strip(), _norm(value)))

    if args.patch:
        if not args.patch.exists():
            raise SystemExit(
                f'wf-patch: --patch 文件不存在 {args.patch}\n'
                f'  → --patch 的清单文件是**你要先写的输入**(扁平 `[{{"path":..,"value":..}}]` JSON)。\n'
                f'    先 write_file 写出它, 再 `--patch` 跑它;\n'
                f'    或字段不多时直接内联: `--set \'p1.s1.inputs[0].type=参考图\' --set ...`(不用文件)。')
        try:
            items = json.loads(args.patch.read_text(encoding='utf-8'))
        except json.JSONDecodeError as e:
            raise SystemExit(f'wf-patch: --patch 不是合法 JSON: {e}')
        for it in items:
            out.append((it['path'], _norm(it['value'])))
    return out


# ===========================================================================
# @quote 标记回填: value/directive 写 `@quote|起锚|止锚` 或 `@quote|关键词`,
# 由 --resolve-quotes 顺标记从原文/配图 OCR 匹配真实内容, 批量替换 (空白无关匹配)
# ===========================================================================

def _q_source_text(path: Path) -> str:
    d = json.loads(path.read_text(encoding='utf-8'))
    if not isinstance(d, dict):
        return str(d)
    return '\n'.join(str(d.get(k, '')) for k in ('title', 'body_text') if d.get(k))


def _q_norm(text: str):
    chars, idx = [], []
    for i, ch in enumerate(text):
        if ch.isspace():
            continue
        chars.append(ch)
        idx.append(i)
    return ''.join(chars), idx


def _q_range_between(text: str, frm: str, to: str):
    ns, idx = _q_norm(text)
    nf, _ = _q_norm(frm)
    nt, _ = _q_norm(to)
    if not nf or not nt:
        return None
    p = ns.find(nf)
    if p < 0:
        return None
    q = ns.find(nt, p + len(nf))
    if q < 0:
        return None
    return text[idx[p]: idx[q + len(nt) - 1] + 1]


def _q_braces(text: str, lo: int, hi: int):
    depth, op, i = 0, None, lo
    while i >= 0:
        if text[i] == '}':
            depth += 1
        elif text[i] == '{':
            if depth == 0:
                op = i
                break
            depth -= 1
        i -= 1
    if op is None:
        return None
    depth, j = 0, op
    while j < len(text):
        if text[j] == '{':
            depth += 1
        elif text[j] == '}':
            depth -= 1
            if depth == 0:
                return (op, j) if j >= hi else None
        j += 1
    return None


def _q_query_block(text: str, query: str):
    ns, idx = _q_norm(text)
    nq, _ = _q_norm(query)
    if not nq:
        return None
    p = ns.find(nq)
    if p < 0:
        return None
    o_lo, o_hi = idx[p], idx[p + len(nq) - 1]
    br = _q_braces(text, o_lo, o_hi)           # 命中落在 {...} 内 → 返回整块
    if br:
        return text[br[0]: br[1] + 1]
    lo = text.rfind('\n', 0, o_lo) + 1          # 否则返回所在行/段
    hi = text.find('\n', o_hi)
    return text[lo: (hi if hi >= 0 else len(text))].strip()


def resolve_quotes(data: dict, corpora: list):
    """扫 value/directive 里的 `@quote|...` 标记, 顺标记从 corpora 匹配真实内容批量替换.

    标记: `@quote|起锚|止锚` (范围, 推荐长段) 或 `@quote|关键词` (命中落 JSON 块返回整块, 否则所在行/段).
    corpora: [(label, text), ...] 依次尝试 (原文优先, 再 OCR). 返回 (filled, warns).
    """
    filled, warns = [], []

    def _resolve(v: str):
        parts = v[len('@quote|'):].split('|')
        for label, text in corpora:
            r = _q_range_between(text, parts[0], parts[1]) if len(parts) >= 2 else _q_query_block(text, parts[0])
            if r:
                return r, label
        return None, None

    for p in data.get('procedures') or []:
        pid = p.get('id')
        for s in p.get('steps') or []:
            sid = s.get('id')
            d = s.get('directive')
            if isinstance(d, str) and d.startswith('@quote|'):
                r, label = _resolve(d)
                if r:
                    s['directive'] = r
                    filled.append(f'{pid}.{sid}.directive ← [{label}] {len(r)} 字')
                else:
                    warns.append(f'{pid}.{sid}.directive: @quote 未匹配 {d[:40]!r}')
            for kind in ('inputs', 'outputs'):
                for k, item in enumerate(s.get(kind) or []):
                    if not isinstance(item, dict):
                        continue
                    v = item.get('value')
                    if isinstance(v, str) and v.startswith('@quote|'):
                        r, label = _resolve(v)
                        if r:
                            item['value'] = r
                            filled.append(f'{pid}.{sid}.{kind}[{k}].value ← [{label}] {len(r)} 字')
                        else:
                            warns.append(f'{pid}.{sid}.{kind}[{k}].value: @quote 未匹配 {v[:40]!r}')
    return filled, warns


def main() -> None:
    ap = argparse.ArgumentParser(
        prog='wf-patch.py',
        description='workflow.json 安全批量字段设置器 (写入即校验, 任何一条非法整批不写)',
    )
    ap.add_argument('--workflow', type=Path, required=True, help='目标 workflow.json')
    ap.add_argument('--set', action='append', metavar='PATH=VALUE',
                    help='单条赋值, 可重复. 只在第一个 = 处切; value 可含 = 和空格 (记得整体加引号)')
    ap.add_argument('--patch', type=Path, default=None,
                    help='批量赋值清单 .json: [{"path":..,"value":..}, ...]')
    ap.add_argument('--set-file', action='append', metavar='PATH=FILE_PATH', default=None,
                    help='从外部文件读取内容注入指定字段. e.g. p1.s1.outputs[0].value=_scratch/prompt.txt')
    ap.add_argument('--unset', action='append', metavar='PATH', default=None,
                    help='删字段, 可重复. e.g. p1.declarations.inputs[0].inferred (字段不存在则跳过). 取代手 Edit 删字段')
    ap.add_argument('--resolve-passthrough', action='store_true',
                    help='把 anchor 为纯透传 (← sN.varname)、value 仍空/占位的 IO, 顺 anchor 从源 output 逐字抄 value. 可单独跑, 也可跟在 --set/--patch 后 (先赋值再解析). 迭代处理链式透传')
    ap.add_argument('--dry-run', action='store_true', help='只校验/预演, 不写')
    ap.add_argument('--create', action='store_true',
                    help='(已默认开启, 保留兼容) 缺失的 procedure / step / IO 元素自动创建、文件不存在从空建')
    ap.add_argument('--no-create', action='store_true',
                    help='关闭自动建: 路径不存在就报错 (严格防 typo; 仅在「纯填充已存在结构、想抓打错的路径」时用)')
    ap.add_argument('--resolve-quotes', action='store_true',
                    help='把 value/directive 里的 @quote|起锚|止锚 (或 @quote|关键词) 标记, 顺标记从 --source 原文 / --ocr 配图文本匹配真实内容批量替换. 跟 anchor patch 一起跑')
    ap.add_argument('--source', type=Path, default=None, help='--resolve-quotes 的原文 case json (匹配语料)')
    ap.add_argument('--ocr', type=Path, default=None, help='--resolve-quotes 的配图 OCR 文本文件 (第二语料)')
    args = ap.parse_args()
    # 默认 upsert (缺路径自动建); 弱模型增量 patch 总需要它, gated 反而一直撞「越界/不存在」死循环。
    # --no-create 才关闭 (回到严格寻址、防 typo)。--create 保留为 no-op 兼容旧命令/文档。
    args.create = not args.no_create

    wf = args.workflow
    repaired = 0
    if args.create and not wf.exists():
        data = {}
        print(f'[wf-patch] + 新建 workflow 文件 {wf.name} (从空开始构建)', file=sys.stderr)
    elif not wf.exists():
        print(f'wf-patch: 文件不存在 {wf} (默认会从空新建; 你传了 --no-create 才不建)', file=sys.stderr)
        sys.exit(2)
    else:
        raw = wf.read_text(encoding='utf-8')
        try:
            data = json.loads(raw)
        except json.JSONDecodeError as e:
            # 兜底: 试着把误写成 ASCII 的中文引号修成「」再 parse (模型直 Write 常见崩因)
            fixed, repaired = repair_ascii_quotes(raw)
            try:
                data = json.loads(fixed)
            except json.JSONDecodeError:
                print(f'wf-patch: {wf} 不是合法 JSON, 无法处理: {e}\n'
                      f'  → 去 workflow.json 第 {e.lineno} 行附近修语法 '
                      f'(最常见: 数组 ] 或 对象 }} 结束后、下一个 "key" 前缺逗号), 修好再重跑。'
                      f'别盲目重试本命令 (JSON 没修, 每次都同样报错)。', file=sys.stderr)
                sys.exit(2)
            print(f'[wf-patch] ⚠️ 原文件 JSON 非法 ({e.msg} @ line {e.lineno}); 已自动把 '
                  f'{repaired} 处误写的 ASCII 引号修成「」→ 解析成功, 修复将随本次写回落盘',
                  file=sys.stderr)

    patches = load_patches(args)
    unsets = args.unset or []
    if not patches and not unsets and not args.resolve_passthrough and not args.resolve_quotes:
        print('wf-patch: 没有 --set / --patch / --unset / --resolve-passthrough / --resolve-quotes, 啥也没干', file=sys.stderr)
        sys.exit(2)

    # 解析 + 校验; 任何一条失败 -> 整批不写
    pending_types = set()
    for path, _ in patches:
        m = re.match(r'^p\d+\.type_registry\.([^.]+)\.(extends|desc)$', path)
        if m:
            pending_types.add(m.group(1))

    plan = []          # set: (parent, key, normalized_value, path, display)
    del_plan = []      # unset: (parent, key, path)
    skipped = []       # unset 跳过 (字段本就不在)
    # 错误分两类, 决定原子 vs 部分应用:
    #   fatal  = 路径/结构错 (locate 失败: 路径对不上、缺 --create 等) → 整批不写, 保护结构
    #   value  = 字段值校验失败 (effect/action/type 等值非法) → **跳过这条、其余照写** (部分应用),
    #            末尾以 exit 1 提示回去补。这样一批里几个值错不再连累其余正确字段一起丢。
    fatal_errors = []  # (path, msg)
    value_errors = []  # (path, msg)
    for path, value in patches:
        try:
            parent, key, proc, field = locate(data, path, create=args.create)
        except PathError as e:
            fatal_errors.append((path, str(e)))
            continue
        ok, norm, msg = validate_field(field, value, proc, pending_types)
        if not ok:
            value_errors.append((path, msg))
            continue
        plan.append((parent, key, norm, path, norm if norm is not None else 'null'))
    for path in unsets:
        try:
            parent, key, _proc, _field = locate(data, path)
        except PathError as e:
            fatal_errors.append((path, str(e)))
            continue
        present = (isinstance(parent, dict) and key in parent) or \
                  (isinstance(parent, list) and isinstance(key, int) and key < len(parent))
        (del_plan if present else skipped).append((parent, key, path) if present else path)

    if patches or unsets:
        n_fail = len(fatal_errors) + len(value_errors)
        print(f'[wf-patch] {wf.name} — set {len(plan)}/{len(patches)} 通过, '
              f'unset {len(del_plan)} 删/{len(skipped)} 跳过, {n_fail} 失败')
        for _p, _k, _n, path, disp in plan:
            print(f'  ✓ set   {path} = {disp}')
        for _p, _k, path in del_plan:
            print(f'  ✓ unset {path}')
        for path in skipped:
            print(f'  · skip  {path} (字段本就不存在)')
        for path, msg in value_errors:
            print(f'  ✗ {path}  — {msg}')
        for path, msg in fatal_errors:
            print(f'  ✗✗ {path}  — {msg}  [路径/结构错]')
        # 结构错 = patch 清单跟文件对不上 → 整批不写 (原子, 别半建坏骨架)
        if fatal_errors:
            print(f'\n有 {len(fatal_errors)} 条路径/结构错误, 整批未写入 (先修: 路径写对 / 该建的加 --create).',
                  file=sys.stderr)
            sys.exit(1)
        # 只有字段值非法 → 跳过这几条、其余照常写, 进度不丢; 末尾 exit 1 提示去补
        if value_errors:
            print(f'\n{len(value_errors)} 处字段值非法**已跳过未写**, 其余 {len(plan)} 处照常应用; '
                  f'只需修正这几条后重跑补上 (进度不丢, 别整批重来).', file=sys.stderr)

    # 应用到内存 data (set 先 unset 后; resolve 要看到这些改动). 是否持久化由 dry-run 决定.
    for parent, key, norm, _, _ in plan:
        parent[key] = norm
    for parent, key, _path in sorted(del_plan, key=lambda d: -d[1] if isinstance(d[1], int) else 0):
        if isinstance(parent, list):
            parent.pop(key)
        else:
            del parent[key]

    # 透传回填
    filled, warns = [], []
    if args.resolve_passthrough:
        filled, warns = resolve_passthrough(data)
        print(f'[resolve-passthrough] 回填 {len(filled)} 处透传 value, {len(warns)} 处填不动')
        for m in filled:
            print(f'  ✓ {m}')
        for w in warns:
            print(f'  ⚠ {w}')

    # @quote 标记回填 (顺 @quote|起锚|止锚 从原文/OCR 匹配真实内容批量替换)
    quoted, qwarns = [], []
    if args.resolve_quotes:
        corpora = []
        if args.source and args.source.exists():
            corpora.append(('原文', _q_source_text(args.source)))
        if args.ocr and args.ocr.exists():
            corpora.append(('配图OCR', args.ocr.read_text(encoding='utf-8')))
        if not corpora:
            print('wf-patch: --resolve-quotes 需要 --source (或 --ocr) 指向匹配语料', file=sys.stderr)
        else:
            quoted, qwarns = resolve_quotes(data, corpora)
            print(f'[resolve-quotes] 回填 {len(quoted)} 处 @quote 标记, {len(qwarns)} 处未匹配')
            for m in quoted:
                print(f'  ✓ {m}')
            for w in qwarns:
                print(f'  ⚠ {w}')

    # 有字段值非法被跳过 → 退出码 1 (告诉 agent 还有几条要补), 但有效改动已照常落盘
    final_exit = 1 if value_errors else 0

    n_changes = len(plan) + len(del_plan) + len(filled) + len(quoted)
    if args.dry_run:
        extra = f' (+ 自动修复 {repaired} 处引号, dry-run 同样不写)' if repaired else ''
        print(f'\n--dry-run: 预演 {n_changes} 处改动{extra}, 未写入.')
        sys.exit(final_exit)

    # repaired>0 时即便无字段改动也要落盘 (否则修好的引号没存下来, 文件还是坏的)
    if n_changes == 0 and not repaired:
        print('\n没有改动 (透传 value 都已填好 / 无可赋值), 未写文件.')
        sys.exit(final_exit)

    # 落盘 (安全序列化, 你从不手写 JSON)
    wf.write_text(json.dumps(data, ensure_ascii=False, indent=2) + '\n', encoding='utf-8')
    tail = f' (含自动修复 {repaired} 处引号→「」)' if repaired else ''
    print(f'\n已写入 {n_changes} 处到 {wf.name}{tail}.')
    sys.exit(final_exit)


if __name__ == '__main__':
    main()