fix(codeflow-emit): 事件节点只读 camelCase eventType/eventFilter——存量 JSON flow（权威 snake_case event_type）emit 出 EVENT(type=None) 必然编译失败，snake_case event_filter 静默丢失

Start in plaita/dsl/codeflow/_emit.py, where _call_event reads only the camelCase eventType and eventFilter keys, and compare with the lookup order in plaita/dsl/codeflow/_nodes.py. Reproduce by running emit_source on tests/integration/server/test_flow.json, which currently yields EVENT(type=None). Done when that input emits EVENT(type=INPUT.event_type, filter={'source': 'test'}), the output compiles, and a new unit test in tests/unit/test_codeflow_emit.py covers snake_case input.

优先级 P1 · 依赖：无

基线 HEAD = 65f48f2

Summary

emit_source 的事件分支只读 camelCase 的 eventType / eventFilter，而存量 JSON flow（以及引擎侧 EventNode 的权威字段名）用的是 snake_case event_type / event_filter。于是把一份合法、可被引擎正常加载运行的 JSON 事件流程交给 emit_source，产出的源码必然是 EVENT(type=None)——event_filter 在 emit 阶段无告警地消失，随后 round-trip 编译必然抛 EVENT 需要 type。sexpr 反编译器同病。

证据 1：事件分支只按 camelCase 取键，没有任何 snake_case 回退。

plaita/dsl/codeflow/_emit.py:489-494

    def _call_event(self, node: Dict[str, Any]) -> str:
        kwargs = [f"type={_invert_value(node.get('eventType'), self.ns)}"]
        if node.get("eventFilter") is not None:
            kwargs.append(f"filter={_invert_value(node['eventFilter'], self.ns)}")
        kwargs.extend(self._common_kwargs(node))
        return f"EVENT({', '.join(kwargs)})"

证据 2：键缺失时 _invert_value(None, ns) 返回字符串 "None"——不是报错，而是把 None 写成源码字面量。

plaita/dsl/codeflow/_emit.py:151-157

    if v is True:
        return "True"
    if v is False:
        return "False"
    if v is None:
        return "None"
    return repr(v)

反证：_emit.py 中 snake_case 事件键零命中。

$ grep -n "event_type" plaita/dsl/codeflow/_emit.py
（零命中，exit=1）
$ grep -n "event_type\|eventType\|eventFilter\|event_filter" plaita/dsl/codeflow/_emit.py
490:        kwargs = [f"type={_invert_value(node.get('eventType'), self.ns)}"]
491:        if node.get("eventFilter") is not None:
492:            kwargs.append(f"filter={_invert_value(node['eventFilter'], self.ns)}")

证据 3：引擎侧事件节点明确声明 snake_case 才是 JSON flow 的历来形态，并加 AliasChoices 兼容编译器发的 camelCase——正好反证「两种键名都合法，但 emit 只认一种」。

plaita/node/event_node.py:36-42

    # AliasChoices：codeflow 编译器（_nodes.py EVENT 分支）发的是 camelCase
    # "eventType"，而 JSON flow 定义历来用 "event_type"——两者都收（2026-10-01
    # dist-demo 实证：此前 DSL 的 EVENT 从未真正构建成功过）。
    event_type: str = Field(
        validation_alias=AliasChoices("event_type", "eventType"),
        description="订阅的事件类型：须与事件发布方的 type 一致（支持 $ 表达式）")
    event_filter: Dict[str, Any] = Field(default_factory=dict, description="事件过滤器：按点路径匹配事件负载字段，如 {\"data.status\": \"ok\"}")

证据 4：仓库自带存量 JSON flow 就是 snake_case，且 event_filter 有实际内容。

tests/integration/server/test_flow.json:35-44

    {
      "id": "wait_for_event",
      "name": "等待事件",
      "type": "event",
      "event_type": "$INPUT.event_type",
      "event_filter": {
        "source": "test"
      },
      "next": "end"
    },

证据 5：这份 JSON 不是死文件——它被集成用例加载并存入 FlowStorage 真跑，所以 snake_case 是引擎真实消费的形态（可达性反证）。

tests/integration/server/test_flow_worker.py:26-31

        # 加载测试流程定义
        with open(os.path.join(os.path.dirname(__file__), 'test_flow.json'), 'r') as f:
            self.test_flow_json = json.load(f)
        
        # 保存流程定义到存储中
        self.flow_storage.save_flow(self.test_flow_json)

证据 6：bug 之所以潜伏，是因为编译器自己产出 camelCase——emit 对「自产 IR」自洽，只在存量 snake_case JSON 上崩。

