fix(core): distributed 档从不合并/解析 flow.timeout，步进路径又丢 max_timeout_ms——流程级超时在 worker/console 生产路径完全失效，ISO 时长炸成误导性节点 TypeError

Start with _prepare_strategy and the DISTRIBUTED entry in plaita/core/executor.py (around lines 382-453), where flow.timeout is parsed and merged for NORMAL/GENERATOR but not for DISTRIBUTED. Then check the missing max_timeout_ms argument at plaita/core/strategies.py:273 and the unparsed caller timeout in plaita/core/runner.py:286-304. Done means the distributed path honors flow.timeout and ISO caller timeouts, and the existing test at tests/unit/test_flow_distributed.py:151 is replaced with one that checks the timeout is actually applied.

优先级 P1 · 依赖：无

基线 HEAD = 65f48f2

Summary

Flow.timeout（流程级超时）只在 NORMAL / GENERATOR 路径被合并与解析；DISTRIBUTED 路径
（worker 与 console 的生产链路）从不读它，且步进路径从第 2 步起丢掉调用方给的最大预算。
调用方若按文档传 ISO 时长字符串，还会在首节点炸成与真因无关的 TypeError。

证据 1 —— 只有 NORMAL/GENERATOR 的入口做「flow.timeout 与调用方 timeout 取更严者」

plaita/core/executor.py:382-397：

    def _prepare_strategy(self, flow, lazy, args, kwargs):
        """Sync orchestration: clean, flow_start, setup_flow. Returns the
        strategy's awaitable/async-generator for later driving."""
        self._ensure_flow_resolved(flow)
        self.clean()
        self.callback_manager.on_flow_start(flow)
        self._ctx.setup_flow(flow, args, kwargs)
        # flow 级超时与调用方超时取更严者
        timeout_ms = self._merge_timeout_ms(
            self._parse_timeout(flow.timeout),
            self._parse_timeout(self.timeout),
        )
        strategy = self._strategies[
            ExecutionMode.GENERATOR.value if lazy else ExecutionMode.NORMAL.value
        ]
        return strategy.execute(flow, self._ctx, self._runner, self.callback_manager, None, timeout_ms)

证据 2 —— DISTRIBUTED 入口把调用期原始 timeout 直送策略，既不 parse 也不 merge

plaita/core/executor.py:426（形参）与 plaita/core/executor.py:449-453（分发）：

        timeout: Optional[int] = None,

            coro = self._strategies[ExecutionMode.DISTRIBUTED.value].execute(
                flow, self._ctx, self._runner, self.callback_manager, params, timeout,
                saved_context=saved_context, resume_type=resume_type, resume_data=resume_data,
                **seed,
            )

反证（flow.timeout 在 core 全仓只有一处被读，且不在 distributed 路径上）：

$ grep -rn "flow\.timeout" --include='*.py' plaita/ plaita-console/
plaita/core/executor.py:391:            self._parse_timeout(flow.timeout),

证据 3 —— 该值被当作 max_timeout_ms 一路下传，只有首节点吃到

plaita/core/strategies.py:212-214：

    async def execute(self, flow, context, runner, callback_manager, params=None, timeout_ms=None, **options):
        saved_context = options.get("saved_context")
        resume_type = ResumeType.coerce(options.get("resume_type", "continue"))

plaita/core/strategies.py:266-267：

        current_node, result, branch = await self._determine_current_node(
            flow, context, runner, callback_manager, max_timeout_ms=timeout_ms)

plaita/core/strategies.py:302-308：

    async def _start_new_flow(self, flow, context, runner, callback_manager, max_timeout_ms=None):
        start_node = flow.start_node
        if not start_node:
            return None, None, None

        result, branch = await runner.run_node(
            flow, start_node, callback_manager=callback_manager, max_timeout_ms=max_timeout_ms)

证据 4 —— 第 2 步起漏传第 6 参（max_timeout_ms），预算归零

plaita/core/strategies.py:273：

        return await self._execute_current_node(flow, context, runner, callback_manager, current_node)

