fix(async): lazy（Generator 模式 / flow.debug）消费者放弃后流程仍在后台线程跑完——on_flow_end 在节点仍在执行时按 result=None 触发，_running 已复位可对同一实例重入

Start with the threaded branch of async_gen_to_sync in plaita/core/async_utils.py (around lines 174-216), where _drive() has no stop check, and with the on_lazy_finally hook in plaita/core/executor_compat.py (lines 34-36), which resets _running too early. Reproduce with the issue's script, then add a test in tests/unit/test_async_utils.py. Done means the node execution count stops growing after close(), on_flow_end never fires while nodes remain unexecuted, and a second run_compatible on the same instance cannot overlap the first.

优先级 P1 · 依赖：无

基线 HEAD = 65f48f2

Summary

同步 lazy 入口（Flow.debug() / FlowExecution.run_compatible(..., lazy=True)）在已运行的事件循环内
被调用时，async_gen_to_sync 走「工作线程 + 队列」分支（plaita/core/async_utils.py:174 起）。该分支
没有任何停止信号：消费者 break 后 close()，后台线程仍持有一个新建事件循环，把 async generator
全速拉到底。后果有三：on_flow_end 在节点仍在执行时按 result=None 触发、_running 提前复位、
消费者可对同一实例再次 run，两遍流程并行执行。

1. plaita/core/async_utils.py:200-216 —— 线程分支的消费者侧 finally 只有 join(timeout=5) 与一条
   warning，没有 stop event、没有取消：

        worker = threading.Thread(target=_drive, daemon=True)
        worker.start()
        try:
            while True:
                item = q.get()
                if item is _DONE:
                    break
                if isinstance(item, BaseException):
                    raise item
                yield item
        finally:
            worker.join(timeout=5)
            if worker.is_alive():
                logger.warning(
                    "async_gen_to_sync worker still alive after join; "
                    "underlying async generator may not have closed cleanly"
                )

_drive() 内部没有停止检查，q.get() 也不设超时；join 超时后仅告警（:213）。

2. plaita/core/async_utils.py:183-192 —— 后台线程在无界队列（:177 的 _queue.Queue()）上全速拉取，
   生产与消费彻底解耦：

                while True:
                    try:
                        item = loop.run_until_complete(agen.__anext__())
                    except StopAsyncIteration:
                        q.put(_DONE)
                        break
                    except BaseException as e:  # noqa: BLE001
                        q.put(e)
                        break
                    q.put(item)

3. plaita/core/async_utils.py:159-173 —— 对照：调用线程没有运行 loop 时走的是消费者驱动分支，
   yield（:167）逐次触发 agen.__anext__()，消费者一停流程就停。两个分支语义不一致，
   本缺陷只在「调用线程已有运行 loop」这一前置条件下成立：

    if not has_loop:
        loop = asyncio.new_event_loop()
        try:
            while True:
                try:
                    item = loop.run_until_complete(agen.__anext__())
                except StopAsyncIteration:
                    break
                yield item

4. plaita/core/executor_compat.py:34-36 —— lazy 的 finally 钩子先发 on_flow_end，紧接着无条件把
   _running 置回 False，不等后台线程停下：

                    on_lazy_finally=lambda exc: (
                        _emit_flow_end_on_close(flow, exc, self.callback_manager), setattr(self, "_running", False),
                    ),

5. plaita/core/flow.py:444-446 —— flow.debug() 就是这条 lazy 路径的公开入口：

    def debug(self, *args, **params):
        _enforce_dict_input(args)
        return FlowExecution().run_compatible(self, True, *args, **params)

6. docs-site/docs/guide/execution-modes.md:39 把 flow.debug() 作为生成器模式标准用法示范
   （for step in flow.debug(name="test"):，并在 :44 示范 break）；:62 明文承诺：

    `on_flow_end` 回调会推迟到生成器**真正被消费完毕或关闭**时才触发，而非 `flow.debug()` 调用返回时。这样保证生命周期回调与实际执行进度一致。

实测行为与该承诺相反：close() 会在节点仍在执行时就触发 on_flow_end。

复现（实跑，/tmp/plaita-audit-venv/bin/python，Python 3.12，plaita editable）

20 个各 time.sleep(0.5) 的节点，消费 1 步后放弃并 close()；随后对同一实例跑第二次 eager
run_compatible，再等 6 秒观察后台：

cd /tmp && /tmp/plaita-audit-venv/bin/python - <<'EOF'
import asyncio, time
from typing import ClassVar
from plaita import Node
from plaita.core.flow import Flow
from plaita.core.executor import FlowExecution
import plaita.node as pn