plaita/dsl/codeflow/_nodes.py:89-95

    if kind == "EVENT":
        etype = _const(kw.get("type") or kw.get("event_type") or kw.get("eventType"))
        if etype is None and pos:
            etype = pos[0].value if isinstance(pos[0], ast.Constant) else _compile_expr(pos[0], ctx)
        if etype is None:
            raise _CodeflowError("EVENT 需要 type", node)
        spec = {"type": "event", "id": nid, "eventType": etype}

证据 7：sexpr 反编译器同样只读 eventType，而 sexpr 编译器要求必须有 :type——同一份 JSON 在 sexpr 链路上也是「产出 nil，随后报错」。

plaita/dsl/sexpr.py:935-936

    elif t == "event":
        parts.append(f":type {_expr_to_src(n.get('eventType'))}")

plaita/dsl/sexpr.py:721-725

def _c_event(args: List[Any], ctx: _Ctx) -> Dict[str, Any]:
    pos, kw = _split_args(args)
    etype = _opt(kw, "type", "event-type", "eventType")
    if etype is None:
        raise SyntaxError("event 需要 :type")

$ grep -rn "event_type" plaita/dsl/sexpr.py
（零命中，exit=1）

复现（实跑，HEAD=ea53e46）：

$ cd /Users/jeff.jie/projects/infra4agent/plaita && /tmp/plaita-audit-venv/bin/python -c "
import json
from plaita.dsl.codeflow import emit_source, compile_source
ir=json.load(open('tests/integration/server/test_flow.json'))
src=emit_source(ir); print(src); compile_source(src)"
@flow('event_flow_demo', desc='使用真正的EventNode节点演示事件驱动流程', version='1.0.0', author='Plaita开发团队', timeout='PT60S', metadata={'version': '1.0.0', 'description': '演示延时和事件节点的流程', 'categories': ['demo', 'event', 'delay']})
def event_flow_demo(INPUT):
    wait_for_event = EVENT(type=None)
    return INPUT.message
round-trip FAILED: _CodeflowError [codeflow] 第 3 行: EVENT 需要 type

（注意产出源码里 EVENT(type=None) 没有任何 filter=——{'source': 'test'} 已消失。）

对照实验：同一 event 节点、只换键名，差异完全由键名造成。

$ /tmp/plaita-audit-venv/bin/python -c "
import json, copy
from plaita.dsl.codeflow import emit_source
ir=json.load(open('tests/integration/server/test_flow.json'))
def ev(x):
    return [n for n in x['nodes'] if n.get('type')=='event'][0]
print('snake:', [l for l in emit_source(ir).splitlines() if 'EVENT' in l][0])
ir2=copy.deepcopy(ir); n=ev(ir2)
n['eventType']=n.pop('event_type'); n['eventFilter']=n.pop('event_filter')
print('camel:', [l for l in emit_source(ir2).splitlines() if 'EVENT' in l][0])"
snake:     wait_for_event = EVENT(type=None)
camel:     wait_for_event = EVENT(type=INPUT.event_type, filter={'source': 'test'})

引擎可达性（同一份 snake_case JSON 加载成功，字段齐全）：

$ /tmp/plaita-audit-venv/bin/python -c "
import json
from plaita.core.flow import Flow
f=Flow.model_validate(json.load(open('tests/integration/server/test_flow.json')))
n=[x for x in f.nodes if getattr(x,'node_type',None)=='event'][0]
print('event_type=',repr(n.event_type),'event_filter=',n.event_filter,'next=',n.next)"
event_type= '$INPUT.event_type' event_filter= {'source': 'test'} next= end

sexpr 链路同证：

$ /tmp/plaita-audit-venv/bin/python -c "
import json
from plaita.dsl.sexpr import flow_to_sexpr, compile_sexpr
s=flow_to_sexpr(json.load(open('tests/integration/server/test_flow.json')))
print([l for l in s.splitlines() if 'event' in l][:1]); compile_sexpr(s)"
['(flow event_flow_demo :desc 使用真正的EventNode节点演示事件驱动流程 (start :id start -> wait_for_event) (event :id wait_for_event :type nil) (end :id end :output $INPUT.message))']
sexpr FAILED: SyntaxError event 需要 :type

三个生产调用点都不对 IR 键做任何归一化，直接把 JSON 交给 emit_source：

plaita-ai/plaita_ai/mcp/server.py:109-114

    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)

plaita-ai/plaita_ai/supervisor.py:161-166

        if not isinstance(ir, dict):
            return ""
        try:
            from plaita.dsl.codeflow import emit_source

            return emit_source(ir)

plaita-ai/plaita_ai/cli/main.py:46-47

    raw = sys.stdin.read() if args.stdin else Path(args.definition).read_text(encoding="utf-8")
    payload = json.loads(flow_from_json(raw))