而它调用的定义在第 322 行确实收这个参数（plaita/core/strategies.py:322）：

    async def _execute_current_node(self, flow, context, runner, callback_manager, current_node, max_timeout_ms=None):

证据 5 —— 该值参与减法，未做类型归一；只有节点级 timeout 被解析

plaita/core/runner.py:286-296：

        total_timeout_ms: Optional[int] = None
        if config_timeout:
            total_timeout_ms = config_timeout
            if max_timeout_ms:
                total_timeout_ms = min(config_timeout, max_timeout_ms)
        elif max_timeout_ms:
            total_timeout_ms = max_timeout_ms

        time_limit_by_flow = (not config_timeout and max_timeout_ms is not None) or (
            max_timeout_ms is not None and config_timeout and max_timeout_ms < config_timeout
        )

plaita/core/runner.py:302-304：

                if total_timeout_ms is not None:
                    elapsed_ms = int((time.time() - start_time) * 1000)
                    remaining_timeout = max(0, total_timeout_ms - elapsed_ms)

plaita/core/runner.py:259：

        config_timeout = self._parse_timeout(node.timeout)

plaita/core/runner.py 内 _parse_timeout 的全部调用点（grep -n "_parse_timeout\|max_timeout_ms" plaita/core/runner.py）：

221:        max_timeout_ms: Optional[int] = None,
232:        result = await self._execute_with_retry(flow, node, max_timeout_ms)
252:        max_timeout_ms: Optional[int],
259:        config_timeout = self._parse_timeout(node.timeout)
289:            if max_timeout_ms:
290:                total_timeout_ms = min(config_timeout, max_timeout_ms)
291:        elif max_timeout_ms:
292:            total_timeout_ms = max_timeout_ms
294:        time_limit_by_flow = (not config_timeout and max_timeout_ms is not None) or (
295:            max_timeout_ms is not None and config_timeout and max_timeout_ms < config_timeout

—— 即：只有 node.timeout 过 _parse_timeout，来自调用方的 max_timeout_ms 原样进入 min() / -。

证据 6 —— 三个 worker 生产入口与三个 console 入口都不传 timeout=

$ grep -rn "run_distributed" --include='*.py' plaita/ plaita-console/
plaita/core/_error_normalization.py:21:difference is a documented external contract (see ``FlowExecution.run_distributed``
plaita/core/executor.py:9:``run_compatible`` / ``arun_compatible`` / ``run_distributed``)。
plaita/core/executor.py:339:            # 统一走 run_distributed, 避免与 run_distributed 维护两份几乎相同的
plaita/core/executor.py:343:            # 应直接实例化 FlowExecution 并复用同一实例调用 run_distributed()，
plaita/core/executor.py:345:            return execution.run_distributed(
plaita/core/executor.py:456:            # 历史上 run_distributed 把任何异常（含具体的 FlowExecutionException 子类）
plaita/core/executor.py:466:    def _run_distributed(self, flow, params=None, timeout=None, context=None,
plaita/core/executor.py:474:        return self.run_distributed(
plaita/server/flow_worker.py:355:    分布式路径上 ``run_distributed`` 把一切异常归一化为 ``FlowErrorException``
plaita/server/flow_worker.py:488:    流程工作器：把 ``run_distributed`` 与 ExecutionStorage / FlowStorage / EventBus 串起来。
plaita/server/flow_worker.py:1081:        映射在 ``run_distributed`` **之前**认领（execution_id 在
plaita/server/flow_worker.py:1384:                result = execution.run_distributed(flow, params=params, execution_id=execution_id)
plaita/server/flow_worker.py:1415:            # 引擎自 run_distributed 归一化抛出（挂 __cause__）或协作节点自杀
plaita/server/flow_worker.py:1629:            # saved_context 由下方 run_distributed 携带。
plaita/server/flow_worker.py:1656:            # 直接使用 run_distributed 恢复执行
plaita/server/flow_worker.py:1657:            result = execution.run_distributed(
plaita/server/flow_worker.py:1841:                    result = execution.run_distributed(
plaita-console/backend/tests/console/test_local_cancel.py:44:    """可编排的假 FlowExecution：首次 run_distributed 长跑阻塞，等待放行。"""
plaita-console/backend/services/local_executor.py:363:            result = execution.run_distributed(flow, params=params)
plaita-console/backend/services/local_executor.py:365:            result = execution.run_distributed(
plaita-console/backend/services/local_executor.py:423:            result = execution.run_distributed(

去掉注释/文档命中后，真实调用点只有六处：worker 三处（:1384、:1657、:1841）、console 三处（local_executor.py:363、:365、:423），加上 executor.py:345（FlowExecution.run 的 distributed 分发，传的是调用期原始值）与 executor.py:474（_run_distributed 兼容包装，把 timeout= 原样转交）——无一处传 timeout=。

plaita/server/flow_worker.py:1657-1662 与 :1841-1846、plaita-console/backend/services/local_executor.py:365-370 三个 resume/step 调用点逐字如下（均无 timeout=）：

            result = execution.run_distributed(
                flow,
                saved_context=state.context,
                resume_type=resume_type,
                resume_data=data,
            )

                    result = execution.run_distributed(
                        flow,
                        saved_context=context,
                        resume_type="continue",
                    )

            result = execution.run_distributed(
                flow,
                saved_context=resume["context"],
                resume_type=resume["resume_type"],
                resume_data=resume.get("data"),
            )

flow_worker.py 全文 timeout 命中只有 drain/join/force_stop/watchdog，无任何流程级兜底：

$ grep -n "timeout" plaita/server/flow_worker.py
294:def _drain_timeout_from_env() -> float:
709:    # 一个节点一旦跑起来就打断不了——agentrun 默认 timeout_secs=1800，取消后
765:            watcher.join(timeout=2.0)
892:        —**flow 结束/挂起/异常三条路径都精确释放实例**，不再干等 AGS 侧 timeout
2344:            watchdog.join(timeout=2.0)
2683:                        t.join(timeout=1.0)
2890:        timeout = _drain_timeout_from_env()
2895:            timeout,
2897:        if timeout <= 0:
2898:            self._force_stop_after_drain(timeout)
2900:        timer = threading.Timer(timeout, self._force_stop_after_drain, args=(timeout,))
2905:    def _force_stop_after_drain(self, timeout: float = 0.0) -> None:
2915:            timeout,
2920:    def wait_for_idle(self, timeout: Optional[float] = None) -> bool:
2922:        deadline = None if timeout is None else time.time() + max(0.0, timeout)

证据 7 —— 文档承诺 ISO 时长与「取更严者 / 不会静默丢弃」

docs-site/docs/guide/error-handling.md:9：

在 `Flow` 上设 `timeout`，整个流程的累计执行时间不得超过该值。格式为 **ISO 8601 时长字符串**或**毫秒整数**：

docs-site/docs/guide/error-handling.md:41：

调用方传入的 `timeout`、节点自身 `timeout`、流程 `timeout` 之间取**更严（更小）者**生效，不会静默丢弃。流程级剩余时间会作为节点执行的最大预算下传。

证据 8 —— 既有测试是假守卫：只钉解析值，不验证超时被应用

tests/unit/test_flow_distributed.py:151-160：

    def test_timeout_handling(self):
        """测试超时设置"""
        flow = create_test_flow()
        flow.timeout = "PT0.1S"  # 设置100ms超时
        
        result = self.execution._run_distributed(flow, timeout=200)  # 使用更大的超时时间
        assert result["id"] == "node1"
        
        # 验证使用了较小的超时时间
        assert self.execution._parse_timeout(flow.timeout) == 100

实跑复现（原样命令 + 输出）

$ /tmp/plaita-audit-venv/bin/python /tmp/repro_pcf_f1.py
WARNING:plaita.core.runner:Sync node n1 timed out after 50ms (cancel_event set)
ERROR:plaita.core.executor:flow error: 执行节点n0出错了: TypeError: unsupported operand type(s) for -: 'str' and 'int'
=== spying on strategy pos6 (timeout_ms) ===
  NORMAL     pos6=7000
  DISTRIBUTED pos6=None

=== E2E: flow.timeout='PT0.05S', node sleeps 0.3s ===
normal       -> NodeTimeoutError: Timeout handler strategy for executing node n1 is abort
generator    -> completed in 0.00s WITHOUT honoring flow.timeout (result=generator)
distributed  -> completed in 0.31s WITHOUT honoring flow.timeout (result=dict)

=== E2E: distributed with ISO caller timeout='PT0.5S' ===
distributed -> FlowErrorException: 执行节点n0出错了: TypeError: unsupported operand type(s) for -: 'str' and 'int'
  __cause__: NodeExecutionError("执行节点n0出错了: TypeError: unsupported operand type(s) for -: 'str' and 'int'")

—— 同一 flow.timeout='PT7S'：NORMAL 策略收到 7000，DISTRIBUTED 收到 None；PT0.05S + 0.3s 节点：normal 抛 NodeTimeoutError，distributed 0.31s 正常完成。ISO 调用期 timeout 在 distributed 首节点炸 TypeError（节点名 n0 取决于流程命名）。

额外实跑：即使调用方硬塞毫秒整数，预算也只在每步内重置（第 2 步起完全失效）

$ /tmp/plaita-audit-venv/bin/python /tmp/repro_pcf_f1b.py
=== distributed, step-by-step, caller timeout=200ms int (ms budget applied? per step) ===
step 0 node=n1 elapsed=0.20s is_end=False
step 1 node=n2 elapsed=0.21s is_end=False
step 2 node=n3 elapsed=0.01s is_end=True

四节点流程（n1、n2 各睡 0.2s）配上 200ms 预算，两步都跑满且都"成功"——预算没有跨步累计（对照 NORMAL：strategies.py:145-148 用同一个 start_time 扣 remaining，strategies.py:158-159 在超时后 raise FlowTimeoutError()）。

回归（既有套件不覆盖本缺陷，全绿）：

$ /tmp/plaita-audit-venv/bin/python -m pytest tests/unit/test_clean_c1_state.py tests/unit/test_call_timeout.py tests/unit/test_flow_distributed.py tests/unit/test_concurrent.py -q
..............................                                     [100%]
30 passed, 6 subtests passed in 2.48s

影响

• 谁踩到：凡是把 plaita 当服务跑的部署——集群档 FlowWorker（flow_worker.py:1384 首跑、:1657 resume、:1841 步进）与 console 本地档（local_executor.py:363/365/423）——全都是 DISTRIBUTED 模式。流程作者在 flow 上写 "timeout": "PT30S" 是完全无效的：不会超时、不会报错、不会告警。
• 代价：一个会挂死或跑几小时的流程（agentrun 节点默认 timeout_secs=1800，见 flow_worker.py:709 注释）没有任何流程级天花板。worker 槽位被长期占用，租约/取消/看门狗都只能靠外部手段收尸；作者在本地 NORMAL 模式测过「超时生效」，上线后行为静默改变，属于最坏的一类「本地通过、生产失效」。
• 不是"仅仅是没生效"：调用方若照文档传 ISO 时长（run(..., timeout="PT0.5S")，FlowExecution.run 对库嵌入方公开），distributed 首节点直接抛 FlowErrorException: ... TypeError: unsupported operand type(s) for -: 'str' and 'int'，__cause__ 是 NodeExecutionError，错误信息把排查方向指向节点代码而不是超时管线，故障定位成本高。
• 错的语义：即使调用方传的是合法毫秒整数，也只有首节点受约束——distributed 逐步调用时每步重新计预算，等于把「整个流程累计执行时间不得超过该值」变成了「每步各自不得超过」。
• 不修的后果：流程级超时这一条公开契约在唯一的服务化路径上不存在，超时保护形同虚设，且以「流程正常完成」的形式掩盖。

建议

1. plaita/core/executor.py run_distributed：与 _prepare_strategy 对齐，把 flow.timeout 与调用期 timeout 一起经 self._parse_timeout 归一后取严——最小修法就是在 :448 之前插入一行
   timeout = self._merge_timeout(self._parse_timeout(flow.timeout), self._parse_timeout(self.timeout), timeout)
   之类（注意 run_distributed(timeout=...) 的既有语义是调用期预算，不要让 merge 覆盖调用方显式传入的更严值）。
2. plaita/core/strategies.py:273：补上第 6 参，max_timeout_ms 不能在第 2 步起丢失；同时确认 DISTRIBUTED 的逐步语义——若 worker 每步重开一次预算，必须由调用方（flow_worker.py / local_executor.py）从 checkpoint 恢复剩余预算，否则只能退化为「每次调用独立预算」，这与 docs-site/docs/guide/error-handling.md:41 的「累计执行时间」矛盾。
3. plaita/core/runner.py:286-296：对 max_timeout_ms 做防御性归一（非法类型 → 明确 ValueError），不要让 str 流到 :304 的减法，把类型错误伪装成节点错误。
4. 可选一致性：Flow.timeout 现为 Optional[str]（plaita/core/flow.py:50），error-handling.md:9 承诺的「毫秒整数」写法会被 pydantic 直接拒绝——与本条同属超时输入面，但不在本条修复范围，见 ## 边界。

验收条件

• 新增单测（放 tests/unit/test_flow_distributed.py，替换/加强现有 test_timeout_handling）：构造 flow.timeout="PT0.05S" + 一个睡 0.3s 的节点，断言 run_distributed 抛出超时异常（而非 0.31s 正常返回）；再断言 spy 到 distributed 策略第 6 个位置参数等于 50（不是 None、不是 "PT0.05S" 字符串）。
• 新增单测：run_distributed(flow, timeout="PT0.5S") 不得抛 TypeError；应等价于 500ms 预算。
• 新增回归：两节点各睡 0.2s、预算 200ms 的 distributed 逐步流程，应判定超时（当前实测两步都"成功"）。
• 复现脚本 /tmp/repro_pcf_f1.py 的形状应变为：DISTRIBUTED pos6=7000、distributed 分支抛超时、ISO 调用期 timeout 不再出现 TypeError。
• 既有套件不得回归：python -m pytest tests/unit/test_flow_distributed.py tests/unit/test_call_timeout.py tests/unit/test_concurrent.py tests/unit/test_clean_c1_state.py -q 保持全绿。

边界

• 只修「DISTRIBUTED 路径的流程级超时合并/解析 + 步进丢 max_timeout_ms」这一条链路。不顺手改 NORMAL/GENERATOR 的超时语义，也不改 _parse_timeout 的格式集合。
• 与 #44（CLOSED，test(strategies): test_strategies_mutations 的 mock 漏收 max_timeout_ms）不同根：#44 是测试替身的形参签名问题（mock 不收 max_timeout_ms 导致既有红），属于测试脚手架；本条是引擎侧 run_distributed 根本不产生该值、strategies.py:273 真实丢参。修本条时若顺手把 test_strategies_mutations 的 mock 签名对齐，可作为附带修正，但不要把 #44 当成本条的重复单。
• 不覆盖 Flow.timeout 字段类型不接受毫秒整数（plaita/core/flow.py:50 为 Optional[str]，传 50 会被 pydantic 以 string_type 拒绝，与 docs-site/docs/guide/error-handling.md:9 的「或毫秒整数」不一致）——独立文档/模型一致性问题，另单。
• 不覆盖 worker 侧的取消/租约/看门狗收尸链路（见 #34、#38、#50），本条只保证「引擎自己会按流程级超时中止」。
• 不重构 FlowExecution.run / run_distributed 的公开签名；timeout 参数语义（调用期预算、ISO 或毫秒）保持不变。