executed, events = [], []
class SlowNode(Node):
    node_type: ClassVar[str] = 'slowprobe_final'
    def execute(self, execution):
        executed.append(self.id); time.sleep(0.5); return f'd-{self.id}'
class CB:
    def on_node_start(self, *a, **k): pass
    def on_node_end(self, *a, **k): pass
    def on_flow_start(self, *a, **k): events.append('flow_start')
    def on_flow_end(self, flow, result=None, error=None, exception=None):
        events.append(('flow_end', result, error, len(executed)))
pn.get_default_registry().register(SlowNode)
N = 20
nodes = [{'id': 's', 'type': 'start', 'next': 'n0'}]
for i in range(N):
    nodes.append({'id': f'n{i}', 'type': 'slowprobe_final',
                  'next': (f'n{i+1}' if i < N-1 else 'end')})
nodes.append({'id': 'end', 'type': 'end', 'output': f'$NODE.n{N-1}'})
flow = Flow.model_validate({'flow_id': 'spf', 'name': 'p', 'nodes': nodes})

async def main():
    fe = FlowExecution(callback_handlers=[CB()])
    g = fe.run_compatible(flow, True, {})   # 公开 lazy 入口（flow.debug 同路径）
    for _ in g: break                       # 消费 1 步后放弃
    t0 = time.time(); g.close(); dt = time.time() - t0
    print('close() returned after %.2fs' % dt)
    print('events right after close:', events)
    print('executed at close: %d/%d' % (len(executed), N))
    print('_running after close:', fe._running)
    try:
        fe.run_compatible(flow, False, {})
        print('SECOND RUN SUCCEEDED on same instance')
    except Exception as e:
        print('second run rejected:', type(e).__name__, e)
    print('executed after 2nd run: %d' % len(executed))
    await asyncio.sleep(6)
    print('executed +6s: %d' % len(executed))

asyncio.run(main())
EOF

输出：

close() returned after 5.00s
events right after close: ['flow_start', ('flow_end', None, None, 10)]
executed at close: 10/20
_running after close: False
SECOND RUN SUCCEEDED on same instance
executed after 2nd run: 40
executed +6s: 40

读法：close() 在 5.00s 的 join 超时后返回，此时只有 10/20 个节点执行完，on_flow_end 已按
result=None, error=None 触发，_running 已是 False。随后同一实例的第二次 eager run_compatible
成功，计数涨到 40 = 首轮后台补完的 10 + 第二轮 20——两遍并行。

同一线程分支上另跑一遍不带回调的等价用例（20 个 0.5s 节点，消费 1 步后 close()），逐秒采样：

close() returned after 5.00s, produced=11
  +6s produced=13
  +7s produced=15
  +8s produced=17
  +9s produced=19
  +10s produced=20

即 close() 返回后后台线程仍以约每秒 2 项（每项 0.5s）的速度继续执行，直到把整条流程跑完。

最小同步生成器复现（同一线程分支）

消费 2 项后放弃，后台 100 项已跑完：

cd /tmp && /tmp/plaita-audit-venv/bin/python - <<'EOF'
import asyncio
from plaita.core.async_utils import async_gen_to_sync
produced=[]; closed=[]
async def agen():
    try:
        for i in range(100):
            produced.append(i); await asyncio.sleep(0.01); yield i
    finally: closed.append(True)
async def main():
    g = async_gen_to_sync(agen())
    got=[]
    for item in g:
        got.append(item)
        if len(got)==2: break
    g.close()
    print('produced at abandon:', len(produced), 'agen finally:', bool(closed))
asyncio.run(main())
EOF

输出：

produced at abandon: 100 agen finally: True

反证：生产代码当前没有 lazy 调用点

$ grep -rn "flow\.debug(\|lazy=True" --include='*.py' plaita/ plaita-console/backend/ plaita-ai/plaita_ai/ | grep -v '/tests/'
（零命中）