影响

• 踩坑者一：用 MCP 工具 flow_from_json（plaita-ai/plaita_ai/mcp/server.py）或被 plaita-ai 的 supervisor 托管的部署，把已经在跑的存量 JSON 事件流程迁移成 @flow 源码。事件类流程上该能力整体不可用：返回 ok:false + EVENT 需要 type，迁移路径被硬堵；supervisor 的 _emit_current_source 会把异常吞成空串（return ""），LLM 只能退回只看 JSON 的旧路，能力静默退化。
• 踩坑者二：plaita-ai CLI 的 emit 子命令（cmd_emit）对同一输入必然非零退出。
• 代价不是「跑错结果」而是「改不动 / 迁不动」：event_filter 的丢失发生在 emit 阶段且无任何告警（emit_source 不抛错、不返回 warnings），只有随后 round-trip 编译才因 type=None 报错；若使用者把 emit 产物当作半成品手工补 type=，那么 event_filter 会无声消失，流程变成「等一个事件但不做任何过滤」——订阅面被放大，可能收到本不该命中该节点的其他事件。
• 生产可达性：完全可达（纯库内 JSON→源码路径，无特殊部署形态要求）；且受害面正是「存量 JSON 事件流程」这个迁移能力的核心场景。唯一兜住的是 flow_from_json 的 round-trip 校验会返回 ok:false，因此不会静默产出错误的可运行源码——这也是本单定 P1 而非 P0 的原因。
• 不修的后果：任何用 snake_case 写的、引擎能正常跑的 JSON 事件流程，都永远无法经 emit 得到可编译的 @flow 源码，JSON→DSL 迁移在事件节点上是死路。

建议

• 修 plaita/dsl/codeflow/_emit.py:489-494 的 _call_event：按与 _nodes.py:90 相同的优先级读键，加 snake_case 回退，例如 node.get('eventType', node.get('event_type')) 与 node.get('eventFilter', node.get('event_filter'))（注意用 if ... is not None 而非 in，避免值为 None 时被当命中）。最小修法就是这两行。
• 可选（同因同治，但属同一缺陷面）：plaita/dsl/sexpr.py:936 的 n.get('eventType') 同样加 event_type 回退；否则 sexpr 链路仍是「产出 nil 再报错」。若想收窄本单范围，可只修 codeflow emit 并在本单注明 sexpr 为同类残留——建议一并修，因为两者读的是同一份 IR。
• 验收条件（可断言）：
  1. 新增单测（建议放 tests/unit/test_codeflow_emit.py），以 tests/integration/server/test_flow.json 为输入直接调用 emit_source(ir)，断言产出源码包含 EVENT(type=INPUT.event_type, filter={'source': 'test'})，且 compile_source(emit_source(ir)) 不抛异常。注意现有 test_roundtrip_code_and_event（tests/unit/test_codeflow_emit.py:310-317）走的是「源码→IR→源码」自产 IR 路径，永远只见到编译器自己写的 camelCase，因此对本缺陷零覆盖，必须补一条「手写 snake_case JSON → emit」方向的用例。
  2. 反向用例：断言 camelCase 输入 {"eventType": ..., "eventFilter": ...} 的 emit 结果与 snake_case 输入完全一致（两条路径收敛到同一份源码）。
  3. 上面 Summary 里的复现命令应从 round-trip FAILED: _CodeflowError [codeflow] 第 3 行: EVENT 需要 type 变为正常打印源码且 round-trip 通过。
  4. plaita-ai 侧回归：flow_from_json(json.dumps(test_flow.json)) 应返回 ok:true。

边界

• 只修「emit/反编译读事件键时的键名回退」，不改 EventNode 的字段与别名（plaita/node/event_node.py 的 AliasChoices 已正确，无需动），不改编译器产出的 camelCase IR 形态（_nodes.py:95 保持不变，全仓 IR 键名规范化是另一个更大的议题），也不动 _invert_value 把 None 渲染成 "None" 的通用行为。
• 不做「IR 键名统一（camelCase vs snake_case）的大一统重构」，也不顺手扫其他节点的同类键名问题——本单只装事件节点这一个 gap。
• 与既有 issue 不重叠：gh issue list -R jeffkit/plaita --state all --limit 100（#11、#16、#22-#52 共 33 单）中无一涉及 plaita/dsl/codeflow/_emit.py、plaita/dsl/sexpr.py 或事件节点键名。同范围的相邻单 #37（fix(builder): LinearBuilder 把 event 当终端节点）也以 event 为对象，但缺陷面是构建期 next 串接（plaita/dsl/builder.py 的 _TERMINAL_TYPES），与本单的 IR 键名读取 无交集；两单可独立提报。