app/orchestrator.py

"""语义级编排：在 client 与 adapter 之间拼接 tool directive，可选拒绝重试。

有 tools 时 prompt 结构固定为：
  [TOOL PROTOCOL 全文 + tools 列表]  ← 始终最顶端
  [base_prompt: system / history / user]
  [TOOL PROTOCOL REMINDER]           ← 文末再钉一次，抗长历史 recency

拒绝检测（``refusal_detect``）默认关：真流式透传。
开启后：整轮 buffer → 检测拒绝 → 换 retry 变体重试（与 deps 账号换号正交）。
"""
from __future__ import annotations

import logging
import sys
from collections.abc import AsyncIterator
from typing import Any

from app.events import IREvent
from app.refusal import is_refusal
from app.tools import (
    ToolDef,
    build_tool_directive,
    build_tool_tail_reminder,
    missing_required,
    parse_tool_calls,
)

logger = logging.getLogger(__name__)

def _compose_prompt(base_prompt: str, tools: list[ToolDef], *, variant: str = "default") -> str:
    """有 tools：directive 置顶 + base + tail；无 tools：原样。"""
    if not tools:
        return base_prompt
    head = build_tool_directive(tools, variant=variant)
    tail = build_tool_tail_reminder(tools) if variant == "default" else ""
    # retry 变体用整段替换头，仍钉 tail 提醒
    if variant != "default":
        tail = build_tool_tail_reminder(tools)
    return f"{head}\n\n{base_prompt}{tail}"


async def _collect_round(
    client: Any, prompt: str, model_id: str | None, tools: list[ToolDef] | None = None,
    **kw: Any,
) -> tuple[list[IREvent], str, bool]:
    """跑一轮 client.stream，收集全部 IREvent 并拼接 text。``kw`` 透传上游。"""
    events: list[IREvent] = []
    parts: list[str] = []
    had_error = False
    # 2026-08-08 实测（生图链路）：response.completed（→finish 事件）后流还没结束，
    # 图片 markdown 在更晚的 response.output_text.done 顶层 text 里。若在 finish 就
    # break，图片 URL 全被截断（只有 tool_call 围栏无媒体）。故 finish 只计数，
    # 继续消费到 async for 自然耗尽（客户端流永远不会无限：上游 SSE 有明确收尾）。
    async for ir in client.stream(prompt, model_id=model_id, tools=tools, **kw):
        events.append(ir)
        if ir.kind == "error":
            had_error = True
            break
        if ir.kind == "text" and ir.text:
            parts.append(ir.text)
        # finish 不 break；图片 done 事件在其后
    return events, "".join(parts), had_error


def _count_history_tool_calls(base_prompt: str, known: set[str]) -> dict[str, int]:
    """扫 base_prompt 里 ``[tools]`` 块的 ``name: <工具名>`` 行，统计每个已知名出现次数。

    extract_user_prompt 把每轮工具调用拍成 ``[tools]`` 块，块内每条形如：
        [call_x]
        name: todowrite
        arguments: {...}
    这里只数 ``name:`` 行（不计 result 行），用于判断"历史里某工具被连调了几次"。
    opencode 死循环现场：历史清一色 todowrite，无 glob/read/bash 等干活工具。
    """
    counts: dict[str, int] = {name: 0 for name in known}
    for line in base_prompt.splitlines():
        s = line.strip()
        # ``name: X`` 形态（[tools] 块内调用条目）；排除结果块可能的杂行
        if s.startswith("name:") and s.split(":", 1)[1].strip() in counts:
            name = s.split(":", 1)[1].strip()
            counts[name] += 1
    return counts


def _latest_is_tool_result_only(base_prompt: str) -> bool:
    """最新一轮是工具结果、其后无新的 ``[user]`` 请求块？

    extract_user_prompt 拍出的历史块顺序为各 ``[system]/[user]/[assistant]/[tools]`` 段；
    找最后出现的 ``[user]`` 顶层标记，若它出现在最后一个 ``[tools]`` 块**之前**（或根本没有
    ``[user]``），说明末尾是工具结果、没有用户新需求 —— 模型应"继续收尾"而非"重启原任务"。

    注意只认顶层 ``\\n[user]``（块首），避开 args 里偶然出现的同类字符串（arguments 是单行 JSON，
    其后跟 ``\\n---`` 不会以裸 ``\\n[user]\\n`` 起块）。
    """
    stripped = base_prompt.rstrip()
    # 末尾不是 [tools] 块（比如以 [assistant] 文本结尾）则不算"纯工具结果轮"
    last_tools = stripped.rfind("\n[tools]")
    if last_tools < 0 and not stripped.startswith("[tools]"):
        return False
    last_user = stripped.rfind("\n[user]")
    # 最后的 [user] 必须在最后的 [tools] 之后才算"有新请求"；否则视为末尾是工具结果
    return last_user < last_tools or (last_tools >= 0 and last_user < 0)


