butler/gateway/task_milestone.py

Rama main · réplica de lectura del repositorio original.

"""Antigravity-style task_boundary milestones for long WeChat turns."""

from __future__ import annotations

import time
from typing import TYPE_CHECKING, Any

from butler.utilities.env_parse import env_truthy, float_env, int_env
from butler.defaults.env_defaults import (
    GATEWAY_TASK_MILESTONE_DEFAULT,
    GATEWAY_TASK_MILESTONE_SECONDS,
    GATEWAY_TASK_MILESTONE_MAX,
)

if TYPE_CHECKING:
    from butler.gateway.outbound_bridge import GatewayOutboundBridge


def task_milestone_enabled() -> bool:
    from butler.gateway.task_milestone_ops import delegate_progress_notify_enabled_safe

    if delegate_progress_notify_enabled_safe():
        return True
    return bool(env_truthy("BUTLER_GATEWAY_TASK_MILESTONE", default=GATEWAY_TASK_MILESTONE_DEFAULT))


def task_milestone_min_seconds() -> float:
    try:
        return float(max(30.0, float_env("BUTLER_GATEWAY_TASK_MILESTONE_SECONDS", GATEWAY_TASK_MILESTONE_SECONDS)))
    except ValueError:
        return float(GATEWAY_TASK_MILESTONE_SECONDS)


def build_milestone_text(bridge: "GatewayOutboundBridge", *, elapsed: int) -> str:
    """Format: 阶段名 | 正在做什么 | 下一步"""
    phase = "处理中"
    doing = "分析并调用工具"
    nxt = "完成后回复摘要；可发 /health"

    if bridge.delegate_role:
        phase = f"委派·{bridge.delegate_role}"
        doing = "子代理执行项目任务"
        nxt = "完成后推送摘要"
    elif bridge.workflow_name:
        phase = f"工作流·{bridge.workflow_name}"
        step = bridge.workflow_step or ""
        if bridge.workflow_step_total > 0:
            doing = f"步骤 {bridge.workflow_step_index}/{bridge.workflow_step_total} {step}".strip()
        else:
            doing = step or "执行工作流步骤"
        nxt = "工作流结束后汇报"
    elif bridge.last_tool_name:
        phase = f"工具·{bridge.last_tool_name}"
        doing = "等待工具结果"
        nxt = "继续推理或委派"

    preview = (getattr(bridge, "_stream_preview", "") or "").strip()
    if preview:
        snippet = preview[-80:].replace("\n", " ")
        doing = f"{doing} …{snippet}"

    return f"【进度】{phase} | {doing} | {nxt}（约 {elapsed} 秒）"


def task_milestone_max_per_turn() -> int:
    try:
        from butler.gateway.completion_notify import delegate_progress_notify_enabled

        default = GATEWAY_TASK_MILESTONE_MAX if delegate_progress_notify_enabled() else 1
        result: int = int_env("BUTLER_GATEWAY_TASK_MILESTONE_MAX", default, min=1, max=8)
        return result
    except ValueError:
        return int(GATEWAY_TASK_MILESTONE_MAX)


def maybe_schedule_task_milestone(bridge: Any) -> None:
    """Send periodic milestones after ack threshold (delegate progress may repeat)."""
    if not task_milestone_enabled():
        return
    if getattr(bridge, "_closed", True) or getattr(bridge, "_final_sent", False):
        return
    if not getattr(bridge, "_ack_sent", False):
        return
    max_n = task_milestone_max_per_turn()
    sent = int(getattr(bridge, "_task_milestone_count", 0) or 0)
    if sent >= max_n:
        return
    started = getattr(bridge, "_started_at", 0.0) or 0.0
    elapsed = int(time.monotonic() - started) if started else 0
    min_elapsed = int(task_milestone_min_seconds()) * max(1, sent)
    if elapsed < min_elapsed:
        return

    text = build_milestone_text(bridge, elapsed=elapsed)
    if sent > 0:
        text = f"{text}\n（进度 {sent + 1}/{max_n} · 可发 /停止 中断）"
    if bridge.schedule_supplementary_reply(text, kind="task_milestone"):
        bridge._task_milestone_count = sent + 1
        if sent == 0:
            bridge._task_milestone_sent = True


__all__ = [
    "build_milestone_text",
    "maybe_schedule_task_milestone",
    "task_milestone_enabled",
    "task_milestone_max_per_turn",
    "task_milestone_min_seconds",
]