fix(codeflow-emit): emit_source 把字面 `== True` 条件还原成 `not x`——IR→@flow round-trip 静默反转分支，flow_from_json/supervisor 全程 ok:true

Start in plaita/dsl/codeflow/_emit.py at _invert_cond. The eq+True special case at lines 189-191 returns not field, while the falsy branch at lines 186-188 already handles real not x IR. Add a round-trip test for source if INPUT.flag == True: to tests/unit/test_codeflow_emit.py using assert_roundtrip, and run that file plus tests/unit/test_reviewfix_a_truthy_cond.py. Done when the new test passes, the emitted branches match the original for flag=True and flag=False, and the existing 54 tests stay green.

优先级 P1 · 依赖：无

基线 HEAD = 65f48f2

Summary

emit_source 的条件逆映射 _invert_cond 把 IR 里的 {operator: "eq", value: True} 无条件当成「not x 的遗留形态」还原成 not x。但 codeflow 编译器对源码里字面写的 if x == True: 产出的正是这个形态，A1 修复后真正的 not x 已经改编译成 {operator: "falsy", value: True}（由 _emit.py:186-188 单独处理）。于是 IR→@flow round-trip 把真值判断反成假值判断，而 round-trip 校验只验「可编译」，全程 ok: true。

证据

1. plaita/dsl/codeflow/_emit.py:189-191 —— 缺陷点，eq+True 被判为 not x：

    if op == "eq" and value is True:
        # `not x` 的遗留形态（ne 取反成 eq+True）——还原为 not 形态，slug 才能对上
        return f"not {field}"

2. plaita/dsl/codeflow/_emit.py:186-188 —— not x 的现行规范形态是 falsy，已被本分支接管：

    if op == "falsy":
        # `not x` 的规范形态（truthy 取反成 falsy）——还原为 not 写法，slug 才能对上
        return f"not {field}"

即 189-191 的「遗留形态」前提在当前编译器下不成立：源码 not INPUT.flag 编译结果为 {'field': '$INPUT.flag', 'operator': 'falsy', 'value': True}（实跑见下），根本不会落到 189-191。

3. plaita/dsl/codeflow/_emit.py:180-185 —— truthy/ne False 两个分支的现状（同一函数内其余「裸写法」还原）：

    if op == "truthy":
        # 裸真值测试的规范形态（编译器对 `if x:` 的产物，A1 修复后）——还原为裸写法
        return field
    if op == "ne" and value is False:
        # 遗留 IR 的裸真值形态（truthy 引入前的编译产物）——同样还原为裸写法
        return field

4. plaita/dsl/codeflow/_expr.py:224-233 —— 字面 == True 走通用比较编译，逐字产出 eq/True，没有任何降级为 truthy 的处理：

    for op, comp in zip(node.ops, node.comparators):
        op_name = _COMPARE_OP.get(type(op))
        if op_name is None:
            raise _CodeflowError(f"不支持的比较运算 {type(op).__name__}", node)
        right = _compile_expr(comp, ctx)
        conds.append({"field": cur_left, "operator": op_name, "value": right})
        cur_left = right
    if len(conds) == 1:
        return conds[0]
    return {"relation": "and", "conditions": conds}

5. plaita/dsl/codeflow/_common.py:97-100 —— ast.Eq 即 "eq"，字面 True 由 _compile_expr 原样作为 value：

_COMPARE_OP = {
    ast.Gt: "gt", ast.GtE: "gte", ast.Lt: "lt", ast.LtE: "lte",
    ast.Eq: "eq", ast.NotEq: "ne", ast.In: "in", ast.NotIn: "notIn",
}

6. plaita/dsl/builder.py:25-26,57 —— builder 入口直达同一形态，不需要写 codeflow 源码也能造出该 IR：