def _dup_call_correction(counts: dict[str, int], known: set[str], dup_limit: int) -> str:
    """判定"只写计划/只探查、不实质推进"的死循环，返回纠正提示，否则空串。

    两类死循环都治：
    1. 全程没干过活（连探查都没有），却反复写计划(todowrite 等) >=limit -> 纠正"别写计划了
       用 read/glob/bash 干活"。
    2. 已经在探查（glob/grep/list 反复）但从未实质读取/读取结果（read/bash/edit/write
       一次都没有）-> 纠正"别再 glob 了，挑文件用 read 读或看 bash 结果，直接给结论"。
       gpt-5.5 卡在反复 glob/git status 列目录、不读文件内容、反烧 token，这一类专治它。
    """
    if dup_limit <= 0:
        return ""
    survey = {"glob", "grep", "list", "bash", "task"}  # 列目录/搜文件名/跑命令列状态——探查性质，不等于读到文件内容
    real = {"read", "edit", "write", "str_replace_editor"}  # 真正读取/写入文件原始内容的工具（read 是核心信号）
    planish = {"todowrite", "todo", "plan"}
    has_any = any(counts.get(h, 0) > 0 for h in survey | real)
    has_real = any(counts.get(h, 0) > 0 for h in real)
    plan_count = sum(counts.get(p, 0) for p in planish)
    real_count = sum(counts.get(r, 0) for r in real)
    survey_count = sum(counts.get(s, 0) for s in survey)

    # 情况零（兜底之王）：计划工具计数远超"所有非计划动作之和" -> 不管有没有 read，计划已刷爆、
    # 实际推进几乎为零。专治 gpt-5.5 的"狂写 todowrite 50 次 + 象征性 glob 2 次、从不 read"形态：
    # 情况一(要 has_any=False)因象征性探查漏判、情况二(要 survey>=4)因探查太少漏判、
    # 情况三(要 has_real)因从不 read 漏判 —— 三道全漏，唯有这里按"计划占绝对多数"兜底。
    non_plan_total = real_count + survey_count
    if plan_count >= max(dup_limit + 3, 5) and plan_count > non_plan_total + 2:
        lead_plan = "todowrite"
        for p in planish:
            if counts.get(p, 0) == plan_count and counts.get(p, 0) > 0:
                lead_plan = p
                break
        read_hint = "用 read 读取核心文件内容" if real_count == 0 else "基于已读到的内容"
        return (
            "\n[system]\n"
            f"提醒：`{lead_plan}` 已累计被调用 {plan_count} 次，而真正读取/执行的动作"
            f"（read/edit/write/bash/glob 等合计）仅 {non_plan_total} 次。计划早已完成，"
            f"继续写 `{lead_plan}` 是空转烧 token，不会再推进任务。\n"
            f"立即停止调用 `{lead_plan}` 及任何计划类工具——{read_hint}，"
            f"然后直接给出最终分析/结论/回答，不要再输出任何 `{lead_plan}` 调用。"
        )

    # 情况一：全程没干过活（连探查都没），只有计划工具反复 -> 停止写计划、去干活
    if not has_any:
        dup_name = None
        for name, c in counts.items():
            if name in planish and c >= dup_limit:
                dup_name = name
                break
        if dup_name:
            n = counts[dup_name]
            return (
                "\n[system]\n"
                f"提醒：历史中已多次成功调用 `{dup_name}`（{n} 次），"
                f"任务清单/计划已经建立并完成，不要再重复调用 `{dup_name}`，也不要再写计划或重述步骤。\n"
                f"立刻执行真正的任务：用 read/glob/grep/bash 等工具读取文件、搜索代码、运行命令，"
                f"拿到结果后直接给出最终分析/结论。不要输出任何新的 `{dup_name}` 调用。"
            )
        dup_name = None
        for name, c in counts.items():
            if c >= dup_limit:
                dup_name = name
                break
        if dup_name:
            n = counts[dup_name]
            return (
                "\n[system]\n"
                f"提醒：`{dup_name}` 已被连续调用 {n} 次且没有实质进展。"
                f"不要再重复调用 `{dup_name}`——换个实际动作：用 read 读取关键文件、"
                f"用 bash 运行命令看结果，然后直接给出分析/结论。"
            )

    # 情况二：已在探查(glob/grep/list 反复)但从未实质读取(read/bash 一次都没) -> 别再列了去读
    survey_total = sum(counts.get(s, 0) for s in survey)
    g = counts.get("glob", 0)
    if survey_total >= max(dup_limit + 2, 3) and not has_real:
        lead = "glob" if g == max(counts.get(s, 0) for s in survey) else "列目录/搜索"
        return (
            "\n[system]\n"
            f"提醒：已反复调用探查类工具（{lead} 等，累计 {survey_total} 次）列出目录/搜索，"
            f"但从未用 read 读取任何文件内容、也没用 bash 运行命令看结果。\n"
            f"目录结构已经清楚，不要再继续 glob/搜索——立刻从已列出的文件里挑 2-3 个核心文件"
            f"用 read 读取其内容（或用 bash 跑出结果后直接看输出），基于读到的内容给出最终分析/结论。"
        )

    # 情况三：已干过活(read 等真实工作)却又大量反复重写计划 -> 读完了就给结论，别再回去重写计划。
    # 实战现场：gpt-5.5 read 完 3 个文件后又切回 todowrite 重写计划，read 弄到的内容白读。
    # 情况一要"无任何 survey/real"才触发、情况二要"无 real"才触发，二者在"已 read 但仍狂写计划"
    # 时都漏判 —— 这一情况专兜底：已真实工作过、计划工具计数远超实际工作 → 读完了给结论。
    if has_real and plan_count >= dup_limit + 3 and plan_count > real_count + 1:
        lead_plan = "todowrite"
        for p in planish:
            if counts.get(p, 0) == plan_count and counts.get(p, 0) > 0:
                lead_plan = p
                break
        return (
            "\n[system]\n"
            f"提醒：你已用 read/bash 等实际读取/执行过内容（{real_count} 次真实工作），"
            f"但 `{lead_plan}` 等计划工具仍累计被调用 {plan_count} 次。\n"
            f"文件内容已经读到，不要再回去重写或更新计划清单——立即基于刚才读到的内容"
            f"直接给出最终分析/结论/回答。不要再输出任何新的 {lead_plan} / 计划类调用。"
        )

    return ""