console dry-run 走的是 eager 入口，不是 lazy：plaita-console/backend/services/dryrun.py:324
result = FlowExecution().run(（三参数形式，无 lazy）。因此本缺陷当前属库嵌入 / 宿主可达，
不是 console 直接触发。

影响

谁：在异步宿主里按文档使用 flow.debug() 或 run_compatible(..., lazy=True) 的消费者——
交互式调试器、可视化单步、Jupyter / async web handler 里「看几步就停」的调用方
（execution-modes.md:44 就示范了 break）。

前置条件：调用线程已有运行中的事件循环（asyncio.get_running_loop() 成功）。纯同步脚本走
:159-173 分支，不触发本缺陷。

代价：

• 生命周期终态造假：on_flow_end 在流程尚未跑完时按 result=None, error=None 触发，看起来像
  「正常结束但无结果」。依赖该回调落库终态、发通知、释放租约 / 配额、关闭外部会话的宿主会提前收尾，
  得到错误终态记录，而流程还在后台继续改状态。
• 放弃 ≠ 取消：消费者 break 后流程仍在后台线程跑完。有副作用且不可回滚的节点
  （写文件、发 HTTP、扣费、调外部 agent）照常执行；close() 最多阻塞 5s，长流程则在之后继续，
  宿主以为已经停下。
• 同实例重入双跑：_running 已复位，同一 FlowExecution 可再次 run_compatible，两次执行
  并行（实测节点执行计数 40），共享 _ctx 与 callback 管理，交叉写上下文、回调次序错乱。
• 仅有的一条信号是无标识 warning：close() 之后唯一的提示是 plaita.core.async_utils 的
  logger.warning（async_utils.py:213），消息里既无 flow id 也无 execution id；宿主的 API 返回值与
  回调都无法反映「后台仍在跑」。

不修的后果：任何宿主按文档用 flow.debug() 做「单步看到一半就退出」，都会稳定得到
「假的流程结束 + 后台继续烧资源与副作用 + 实例可重入双跑」三连，且只能靠一条无标识的 warning 察觉。

建议

• 首选（停止信号）：plaita/core/async_utils.py:174-216 的线程分支引入停止位——消费者 finally
  先 set() 一个 threading.Event 再 join()；_drive() 的 while True 顶部检查该 Event，
  置位时跳出并 loop.run_until_complete(agen.aclose())，让 async generator 的 finally
  （含 _flow_session_scoped_agen 里 close_flow_session）在线程内真正执行。join 超时不应只
  告警，应视为未收敛（至少把执行 id / flow id 带上，并按错误处理）。
• 背压/统一语义：把无界 _queue.Queue()（:177）改为有界（如 maxsize=1）以给生产者施加背压，
  或让线程分支与 :159-173 的消费者驱动分支共用同一「按需拉取」语义，消除两分支行为差异。
• 复位时机：plaita/core/executor_compat.py:34-36 的 _running 复位应改为「确认后台已停止」
  之后，而不是在消费者 finally 里先发 on_flow_end 再无条件置 False。
• 验收条件（可执行、可断言）：
  1. 新增测试（建议放 tests/unit/test_async_utils.py，并补一个走 run_compatible 的
     tests/unit/test_executor.py 用例）：在 asyncio.run() 内用多节点 lazy 流程，消费 1 步后
     close()；断言 close() 返回后节点执行计数不再增长（记录计数、time.sleep 或
     await asyncio.sleep 一段后再断言相等）。
  2. 断言 on_flow_end 不会在「流程仍有节点未执行」时触发；若确需提前收尾，则其
     result 不得为 None 伪装成正常结束。
  3. 断言 close() 返回后立即对同一 FlowExecution 实例再次 run_compatible(..., False)
     不再与后台残留执行并发（节点执行计数不出现并发交叉）。
  4. 现有 tests/unit/test_executor.py:105 的 test_generator_flow_end_fires_only_after_consumption
     只覆盖无 loop 分支（成功路径），需补上「已有运行 loop」分支的对照断言。

边界

• 只处理「同步 lazy 入口在调用线程已有运行 loop 时的线程分支不可停止」这一个根因。异步 lazy 路径
  （_drive_lazy_async）与无 loop 分支（:159-173，语义正确）不在本单范围。
• 不改 on_flow_end 的语义定义，也不改 _error_normalization.emit_flow_end_on_close 的错误归一化
  逻辑；本单只要求它「不早于真实停止」触发。
• 不要求为 console 增加 lazy 入口，也不改 dry-run（当前是 eager，dryrun.py:324）；生产代码当前
  无 lazy 调用点，本单定位为库嵌入 / 宿主可达的正确性缺口。
• 不顺手重构 run_compatible / arun_compatible 整体的重入守卫与 _running acquire 设计，
  仅调整「lazy 结束时的复位时机」这一处。
• 与既有单 #34（集群档跨进程取消传播）不同层：那是 worker 与执行之间的跨进程取消件缺失，
  本单是 core 内 sync↔async 桥接线程的停止语义缺失；两者不互为补充评论。