_OP_ALIASES: Dict[str, str] = {
    "==": "eq",

    return {"field": field, "operator": _normalize_op(operator), "value": value}

实跑 cond("$INPUT.ok", "==", True) → {'field': '$INPUT.ok', 'operator': 'eq', 'value': True}；喂给 emit_source 同样得到 if not INPUT.ok:。

7. plaita-ai/plaita_ai/mcp/server.py:109-127 —— flow_from_json 在 json.loads 之后直接把 IR 交给 emit_source，中间无任何 operator/value 归一化；ok 只取重编译是否成功，不比对 IR 语义：

    try:
        ir = json.loads(definition)
    except ValueError as exc:
        return result_json({"ok": False, "error": f"definition is not valid JSON: {exc}"})
    try:
        source = emit_source(ir)
    except EmitError as exc:
        return result_json({"ok": False, "error": f"definition cannot be expressed as @flow: {exc}"})
    except Exception as exc:
        return result_json({"ok": False, "error": f"emit failed: {exc}"})
    compiled = compile_flow(source)
    payload: Dict[str, Any] = {
        "ok": bool(compiled.ok),
        "source": source,
        "flow_id": (ir or {}).get("flow_id") if isinstance(ir, dict) else None,
    }
    if not compiled.ok:
        payload["error"] = "round-trip compile failed"
        payload["errors"] = [e.to_dict() for e in compiled.errors]
    return result_json(payload)

8. plaita-ai/plaita_ai/supervisor.py:164,166 —— supervisor 回填当前源码时同样把 IR 原样交给 emit_source，无归一化/无提示：

            from plaita.dsl.codeflow import emit_source

            return emit_source(ir)

9. plaita/dsl/codeflow/_emit.py:6 —— 被本缺陷打破的模块契约（语义等价）：

    compile_source(emit_source(ir))  ≡  ir   （语义等价）

10. tests/unit/test_reviewfix_a_truthy_cond.py:173-199 —— 现有 emit 回归只覆盖 ne False→裸写法与显式 falsy→not，无一条覆盖源码字面 == True：

def test_emit_legacy_ne_false_still_inverted_to_bare():
    """遗留 IR（truthy 引入前的 ne False 产物）仍可逆为裸写法。"""

def test_emit_falsy_inverted_to_not():

反证命令（除文档字符串外零命中）：

$ grep -rn '== True' tests/unit/test_codeflow_emit.py tests/unit/test_reviewfix_a_truthy_cond.py
tests/unit/test_reviewfix_a_truthy_cond.py:14:  ``ne False`` / ``eq True`` IR 仍可逆。

复现 1：运行期分支互换

$ cd /Users/jeff.jie/projects/infra4agent/plaita && /tmp/plaita-audit-venv/bin/python - <<'PY'
from plaita.dsl.codeflow import compile_source, emit_source, flow_from_source
src = '''@flow("flag_check")
def flag_check(INPUT):
    if INPUT.flag == True:
        return "TAKEN_TRUE_BRANCH"
    else:
        return "TAKEN_FALSE_BRANCH"
'''
ir = compile_source(src)
print("IR condition:", [n["condition"] for n in ir["nodes"] if n["type"] == "if"])
emitted = emit_source(ir)
print(emitted)
a, b = flow_from_source(src), flow_from_source(emitted)
for f in (True, False):
    x, y = a.run({"flag": f}), b.run({"flag": f})
    print(f"flag={f}: original={x!r} emitted={y!r} MATCH={x == y}")
PY

输出：

IR condition: [{'field': '$INPUT.flag', 'operator': 'eq', 'value': True}]
@flow('flag_check')
def flag_check(INPUT):
    if not INPUT.flag:
        return 'TAKEN_TRUE_BRANCH'
    else:
        return 'TAKEN_FALSE_BRANCH'
flag=True: original='TAKEN_TRUE_BRANCH' emitted='TAKEN_FALSE_BRANCH' MATCH=False
flag=False: original='TAKEN_FALSE_BRANCH' emitted='TAKEN_TRUE_BRANCH' MATCH=False

同一编译器对三种写法的产物对照（实跑）：

x == True  -> [{'field': '$INPUT.flag', 'operator': 'eq', 'value': True}]
not x      -> [{'field': '$INPUT.flag', 'operator': 'falsy', 'value': True}]
x          -> [{'field': '$INPUT.flag', 'operator': 'truthy', 'value': True}]

复现 2：仓库自带的 round-trip 口径已能判红

复用 tests/unit/test_codeflow_emit.py:109 的 assert_roundtrip（源码→IR→源码→IR，归一化 IR 相等）：

$ cd /Users/jeff.jie/projects/infra4agent/plaita && /tmp/plaita-audit-venv/bin/python - <<'PY'
import sys
sys.path.insert(0, 'tests/unit')
from test_codeflow_emit import _canonicalize
from plaita.dsl.codeflow import compile_source, emit_source
src = '''@flow("flag_check")
def flag_check(INPUT):
    if INPUT.flag == True:
        return "T"
    else:
        return "F"
'''
ir1 = compile_source(src)
ir2 = compile_source(emit_source(ir1))
c = lambda ir: [n["condition"] for n in _canonicalize(ir)["nodes"] if n["type"] == "if"]
print("canonical eq :", c(ir1))
print("canonical rt :", c(ir2))
PY

输出（归一化后条件已由 eq 变成 falsy，assert_roundtrip 在该源码上抛 AssertionError: round-trip 不等）：

canonical eq : [{'field': '$INPUT.flag', 'operator': 'eq', 'value': True}]
canonical rt : [{'field': '$INPUT.flag', 'operator': 'falsy', 'value': True}]

回归基线（未受影响，可作修复后的对照）：

$ python -m pytest tests/unit/test_codeflow_emit.py tests/unit/test_reviewfix_a_truthy_cond.py -q
54 passed

影响

踩到的角色：任何把 JSON / 存量 IR 转成 @flow 源码的人或 agent——经 plaita-ai MCP 的 flow_from_json（迁移/审查工具的公开入口），或经 supervisor 的 _emit_current_source（JSON definition → CoDeFlow 源码回填，供 LLM 做最小改动）。IR 中条件为 {operator: "eq", value: true} 的三种来源都命中：手写 JSON flow、builder 的 cond(field, "==", True)、以及 codeflow 源码 if x == True: 自身的编译产物。

触发条件与代价：只要条件里有字面 == True，迁移后的源码就变成 not x，分支静默反转——flag=True 走 else、flag=False 走 then（实跑两例均 MATCH=False）。对非布尔输入二者语义同样不同（not x 是 Python 真值取反，x == True 是相等比较），因此不止布尔入参会走反。错误发生在 IR→源码这一步，产出的 @flow 源码本身语法合法、可编译、可运行，不会在任何检查点报错：flow_from_json 返回 ok: true，supervisor 静默回填，compile_flow 的可编译性校验与 emit_source 的模块契约（语义等价）都不参与比对。

不修的后果：一次迁移即把分支语义写反，且结果是一段看起来完全正常的源码，会随迁移直接进入版本库并长期存活；后续以该源码为准的任何修改都会继续放大这个反向。

建议

修复方向：在 plaita/dsl/codeflow/_emit.py 的 _invert_cond 中，不要在缺少依据的情况下把 eq+True 当作 not x。最小修法是删除/停用 _emit.py:189-191 的 eq+True 特判，让它落入下方的通用算子分支，经 _CMP_PY["eq"] 产出 if INPUT.flag == True:——这正是该 IR 的等价源码（不要改写成 truthy 裸写法：x == True 与真值测试对不同输入语义不同）。同时修正 189-191 的注释（当前编译器的 not x 产物是 falsy，见证据 2）；若确实还要兼容旧版 IR 的 eq+True，必须给出可区分的依据（节点 id/slug 或显式版本标记），不能仅凭 operator+value 判定。

验收条件（可执行、可断言）：

1. 新增回归到 tests/unit/test_codeflow_emit.py，复用同文件 assert_roundtrip：对源码 if INPUT.flag == True: 调用 assert_roundtrip(...) 通过。当前实跑该源码使该断言失败（证据 10 的复现 2），归一化条件由 eq/True 变成 falsy/True。
2. 运行期等价断言：flow_from_source(src).run({"flag": True}) 与 flow_from_source(emit_source(compile_source(src))).run({"flag": True}) 同为 "TAKEN_TRUE_BRANCH"，flag=False 同为 "TAKEN_FALSE_BRANCH"。
3. builder 路径同断言：cond("$INPUT.ok", "==", True) 组成的 IR 经 emit_source 后，重编译的归一化条件仍为 {'field': '$INPUT.ok', 'operator': 'eq', 'value': True}。
4. 回归门保持：python -m pytest tests/unit/test_codeflow_emit.py tests/unit/test_reviewfix_a_truthy_cond.py -q 全绿（当前基线 54 passed），且新增用例在未修复时红。

边界

• 本单只覆盖 eq+True 这一条条件逆映射。同一 emit 通道的其它缺陷各自独立成单，不在本单范围：end 节点的 resultType/error 载荷丢失、EVENT 只读 camelCase eventType/eventFilter、集合/while 子作用域 _CompileCtx 不继承 childflows。
• 不顺手重构 _invert_cond 的算子表，也不引入 IR 版本字段这类结构性改动（除非维护者选择用它来区分遗留 IR）。
• _emit.py:183-185 的 ne+False→裸写法沿用同一「遗留形态」假设，属同类但不同分支，本单不覆盖，建议单独评估其非布尔输入下的语义差异。
• 查重：gh issue list -R jeffkit/plaita --state all --limit 100 --json number,title 返回 #11、#16、#22-#52，无一涉及 plaita/dsl、emit_source 或 codeflow 条件逆映射，无同根既有单。