async def stream_with_retry(
    client: Any,
    base_prompt: str,
    tools: list[ToolDef],
    model_id: str | None = None,
    *,
    max_retries: int | None = None,
    **kw: Any,
) -> AsyncIterator[IREvent]:
    """拼 tool directive 后驱动 client.stream；可选拒绝检测换变体重试。

    - ``base_prompt`` **不含** directive；有 tools 时 head+base+tail。
    - ``refusal_detect=false``（默认）或无 tools：真流式透传。
    - ``refusal_detect=true`` 且有 tools：buffer 一轮；命中拒绝则换 retry 变体重试。
    - ``**kw``（如 ``image_model``）原样透传给上游 client（生图链路）。
    """
    has_tools = bool(tools)
    if max_retries is None:
        from app.config import get_settings

        settings = get_settings()
        max_retries = settings.tool_call_retries if settings.refusal_detect else 0

    # native 模式：上游原生支持 function-calling → 不注入 directive，tools 直通上游。
    # 上游服务端还会自动注入自己的工具（AnumaSearchMCP-*、memory_vault_* 等），客户端并不认识；
    # 且模型会发出 arguments 为 {} 的空壳调用（缺 schema 必填字段，opencode 校验即报错）。
    # 缓冲整轮：非客户端工具名 / 全坏调用（缺必填参数）拦截掉，注入错误提示重试。
    if settings.upstream_strategy == "native" and has_tools:
        known = {t.name for t in tools}
        # 注意：函数开头的 max_retries 已被 refusal_detect 逻辑压成 0（refusal_detect=false 时），
        # 不能用于 native 分支；这里重新从配置读取独立重试预算。
        retries = settings.tool_call_retries or 2
        max_attempts = 1 + retries
        prompt = base_prompt
        # 历史工具调用计数：扫 base_prompt 里 [tools] 块的 name 行。dup_limit 软纠正与
        # forge_limit 硬禁用两道防线共用这一份 counts，不重复计算。
        counts = _count_history_tool_calls(base_prompt, known)
        # 真实推进判定：read ≥2 次才算真干活。1 次 read 可以是试探糊弄——实战见模型
        # read 1 次后继续狂刷 glob/bash 25+ 次、has_real=True 让硬禁整块停火，必须堵上。
        read_count = sum(counts.get(r, 0) for r in
                         {"read", "readfile", "fs_open_file", "Read"})
        has_real = read_count >= 2
        # 历史重复调用纠正：扫历史里 [tools] 块的 name 行；若某个工具（尤其在
        # opencode 里反复写的 todowrite）历史已累计 ≥ dup_limit 次且没干过活，
        # 追加"停止重复、立即干活"提示，破 gpt-5.5 的 todowrite 死循环。
        # 第一轮/正常多轮历史这里返回 ""，不注入。
        dup_limit = int(getattr(settings, "tool_call_dup_limit", 2) or 0)
        if dup_limit > 0:
            corr = _dup_call_correction(counts, known, dup_limit)
            if corr:
                prompt = base_prompt + corr
                dup_name = max(counts, key=counts.get)
                logger.warning(
                    "history has repeated tool call `%s` (x%d); "
                    "injecting 'stop repeating, do real work' correction",
                    dup_name, counts[dup_name],
                )
            # STATE NOTICE（借 qwen2API 思路）：最新一轮是工具结果、其后无 [user] 新请求时，
            # 钉一句"这是工具结果不是新请求，从结果继续收尾、别重启原任务"——专治模型把结果误
            # 当新任务从头来 / 读完文件又回 todowrite 重写计划去。与情况三正交、协同封死回路。
            if _latest_is_tool_result_only(base_prompt):
                prompt = prompt + (
                    "\n[system]\n"
                    "提示：最新输入是工具调用的结果，不是用户的新请求。请基于该结果继续当前任务或"
                    "直接收尾给结论，不要把结果当成新任务从头开始，也不要重新建立/重写计划清单。"
                )
                logger.info("latest context is a tool result; injecting 'continue, don't restart' notice")
        # 贪读收口（soft）：read 是真实工作不硬禁，但子代理会"贪读不收尾"——一直 read 不返回结论
        # 给上层，烧光上下文照样卡死。read 累计远超其它动作(>=read_cap 且 >= non_read×2)时注入
        # "读够了立即给结论"软纠正。 配额天花板把 bash/glob/grep 剥光后 non_read 趋近 0，此条件自然
        # 命中——正好兜底"剥光只剩 read"的终态。
        read_cap = int(getattr(settings, "tool_read_cap", 30) or 0)
        if read_cap > 0 and read_count >= read_cap:
            non_read = sum(
                c for n, c in counts.items()
                if n not in {"read", "readfile", "fs_open_file", "Read"}
            )
            if read_count >= non_read * 2:
                prompt = prompt + (
                    "\n[system]\n"
                    f"提醒：你已用 read 累计读取 {read_count} 个文件（远超其它动作合计 {non_read} 次）。"
                    "信息已经足够，继续 read 是空转烧 token、不会带来新结论。\n"
                    "立即停止读取——基于已读到的文件内容直接给出最终分析/结论/回答，"
                    "不要再发起新的 read 或任何探查工具调用，把结论返回给上层。"
                )
                logger.warning(
                    "history shows greedy read (read=%d, non_read=%d, >=tool_read_cap %d); "
                    "injecting 'stop reading, give conclusion' correction",
                    read_count, non_read, read_cap,
                )
        # 硬禁用（tool forge limit，物理防线）：某工具历史空刷 ≥ forge_limit 次且无真实推进
        # （read/edit/write 一次都没成功）-> 本轮从 tools 列表物理剔除它，上游拿不到 schema、
        # 模型物理上无法再生成该调用；known 集合也剔，模型若仍文本输出 -> 进 unknown 拦截重试，
        # 文案"工具不存在、可用工具仅限 read/..."信号最硬。逐步勒索：逃到别的工具，那个工具也会
        # 涨到阈值被依次禁，最终只剩 read。has_real=True（真有读/写推进）时一律不禁，避免误伤
        # 合理多步任务（如 bash 跑测试 8 次）。
        forge_limit = int(getattr(settings, "tool_forge_limit", 8) or 0)
        disabled: set[str] = set()
        if forge_limit > 0:
            # read 给的"边读边刷"配额：read×3，但设天花板 forge_limit×2。
            # 无天花板时模型用少量 read 无限买配额（实战：read 9 撑 27 配额、glob/bash 贴线
            # 无限刷）——天花板保证任何工具刷过 2×forge_limit 必被禁，read 再多也救不了。
            quota = min(read_count * 3, forge_limit * 2)
            for name, c in counts.items():
                if name in {"read", "readfile", "fs_open_file", "Read",
                            "edit", "write", "str_replace_editor"}:
                    continue  # 真实工作类工具永不硬禁
                if c >= forge_limit and (not has_real or c > quota):
                    # 要么全程无真实读取；要么"边 read 边狂刷"——该工具空刷远超配额(>quota)
                    disabled.add(name)
        if disabled:
            logger.warning(
                "history shows empty-spin tool(s) %s (>=tool_forge_limit %d, read=%d); "
                "removing them from this round's tools",
                sorted(disabled), forge_limit, read_count,
            )
            active_tools = [t for t in tools if t.name not in disabled]
            active_known = {t.name for t in active_tools}
        else:
            active_tools = tools
            active_known = known
        for attempt in range(max_attempts):
            events, full_text, had_error = await _collect_round(
                client, prompt, model_id, tools=active_tools, **kw)
            if had_error:
                for ev in events:
                    yield ev
                return
            calls = parse_tool_calls(full_text, known_names=active_known)
            unknown = [c for c in calls if c.name not in active_known]
            known_calls = [c for c in calls if c.name in active_known]
            bad = [c for c in known_calls if missing_required(c, active_tools)]
            good = [c for c in known_calls if not missing_required(c, active_tools)]
            # 好坏混合时透出整轮（adapter 侧过滤空壳调用）；全坏或无 unknown 且无 good 才重试
            if not unknown and (good or not bad) or attempt + 1 >= max_attempts:
                for ev in events:
                    yield ev
                return
            if unknown:
                names = sorted({c.name for c in unknown})
                print(
                    f"[orchestrator] blocked upstream-injected tool(s): {names}; retry "
                    f"{attempt + 1}/{max_attempts - 1}",
                    file=sys.stderr,
                )
                err = (
                    "\n[user]\n"
                    "你刚才调用的工具不存在于当前环境，调用失败且不会有任何结果："
                    f"{'、'.join(names)}。\n"
                    f"本环境可用工具仅限：{', '.join(sorted(active_known))}。\n"
                    "需要读文件、跑命令、搜代码或做其他操作时，请立即改用上述可用工具调用，"
                    "不要再次调用不存在的工具。"
                )
            else:
                names = sorted({c.name for c in bad})
                print(
                    f"[orchestrator] blocked empty-args tool call(s): {names}; retry "
                    f"{attempt + 1}/{max_attempts - 1}",
                    file=sys.stderr,
                )
                err = (
                    "\n[user]\n"
                    "你刚才发出的工具调用参数不完整（缺少必填字段），调用失败且不会有任何结果："
                    f"{'、'.join(names)}。\n"
                    "请根据工具的 JSON schema 补全 arguments 中的必填字段后重新调用，"
                    "不要发出空的 arguments 占位。"
                )
            prompt = prompt + err
        return

    # 默认路径：不 buffer，流式透传
    if not has_tools or max_retries <= 0:
        prompt = _compose_prompt(base_prompt, tools, variant="default")
        async for ir in client.stream(prompt, model_id=model_id, **kw):
            yield ir
            if ir.kind == "error":
                return
            # 2026-08-08 实测（生图链路）：response.completed（→finish）后流还没结束，
            # 图片 markdown 在更晚的 response.output_text.done 顶层 text 里。生图时若在
            # finish 就 return，图片 URL 全被截断（只剩 tool_call 围栏无媒体）。finish 不
            # 截断，继续消费到 async for 自然耗尽（与 _collect_round 同款逻辑）。
        return

    # 拒绝检测路径：buffer + 换变体重试
    known = {t.name for t in tools}
    max_attempts = 1 + max_retries
    chosen: list[IREvent] = []
    for attempt in range(max_attempts):
        variant = "retry" if attempt > 0 else "default"
        prompt = _compose_prompt(base_prompt, tools, variant=variant)
        events, full_text, had_error = await _collect_round(client, prompt, model_id)
        if had_error:
            for ev in events:
                yield ev
            return
        chosen = events
        if parse_tool_calls(full_text, known_names=known):
            break
        if not is_refusal(full_text, has_tools=True):
            break
        if attempt + 1 >= max_attempts:
            break
        print(f"[orchestrator] refusal detected (variant={variant}); retry", file=sys.stderr)
    for ev in chosen:
        yield ev