guanlan-src/app_ontology/app_ontology_guanlan/ontology/fast_agent.py

# -*- coding: utf-8 -*-
"""快答直连编排 (2026-08-28 用户: 还是太慢 → 结构性提速).

本地模型不再走 dsh/node/MCP 三层 IPC — ollama /api/chat + tools 同进程直连,
工具与闸 = mcp_server 的同一套函数 (零复制); dsh 路径保留给云端 deepseek 完整编排.
提速三源: ①砍 dsh 冷启+MCP 握手 ②台号预取 brief_pack 注入首轮 (省 1-2 轮工具往返)
③工具调用零 IPC。闸不放松: claim_check 未通过不出答案 (最多补 2 轮改写)。
跑法: .venv/bin/python -m src.ontology.fast_agent "问题" [model]"""
from src import paths as P
import json, pathlib, re, sys, time, urllib.error, urllib.request
from . import mcp_server as om
from src.windscada.audit import grounding

OLLAMA = 'http://127.0.0.1:11434/api/chat'
# ── 事实契约预取 (用户令 2026-09-06 "离线问答接契约, 校闸不过不出文") ──
# 2026-09-06 实测 (outputs/rudong/windscada/_demo/offline_qa_test_20260906.md): 三条问题经本体检索
# 全部答反 (契约 RD-2026-08-003 断裂仍在发生 ↔ 答"无记录"; RD-idx-075 点名 24 台 ↔ 答"0 台") ——
# 本体里没有契约那 119 条已审级结论, 模型再怎么检索也拿不到。故按问题关键词把 ≤5 条契约塞进首轮
# (与台号预取同一机制: 有唯一确定入口的问法, 预取即答案), 并进 facts 供接地闸认数字。
CONTRACT = P.guanlan() / 'facts_contract_v0.json'
_CON = {'key': None, 'claims': []}


def contract_claims():
    """契约 119 条 (内部面孔, 含场名台号 — 离线版是内部件). 按 mtime 缓存; 文件缺 → [] (不伪造)."""
    try:
        st = CONTRACT.stat()
    except OSError:
        return []
    key = (st.st_mtime_ns, st.st_size)
    if _CON['key'] != key:
        _CON.update(key=key, claims=json.loads(CONTRACT.read_text(encoding='utf-8')).get('claims', []))
    return _CON['claims']


_QSTOP = {'现在', '什么', '问题', '哪些', '怎么', '怎样', '有没', '没有', '是否', '多少', '情况', '引用', '说明', '编号', '结论', '这条', '关注', '需要', '请问', '意味', '限制', '条款', '出来', '看看', '一下', '系统', '机组', '本场', '目前'}   # '系统' 等泛词: v25 实逮 '[回改系统①]' 靠它抢到变桨题 top-1


def _sys_of(text):
    """文本 → 九系统 (None = 判不出/九系统无家). 复用 kb_ingest._sys_of: 词表取自原厂结构树 + RDS-PP 叶子,
    二级节点映射交付九系统 (用户 2026-09-02 裁 "系统归属由结构树派生, 不手写关键词"; 2026-09-06 再令契约预取同源)."""
    try:
        from .kb_ingest import _sys_of as _k
        return _k(text)
    except Exception as e:                                   # 分类器不可用必须响亮, 不静默退化成"不对齐"
        print(f'[fast_agent] 结构树分类器不可用: {e}', file=sys.stderr)
        return None


def contract_refs(q, k=5):
    return [c for _, c in contract_refs_scored(q, k)]


def contract_refs_scored(q, k=5):
    """问题 → 契约 top-k. 计分 = 标题 2-gram 重叠 (问句套话不计) + 系统对齐 (结构树分类器判问题与契约标题所属九系统; 同系统加分,
    异系统减分 — 用户记得 "问变桨出来偏航", 系统对不上的条目不进首轮) + 点名 claim_id 加权; 撤回条不进.
    不是检索系统, 只是把回答钉在已审级结论上; 无命中 → [] (模型照旧走本体工具)."""
    toks = set(re.findall(r'[一-鿿]{2,}|[A-Za-z][A-Za-z0-9\-]{2,}', q or ''))
    grams = ({t[i:i + 2] for t in toks if re.match(r'[一-鿿]', t) for i in range(len(t) - 1)} | toks) - _QSTOP
    qsys = {s_ for s_ in [_sys_of(q)] if s_}          # 问题点名的系统 (结构树判; 判不出 = 不做系统对齐)
    if not grams:
        return []
    out = []
    for c in contract_claims():
        if c.get('verdict') == '撤回':
            continue
        title = (c.get('display') or {}).get('title_zh') or ''
        tagged = set(c.get('scope', {}).get('systems') or []); inferred = {s_ for s_ in [_sys_of(title[:60])] if s_}
        sc = sum(1 for g in grams if g in title) + (5 if c.get('claim_id', '').lower() in (q or '').lower() else 0)
        if qsys:   # 显式系统标签 +3 > 仅结构树从标题推断 +1 (v25 实逮: '70 号蓄能器/轮毂' 被推断成变桨, 与真变桨条同分抢 top-1); 两者都不含 → -3
            sc += 3 if (qsys & tagged) else (1 if (qsys & inferred) else (-3 if (tagged | inferred) else 0))
        if sc >= 2:
            out.append((sc, c))
    _rank = {'定论': 0, '准定论·预警': 1, '候选': 2, '参考': 3, 'INSUFFICIENT': 4}   # 同分时审级高的排前
    out.sort(key=lambda x: (-x[0], _rank.get(x[1].get('verdict'), 9), x[1]['claim_id']))
    return out[:k]


def contract_text(refs):
    """契约条 → 注入文本. 每条: [claim/<id>] 六枚举 · 标题 · 窗 · 点名台数 · 必随行限制 (引用时不得省略)."""
    lines = []
    for c in refs:
        tw, sl, d = c.get('time_window') or {}, c.get('sample_level') or {}, c.get('display') or {}
        me = c.get('missing_evidence') if isinstance(c.get('missing_evidence'), str) else ''
        lines.append(f"[claim/{c['claim_id']}] 【{c['verdict']}】{(d.get('title_zh') or '')[:260]} · 窗 {tw.get('window_days')} d · 点名 {sl.get('n_machines_named') or 0} 台"
                     + (f" · 必随行限制: {me[:200]}" if me else ''))
    # 契约标题写台号用 "22C / 36B / tid11" 这类叶位代号, 接地闸的台号识别只认 WTG22 / 22# / 22号 ⇒ 模型据契约写 "22#"
    # 会被判 "台号无出处" (18084 实测 v9 偏航题因此拦下). 把契约里出现的台号按闸认的写法列一行, 信息没有增加, 只是换了写法.
    tids = set()
    for c in refs:
        txt = ((c.get('display') or {}).get('title_zh') or '') + ' ' + (c.get('missing_evidence') if isinstance(c.get('missing_evidence'), str) else '')
        tids |= {int(x) for x in re.findall(r'(?<![\d.])(\d{2})[A-F](?![A-Za-z0-9])', txt)} | {int(x) for x in re.findall(r'tid(\d{1,2})', txt)} | {int(x) for x in re.findall(r'WTG(\d{2})', txt)} | {int(x) for x in re.findall(r'(?<![\d.])(\d{1,2})#', txt)}
        tids |= {int(x) for x in (c.get('sample_level') or {}).get('named') or [] if str(x).isdigit()} if isinstance((c.get('sample_level') or {}).get('named'), list) else set()
    if tids:
        lines.append('涉及台号: ' + ', '.join(f'WTG{t:02d}' for t in sorted(tids)))
    return '\n'.join(lines)


# 校闸未过 → 不出正文 (用户令 2026-09-06). 原实现三轮改写仍未过就把最后一稿照出 (带"闸未全过"尾注),
# 实测尾注挡不住读者把正文当结论. 现在: 未过 = 只出下面这段, 正文不出.
SIX = ('定论', '准定论·预警', '候选', '参考', 'INSUFFICIENT', '撤回')
_CITE_PREFIX = re.compile(r'\[\s*(?:对象id|对象|来源|出处|id)\s*[:：]\s*')


_META_RE = re.compile(r'^\s*(?:首先|好的|用户要求|用户的问题|我需要|让我|根据用户|根据要求)[^\n]*(?:\n(?![\-•*#\d])[^\n]*)*\n+', re.S)


def strip_meta(t):
    """a3b 实逮 (2026-09-07): 改写轮把 '重新输出完整答案' 当思考题, content 开头是 '首先，用户要求…' 一整段元话语, 后面才是答案.
    只剥答案开头那段 (到首个空行), 正文不动; 剥完为空则不剥 (宁可让闸拦)."""
    if not t: return t
    m = _META_RE.match(t)
    if m and len(t) - m.end() > 60: return t[m.end():]
    return t


def canon_cites(t):
    t = strip_meta(t)
    """引用格式由机器规范化, 不靠模型改写 (18084 实测 v7: 8B 模型三轮改写仍把裁决词塞进方括号 → 全拦, 3/6 不出文):
    ① [对象id:xxx/yyy] → [xxx/yyy]  ② [claim/ID 候选] / [claim/ID（候选）] → [claim/ID], 尾巴丢弃
    ③ 引用的契约条若答案里没写其裁决词 → 在首次引用前补注【裁决词】(取自契约 = 机器来源, 比模型转述可靠).
    只动格式与补注, 不改任何内容; 裁决词写错 (与契约不符) 不在此修, 由 _claim_bad 拦."""
    t = _CITE_PREFIX.sub('[', t or '')
    # 全角括号写法 (r1:32b 实逮 【对象id: claim/X】 / （claim/X）) → 半角方括号; 溯源核验只认 [type/id]
    t = re.sub(r'【\s*(?:对象id|对象|来源|出处|id)?\s*[:：]?\s*([a-z]+/[A-Za-z0-9_/.\-]+)\s*】', r'[\1]', t)
    t = re.sub(r'[（(]\s*(?:对象id|对象|来源|出处|id)?\s*[:：]?\s*((?:claim|verdict|evidence|mechanism|alarmcode|loss|workorder|maintsystem|component|turbine|doc)/[A-Za-z0-9_/.\-]+)\s*[）)]', r'[\1]', t)
    t = re.sub(r'\[\s*(?:objects|本体|ontology)/', '[', t)   # v19 实逮: 模型写 [objects/evidence/L1/…] (本体文件叫 objects.json), 溯源核验判 MISSING
    t = re.sub(r'\[\s*claim/([A-Za-z0-9_.\-]+)[^\]]*\]', r'[claim/\1]', t)
    t = re.sub(r'(\d{1,3})号(蓄能器|油管|单向阀|阀块|阀|泵站|泵|电机|柜|轴承|螺栓|叶片)', r'\1 号\2', t)   # v19 实逮: "70号蓄能器" 被接地闸当 70 号机
    for cid in dict.fromkeys(re.findall(r'\[claim/([A-Za-z0-9_.\-]+)\]', t)):
        c = next((x for x in contract_claims() if x.get('claim_id') == cid), None)
        if not c: continue
        # v19 实逮: "裁决词写错" 启发式 (看引用前 60 字) 把相邻条目的 INSUFFICIENT 当成本条的 → 27B 正确稿被拦. 撤掉判定,
        # 改为机器**始终**在引用前补注契约裁决词 (机器来源, 与契约一致; 溯源核验按此认); 模型自己写的裁决词留着, 读者以补注为准.
        tag = f'【{c["verdict"]}】[claim/{cid}]'
        t = re.sub(r'【[^】]*】\[claim/' + re.escape(cid) + r'\]', f'[claim/{cid}]', t)   # 先去旧补注 (幂等)
        t = t.replace(f'[claim/{cid}]', tag)
    return t


MUST_CITE_SCORE = 5   # 契约 top-1 为 定论/准定论·预警 且匹配分 ≥ 此值 → 答案必须引用它 (18084 实测 v8: 叶根螺栓题契约 RD-2026-08-003 得分 11, 模型却只引 L1 代理证据答 '工单 2 条', 与契约 9 起相左而闸放行)
BLOCKED_ZH = '【校闸未过, 不出正文】{why}。请换个问法 (点名机组号 / 结论编号), 或查看契约结论清单。'
BLOCKED_EN = '[Grounding gate failed - no answer issued] {why}. Rephrase with a unit id or claim id.'

SYSTEM = ('你是如东风场检修决策助手。规则: ①判级只转述工具返回的原文并注明对象id与审级, 禁自创判级 '
          '②每个结论句末尾用[对象id]标注来源 ③证据窗外不背书, 未列≠正常, 不可判≠阴性 '
          '④金额必带电价区间口径, 限电类损失≠设备责任 ⑤取证够即答, 勿重复查询 '
          '⑦正文四百字以内, 要点式, 一句一个结论, 不铺垫不复述问题, 不写套话 (用户 2026-09-07: 简单清晰明了) '   # 用中文数字: 阿拉伯数字会被模型抄进答案再被接地闸判无出处 (a3b 实逮 400)
          '⑥[预取事实契约]里的条目是已审级结论 (六枚举: 定论/准定论·预警/候选/参考/INSUFFICIENT/撤回), '
          '与本体检索冲突时以契约为准; 引用写 [claim/结论编号], 转述其六枚举与必随行限制, 不得改写判级。 '
          '★引用格式: 方括号里只写完整对象 id 本身, 如 [evidence/loop/WTG22/齿轮箱]; '
          '不要写成 [对象id:...] 或 [WTG22] —— 前者带前缀、后者不是 id, 都无法回溯。')

# F10: 台号预取正则 — 收紧防误配:
#   保留: '29号机组' / '29#' / 'WTG29'
#   排除: '3号叶片' '9号桨叶' (号后跟部件词) / '9号台风' (号后跟台风) / '台风9号' '台风 9 号' (数字前是台风)
_TID_RE = re.compile(
    r'(?<!台风)(?<!台风\s)'                              # 前文是"台风"/"台风 " → 不是台号
    r'(?:WTG\s*(\d{1,2})'                               # WTG29 (可带空格)
    r'|(\d{1,2})\s*(?:号(?!\s*(?:叶片|桨叶|台风))|#))',   # 29号/29# — 号后跟叶片/桨叶/台风的不算
    re.IGNORECASE)


def _extract_tid(q):
    """从问题里取台号; 无合法台号返回 None (预取跳过, 不影响模型自主调工具)."""
    for m in _TID_RE.finditer(q):
        return m.group(1) or m.group(2)
    return None


def _selftest():
    """F10 正反样例 (可复跑: python -m src.ontology.fast_agent --selftest)."""
    cases = [
        ('29号机组现在什么状态', '29'),            # 正: N号机组
        ('WTG29 主轴承怎么样', '29'),              # 正: WTG 前缀
        ('看下29#的润滑泵', '29'),                 # 正: N#
        ('3号叶片有裂纹吗', None),                 # 反: 号后跟叶片
        ('9号台风影响评估', None),                 # 反: 号后跟台风
        ('台风9号来了怎么办', None),               # 反: 前文是台风
        ('台风 9 号过境后 29# 状态', '29'),        # 混合: 跳过台风编号, 取真台号
    ]
    for q, want in cases:
        got = _extract_tid(q)
        assert got == want, f'F10 selftest FAIL: {q!r} → {got!r}, 预期 {want!r}'
    print(f'F10 台号预取正则 selftest: {len(cases)} 例全过 (正3/反3/混合1)')


def _tools_schema():
    ts = []
    for n, (fn, ex, desc) in om.TOOLS.items():
        props = {k: {'type': 'number' if isinstance(v, (int, float)) and not isinstance(v, bool) else 'string'}
                 for k, v in (ex or {}).items()}
        ts.append(dict(type='function', function=dict(name=n, description=desc,
                                                      parameters=dict(type='object', properties=props))))
    return ts


# ── 在线模型 (OpenAI 兼容) ──────────────────────────────────────────
# 为什么把在线接进快路径 (2026-08-30 演示需求): 实测三条链路
#     本地 8B 走 dsh 三层 IPC    116s
#     本地 8B 走本模块快路径      19s
#     DeepSeek 走 dsh 完整编排   194s
# 瓶颈是**链路**不是模型 —— 快路径 + 在线快模型才是演示该用的组合。
# 且本模块纯 Python (无 node / ollama 依赖), 2C2G 云服务器直接能跑;
# 走 dsh 则要在服务器上再装一套 node 运行时。
# 注意选**非推理**模型: 推理税实测 10:1 (同任务 164s vs 13s)。
def _online(model):
    """model → OpenAI 兼容 endpoint 配置; 未配置密钥则返回 None, 回落 ollama。"""
    import os
    T = {
        'deepseek-chat': ('DEEPSEEK_API_KEY', 'DEEPSEEK_BASE_URL',
                          'https://api.deepseek.com', 'deepseek-chat'),
        'kimi-fast':     ('MOONSHOT_API_KEY', 'MOONSHOT_BASE_URL',
                          'https://api.moonshot.cn/v1', 'kimi-k2-turbo-preview'),
        'qwen-plus':     ('DASHSCOPE_API_KEY', 'DASHSCOPE_BASE_URL',
                          'https://dashscope.aliyuncs.com/compatible-mode/v1', 'qwen-plus'),
    }
    if model not in T:
        return None
    ek, eb, defb, real = T[model]
    # .strip(): CRLF 换行的 .env 经 source 后值会带 \r, 直接进 URL 就是
    # InvalidURL: URL can't contain control characters (2026-08-30 实逮)。
    key = (os.environ.get(ek) or '').strip()
    if not key:
        return None
    base = (os.environ.get(eb) or '').strip() or defb
    return dict(key=key, base=base.rstrip('/'), model=real)


def _chat(model, messages, tools, npred=1400):
    ep = _online(model)
    if ep:
        # ★外发脱敏: 演示时客户每问一次, 证据包就整包发给第三方一次。产物脱敏
        #   (cloud_pack) 管不到这条出口 —— 那是两个不同的路径。
        try:
            from src.windscada.deid import scrub_messages
            messages = scrub_messages(messages)
        except Exception:
            raise RuntimeError('外发脱敏模块不可用, 拒绝外发')   # 失败必须响亮, 不静默裸发
        body = dict(model=ep['model'], messages=messages, tools=tools,
                    stream=False, max_tokens=npred)
        req = urllib.request.Request(
            ep['base'] + '/chat/completions', data=json.dumps(body).encode(),
            headers={'Content-Type': 'application/json',
                     'Authorization': 'Bearer ' + ep['key']})
        with urllib.request.urlopen(req, timeout=420) as r:
            # ★不在此归一化 tool_calls.arguments: OpenAI 要求 assistant 消息**原样回传**
            #   (arguments 必须仍是字符串), 而下游第 227 行本来就做了 str→dict 的容错。
            #   在这里转反而会让回传的历史消息格式非法。
            return json.loads(r.read())['choices'][0]['message']

    # num_predict 上限 (2026-08-28): 原来不限, 模型可跑很长; 每轮又带全量 num_ctx 重算,
    # 轮次一多耗时二次增长 — 实测同一问题 44s(简单) vs 229s(三问复合)。
    # ★但固定 1400 对"逐台列举 32 台"不够: 27B 是推理模型, 先输出思考再出正文,
    #   预算被吃光 ⇒ 首轮"成文 0 字", 靠校闸补救也只出 155 字就断。改为按问法给预算。
    body = dict(model=model, messages=messages, tools=tools, stream=False,
                keep_alive='30m', options=dict(num_ctx=16384, num_predict=npred))
    if str(model).startswith('qwen3'):
        # 2026-09-07 a3b 实逮: 思考把 num_predict 吃光, content 回空 → 有效闸判空. /no_think 前缀对 30b-a3b 不够,
        # Ollama 0.33 的 think=false 才是硬开关 (qwen3 全家族支持; deepseek-r1 不加, 它靠思考出答案).
        body['think'] = False
    from src.ontology import llm_gate   # 模型闸: 确定性/单飞/重试/熔断/审计 (2026-09-07)
    try:
        msg = llm_gate.post(body, timeout=420, purpose='chat')['message']
    except urllib.error.HTTPError as e:
        detail = e.read().decode('utf-8', 'replace')[:300]
        if tools and e.code == 400 and 'tool' in detail.lower():
            # deepseek-r1 等档 Ollama 模板不带 tools (2026-09-06 加档实逮): 退化为无工具单发 —— 只据首轮已预取的
            # 契约/证据包作答, 闸不放松 (接地/契约引用/系统聚焦照旧). 响亮记一笔, 不静默.
            print(f'[fast_agent] {model} 不支持 tools ({detail[:80]}), 退化为预取直答', file=sys.stderr, flush=True)
            body['tools'] = []
            msg = llm_gate.post(body, timeout=420, purpose='chat_no_tools')['message']
            msg['_no_tools'] = True
        else:
            raise
    # 推理型模型 (deepseek-r1 / 未关 think 的 qwen) 会把 <think>…</think> 混进 content: 思考不是答案, 剥掉再进闸
    if isinstance(msg.get('content'), str) and '<think>' in msg['content']:
        msg['content'] = re.sub(r'<think>.*?</think>', '', msg['content'], flags=re.S).strip()
    return msg


# ── 离线交叉审核 (2026-08-28 用户令: 内置离线模型推理互相审核) ──
# 已有的 grounding 是**机器闸**: 只查数字与台号是否出现在证据里, 查不出"这句话证据支持不了"
# 这类语义问题。互审补的正是这一层, 且用**另一个模型**——同一个模型自审等于自己给自己判卷。
# 审核者拿到的是同一份证据包, 所以它挑的毛病可复核, 不是凭空质疑。
REVIEW_SYS = ('你是审核员, 只做一件事: 找出下面这份答案里**证据支持不了**的地方。\n'
              '三类问题各自标注: [无证据] 答案里的断言在证据中找不到依据; '
              '[矛盾] 与证据明确冲突; [过头] 证据只够说"候选/可能"却写成了确定。\n'
              '每行一条, 格式: 类型|原文片段(不超过25字)|为什么\n'
              '没有问题就只输出一行: 通过\n'
              '不要复述答案, 不要提改进建议, 不要输出别的。')


def cross_review(q, ans, facts, model='deepseek-r1:14b', timeout=420):
    """另一个本地模型复核答案与证据是否对得上。回 dict(verdict, issues, model)。
    审核档 = DeepSeek r1:14b (用户裁 2026-09-07 "审核用 DeepSeek 的模型"): 与作答的 qwen 跨家族 = 真互审, 12 GB 常驻, 一轮 40~100 s;
    作答本身是 r1 时由 serve 侧换回 qwen 27B 审, 保持跨家族。"""
    ev = '\n'.join(facts)[:9000]
    msgs = [dict(role='system', content=REVIEW_SYS),
            dict(role='user', content=f'问题: {q}\n\n=== 证据 ===\n{ev}\n\n=== 待审答案 ===\n{ans}')]
    body = dict(model=model, messages=msgs, stream=False, keep_alive='30m',
                options=dict(num_ctx=16384, num_predict=700))
    from src.ontology import llm_gate
    try:
        txt = (llm_gate.post(body, timeout=timeout, purpose='review')['message'].get('content') or '').strip()
    except Exception as e:
        return dict(verdict='审核未完成', issues=[], model=model, err=str(e)[:120])   # 审核不可用 = 明说, 不冒充通过
    txt = re.sub(r'<think>.*?</think>', '', txt, flags=re.S).strip()
    issues = []
    for line in txt.splitlines():
        line = line.strip().lstrip('-·* ')
        if not line or line in ('通过', '无问题'):
            continue
        parts = [x.strip() for x in line.split('|')]
        if len(parts) >= 3:
            t = parts[0].strip('[]【】')
            if t in ('无证据', '矛盾', '过头'):
                issues.append(dict(type=t, quote=parts[1][:40], why='|'.join(parts[2:])[:160]))
    return dict(verdict=('通过' if not issues else f'{len(issues)} 处存疑'),
                issues=issues, model=model, raw=txt[:600])


def _npred_for(q, model):
    """按问法给生成预算: 逐台列举要写几十行, 固定 1400 会被截断。"""
    listy = re.search(r'逐台|列出|全场.*(哪些|列)|所有机组|一览', q or '')
    base = 1100 if listy else 800   # 2026-09-07: 正文 ≤400 字 ≈ 700 token; 列举题留 1100 (a3b 实测 2300 字/轮 = 预算没压住)
    return int(base * (1.4 if str(model).startswith('qwen3.8') else (2.0 if 'deepseek-r1' in str(model) else 1.0)))   # 推理模型先出思考, 预算加成 (r1 思考更长, ×2)


# ★英文交付件要英文答案。语言指令必须进 **system**, 不能只加在问题前缀上 ——
#   实测: SYSTEM 是中文写的, 模型跟系统提示走, 用户侧前缀压不过去 (答案 555 个汉字)。
#   台号/部件名/对象 id 保持原样, 否则引用与检索都对不上。
# ★与 SYSTEM 的「判级只转述工具返回的原文」有张力: 英文交付件不能露中文,
#   但又不能让模型自创判级。解法是把判级限定到标准英文术语, 把可核性锚点
#   放在 [object/id] 上, 不再把中文证据原文直接显示给客户。
LANG_EN = (
    'Write for an English-speaking reader. The public English page must contain English only. '
    'Use English headings, bullets, ratings and evidence wording. Do not output Chinese, Japanese '
    'or Korean characters anywhere in the final answer.\n'
    'The tool output may be in Chinese. Translate the meaning into precise wind-turbine English '
    'instead of quoting the Chinese. Never invent a rating, mechanism, date or number: if the '
    'evidence only supports a weaker statement, keep that weaker statement.\n'
    'Keep unit ids, object ids, component names, code numbers, dates and numeric values exactly as '
    'they appear. Every conclusion sentence must end with a source tag like [object/id].')


BUDGET_S = {'qwen3:8b': 30, 'qwen3:30b-a3b': 30, 'qwen3:32b': 120}   # 用户令 2026-09-07: 复杂题最多 30 s;
# ★2026-09-22: 升档目标改为本机实装的 qwen3:32b（18.8 GB 权重, 单轮数分钟 ⇒ 给 120 s 预算, 属"离线慢审"）。


def answer(q, model='qwen3:8b', max_rounds=5, on_step=None, lang='zh', budget_s=None):
    # max_rounds 8→5 (2026-08-28): 台号预取已把一站式证据包塞进首轮, 8 轮是给"无预取"留的余量;
    # 实测复合问题会把轮次跑满且每轮上下文更长 ⇒ 收轮次比换模型有效。
    # on_step(阶段, 详情) 供服务端做进度反馈 — 229s 全程无反馈会被读成"坏了"。
    def _step(k, d=''):
        if on_step:
            try:
                on_step(k, d)
            except Exception:
                pass
    facts = []
    _t_start = time.time(); budget_s = budget_s or BUDGET_S.get(model)
    if budget_s:
        max_rounds = min(max_rounds, 3)   # v17 实逮: 8B 5 轮工具往返都快 (没触发时间闸) 但一直不成文, 最后补写一次 250 s (上下文被 5 份工具回包撑满)
    _tool_cut = 1500 if budget_s else 2000
    msgs = [dict(role='system', content=SYSTEM)]
    if lang == 'en':
        msgs.append(dict(role='system', content=LANG_EN))
    # 台号预取: 问题带台号则首轮直接注入一站式取证包 (省工具往返; F10 正则见 _TID_RE)
    # ── 预取路由 (2026-08-28 检索层优化) ──
    # 问答慢的根因不是模型慢 (裸响应 1.1s), 是它要多轮试探才找到该调哪个工具:
    # 实测损失题 170s / 15 轮, 其中 5 轮在反复调同一个 ont_loss。
    # 向量检索需要 embedding 模型 + 索引维护, 对 5,617 个结构化对象是过度工程;
    # 按问题关键词直接把对应证据包塞进首轮更省 — 台号预取已实证省 1~2 轮。
    # 这里把覆盖面从"台号 + 全场"扩到"报警码 / 损失 / 风险 / 系统部件"四类常见问法。
    # 预取上限保险: 一个问题可能同时命中多类 (如"29号的损失和风险"), 每类塞 2400 字会把
    # 上下文顶大, 而每轮都要重算全部上下文 ⇒ 预取本是为提速, 过量反而拖慢。
    # 最多 2 类、合计 4000 字, 超出即停 (台号/全场预取另计, 它们是主证据不受此限)。
    _pre_n, _pre_len = [0], [0]

    def _pre(name, fn, label):
        if _pre_n[0] >= 2 or _pre_len[0] >= 4000:
            return
        try:
            r = fn()
            if r and r.get('text'):
                t = r['text'][:2400]
                facts.append(t)
                msgs.append(dict(role='system', content=f'[预取{name}] ' + t))
                _pre_n[0] += 1; _pre_len[0] += len(t)
                _step('预取', label)
        except Exception:
            pass

    # 换件效果/在升 类问法预取 (2026-08-28): 这类问题有唯一确定入口 —— 就是那 6 条
    # 闭环与在升证据 (evidence/loop|trend), 总量不到 1KB。
    # 实测不预取时: 模型虽然调了 ont_search, 但更早的 ont_query/ont_workorder 结果已把
    # 48 条齿轮箱 verdict 灌进上下文, 它据此答"未发现换件后振动下降的证据" —— 与事实相反
    # (05#/20#/22# 换件后分别降到 0.19/0.25/0.13×)。给正确答案就得让它第一眼看到。
    if re.search(r'降下来|降下去|恢复|好转|闭环|换件后|换了.*后|修好|效果|有没有用|在升|恶化', q):
        try:
            _lp = [v for k, v in om.db().items() if k.startswith(('evidence/loop/', 'evidence/trend/'))]
            if _lp:
                _t = '换件闭环与在升证据 (全量, 出自振动线 component_history):\n' + '\n'.join(
                    f"[{o['id']}] {(o.get('props') or {}).get('summary', '')}" for o in _lp)
                facts.append(_t)
                msgs.append(dict(role='system', content='[预取闭环/在升证据] ' + _t))
                _step('预取', f'闭环与在升证据 {len(_lp)} 条')
        except Exception:
            pass

    _code = re.search(r'(?<!\d)(\d{4,5})(?!\d)', q)          # 报警码 4~5 位
    if _code and not re.search(r'20\d{2}', _code.group(1)):    # 排除年份
        _pre('报警码链', lambda: om.ont_chain_fault(code=_code.group(1)), f'报警码 {_code.group(1)}')
    # ★损失/风险/系统三类预取已回滚 (2026-08-28 实测): 加上后反而变差 —
    #   损失题 57s通过 → 99s 闸未过; 风险题 40s通过 → 74s 闸未过。
    #   原因推测: 这三类的预取包本身信息密度低(汇总口径), 塞进首轮既占上下文又没直接给出
    #   模型需要的对象 id, 它仍要再调工具, 等于多花一轮上下文重算。
    #   报警码预取保留 — 那类问题有明确入口(码→链), 预取即完整答案 (33s 通过)。
    #   结论: 预取只对"有唯一确定入口"的问法有效, 对汇总类问法无效甚至有害。

    # 全场列举类预取 (2026-08-28 功能测试逮): "全场/哪些机组/逐台/列出" 这类问题原来走
    # ont_query(type=Verdict) 拿 371 个对象全文, 进上下文被截断 → 模型只看到前几条,
    # 于是写出"(需补充, 当前数据截断)" —— 答案不是错而是残。预取紧凑汇总表, 一轮取齐。
    if re.search(r'全场|哪些机组|哪些台|逐台|所有机组|列出.*机组|机组.*列出|要关注|有什么问题|有哪些问题', q):
        try:
            _fp = om.ont_fleet_verdicts()
            if _fp.get('text'):
                facts.append(_fp['text'])
                # 注: 曾试过加"已是全量, 不要再调"的提示想省掉重复工具调用, 实测反而让模型提前收口
                # (1158字→106字且闸未过) — 重复调用是它自我确认的一部分, 不要为省轮次动这句。
                msgs.append(dict(role='system', content='[预取全场判级表] ' + _fp['text']))
                _step('预取', f"全场判级 {_fp.get('n_turbine')} 台")
        except Exception:
            pass
    tid = _extract_tid(q)   # 提前: 契约优先的台号判据要用 (原定义在台号预取处, 先用后定义 → NameError, 2026-09-07 实逮)
    _scored = contract_refs_scored(q); _refs = [c for _, c in _scored]
    _must = [c for sc, c in _scored[:1] if c.get('verdict') in ('定论', '准定论·预警') and sc >= (3 if _sys_of(q) else MUST_CITE_SCORE)]   # 契约优先: top-1 可交付级必引; 点名系统的题门槛降到 3 (v21 实逮 8B 全场类只列 L1 表不引契约 → 浅)
    if tid and _must:   # 台号题: 只有该条点名了这台 (或不点名台) 才强制引用, 否则与台号相关性闸自相矛盾
        _nm = {int(m.group(1)) for x in ((_must[0].get('sample_level') or {}).get('named') or []) for m in [re.search(r'(\d{1,2})', str(x))] if m}
        if _nm and int(tid) not in _nm: _must = []
    if _refs:
        _ct = contract_text(_refs)
        facts.append(_ct)
        msgs.append(dict(role='system', content='[预取事实契约] ' + _ct))
        _step('预取', f'事实契约 {len(_refs)} 条')
    if tid:
        _step('预取', f'{tid} 一站式证据包')
        try:
            pack = om.ont_brief_pack(tid)
        except Exception as e:      # 预取是提速手段不是前提: 某个产物缺 (实逮 loss_monthly.parquet 不在 worktree) 不该让整问崩掉
            pack = dict(error=f'{type(e).__name__}: {str(e)[:120]}')
            _step('预取', f'{tid} 证据包不可用 ({pack["error"][:60]}), 转工具取证')
        if not pack.get('error'):
            facts.append(pack['text'])
            msgs.append(dict(role='system', content='[预取证据包] ' + pack['text']))
    if lang == 'en':
        _ask_text = (q + ' Answer in English only. Use standard terms: Excellent, Good, Alert, '
                     'Alarm, Critical, Not evaluable, Established, Candidate, Watch, Closed. '
                     'End every conclusion sentence with a source tag like [object/id].')
    else:
        _ask_text = q + ' 每个结论句末尾用[对象id]标注来源。'
    _qsys = _sys_of(q)
    if _qsys and lang != 'en':
        # 2026-09-06 18084 实测逮: "29号机组变桨有没有异常" 答成主轴承润滑 (台号预取整包把别的系统灌进首轮,
        # 模型挑最响的写). 用户记得的 "问变桨出来偏航" 同型. 问题点名了系统就只答该系统.
        _ask_text += f' 问题问的是「{_qsys}」系统, 只回答该系统相关内容, 其他系统的证据不要写。'
    msgs.append(dict(role='user', content='/no_think ' + _ask_text))
    tools = _tools_schema()
    final = ''
    for _rd in range(max_rounds):
        if budget_s and tools and _rd >= max_rounds - 1:
            # 最后一轮不再给工具: 有预算的档必须在这一轮成文 (轮次闸, 与时间闸互补)
            tools = []; msgs.append(dict(role='user', content='/no_think 取证到此为止, 只据上面已有证据直接作答, 每句末尾带[对象id], 不要再调用工具。'))
            _step('预算', f'第 {_rd + 1} 轮为末轮, 收工具强制成文')
        elif budget_s and tools and (time.time() - _t_start) > budget_s * 0.6:
            # 预算过 60% 还在取证: 收掉工具, 下一轮必须成文 (8B 实测 5 轮 ont_turbine_brief 往返 = 12~325 s 却仍退化稿; 取证够即答)
            tools = []; msgs.append(dict(role='user', content='/no_think 取证到此为止, 只据上面已有证据直接作答, 每句末尾带[对象id], 不要再调用工具。'))
            _step('预算', f'{time.time() - _t_start:.0f}s 已过 60%, 收工具强制成文')
        _step('推理', f'第 {_rd + 1}/{max_rounds} 轮')
        rsp = _chat(model, msgs, tools, _npred_for(q, model))
        if rsp.pop('_no_tools', False):
            tools = []                       # 该档不支持 tools: 后续改写轮不再带, 省一次 400 往返
            _step('预取直答', '该模型档不支持工具取证')
        msgs.append(rsp)
        calls = rsp.get('tool_calls') or []
        if not calls:
            final = (rsp.get('content') or '').strip()
            _step('成文', f'{len(final)} 字')
            break
        for c in calls:
            fn = c['function']['name']
            args = c['function'].get('arguments') or {}
            if isinstance(args, str):
                try:
                    args = json.loads(args)
                except Exception:
                    args = {}
            try:
                r = om.TOOLS[fn][0](**args) if fn in om.TOOLS else dict(text=f'未知工具 {fn}', error=True)
            except Exception as e:
                r = dict(text=f'工具错误: {e}', error=True)
            _step('取证', fn)
            # 截断收紧 4000/6000 → 2000/2500: 工具返回直接进上下文, 多轮累积把 num_ctx 撑满,
            # 后续每轮都要重算这些 token ⇒ 慢的复利来源。
            facts.append(str(r.get('text', ''))[:_tool_cut])
            # tool_call_id 是 OpenAI 的硬要求 (缺了直接 400); ollama 会忽略该字段,
            # 所以无条件带上即可双向兼容。
            msgs.append(dict(role='tool', tool_call_id=c.get('id', ''),
                             content=str(r.get('text', ''))[:_tool_cut + 300]))
    # 双闸 (服务端强制, 不依赖模型自觉): ①有效性=答案须带[对象id]引用且非空洞 ②接地=数字台号不越界
    _N_FLEET = max([int(m.group(1)) for k in om.db() for m in [re.match(r'turbine/WTG(\d{2})$', k)] if m] or [38])

    def _ground(t):
        """接地闸 + 机队规模过滤: '70 号蓄能器/70号垮塌' 的 70 > 机队最大台号 38, 不可能是台号 (v22 实逮两题两档全拦)."""
        ok, bad = grounding(t, '\n'.join(facts))
        bad['turbines'] = [x for x in bad.get('turbines') or [] if x <= _N_FLEET]
        return (not bad['turbines'] and not bad.get('numbers')), bad

    def _valid(t):
        # 容错常见前缀 (2026-08-28 实测): 模型会写 [对象id:evidence/...] 或 [来源:...],
        # id 本身是对的只是多了前缀 —— 内容正确却判不通过, 会触发无谓的三轮改写。
        # 但 [WTG22] 这种不是 id 的仍然不算, 因为它回溯不到对象。
        tt = re.sub(r'\[\s*(对象id|对象|来源|出处|id)\s*[:：]\s*', '[', t or '')
        return len(tt) > 30 and re.search(r'\[[a-z]+/', tt)   # 2026-09-07: 带引用的 45 字 ('32号机组换的是70 号蓄能器【定论】[claim/…]') 已是完整答案; 空洞稿由质量/相关性/契约优先闸管, 不靠字数
    # 契约引用闸 (2026-09-06 复测逮: 模型引 [claim/X] 却漏写/改写其裁决词, 接地闸查不出 — 它只认数字台号):
    #   引用的结论编号必须在契约里, 且该条的六枚举词必须原样出现在答案中; 否则改写, 三轮仍不过 → 不出文.
    def _claim_bad(t):
        """契约引用闸 (格式与漏写已由 canon_cites 机器处理): 拦 ① 契约无此编号 ② 裁决词写错 —
        引用句里出现了别的六枚举词而该条正确裁决词不在 (如契约是准定论·预警, 模型写成定论)."""
        bad = []
        for cid in dict.fromkeys(re.findall(r'\[claim/([A-Za-z0-9_.\-]+)\]', t or '')):
            c = next((x for x in contract_claims() if x.get('claim_id') == cid), None)
            if not c:
                bad.append(f'{cid}: 契约无此编号'); continue
            named = (c.get('sample_level') or {}).get('named') or []
            nums = {int(m.group(1)) for x in named for m in [re.search(r'(\d{1,2})', str(x))] if m}   # 点名台形态多样: '32' / '32E' / 'WTG32' (v20 实逮 isdigit 全空 → 误拦)
            if tid and nums and int(tid) not in nums:
                bad.append(f'{cid}: 该条点名台 {sorted(nums)[:6]} 不含本题 {int(tid)} 号')
        return bad

    _SLUG2SYS = {'pitch': '变桨', 'yaw': '偏航', 'main_bearing': '主轴承', 'gearbox': '齿轮箱', 'generator': '发电机', 'converter': '变流器', 'control_sensor': '主控与传感网', 'blade_and_root': '叶片与叶根', 'tower_foundation': '塔架与基础'}

    def _obj_sys(i, db_):
        o = db_.get(i)
        if not o: return None
        pr = o.get('props') or {}
        if pr.get('system'): return pr['system']
        m = re.match(r'verdict/L1/WTG\d{2}/([a-z_]+)', i)
        if m: return _SLUG2SYS.get(m.group(1))
        sy = _sys_of(' '.join(str(pr.get(k) or '') for k in ('name', 'name_zh', 'summary', 'mechanism', 'title'))[:120] or i)
        if sy: return sy
        for rel in ('affects', 'about', 'of', 'basedOn'):          # v20 实逮: mechanism/WTG29/lub_starvation 名 '润滑断供·泵本体' 无系统词, 但 affects → 后主轴承/主轴承润滑泵 → 主轴承
            for tg in (o.get('links') or {}).get(rel) or []:
                tp = (db_.get(tg) or {}).get('props') or {}
                sy = tp.get('system') or _sys_of(str(tp.get('name') or tp.get('name_zh') or ''))
                if sy: return sy
        return None

    def _level_bad(t):
        """引用 [verdict/…] 的句子必须写出该对象的判级词 (与 scripts/ontology_p2_verify 的 LEVEL_MISMATCH 同判据, 提前到校闸).
        v21 实逮: 8B 写 'WTG02 变桨报警 [verdict/L1/WTG02/pitch]' 而本体判级是 优秀 = 事实错."""
        db_ = om.db(); bad = []
        for x in re.split(r'[。；;\n]', t or ''):
            for i in re.findall(r'\[(verdict/[A-Za-z0-9_/.\-]+)\]', x):
                lv = ((db_.get(i) or {}).get('props') or {}).get('level')
                if lv and lv not in x:
                    bad.append(f'{i[:36]} 判级是「{lv}」句里没写'); break
            if len(bad) >= 3: break
        return bad

    def _id_bad(t):
        """引用了本体里不存在的对象 id (模型编造/拼错) → 与溯源核验同判据, 在校闸就拦 (v19 实逮 objects/ 前缀过闸后核验红)."""
        db_ = om.db(); bad = []
        for i in dict.fromkeys(re.findall(r'\[([a-z]+/[A-Za-z0-9_/.\-]+)\]', t or '')):
            if i.startswith('claim/') or i in db_: continue
            if i.replace('/site/', '/rudong/') in db_: continue
            bad.append(i[:50])
            if len(bad) >= 3: break
        return bad

    def _quality_bad(t):
        """v16 实逮 (2026-09-07) 两种过闸的坏稿: ① 同一句抄多遍只换引用 id (29 号题 5 句全是"无直接证据异常")
        ② 句里写"偏航报警"却引 [verdict/L1/WTG14/pitch] (系统错配). 都是 8B 的典型, 拦下交给升档."""
        sents = [x.strip() for x in re.split(r'[。；;\n]', t or '') if x.strip()]
        core = [re.sub(r'\[[^\]]*\]|【[^】]*】|\s+', '', x) for x in sents]   # 保留台号 ('WTG01 偏航报警 / WTG02 …' 是合法列表, v18 误拦); 剥掉引用与机器补注的【裁决词】 (v18 29 号题四句只差补注, 漏拦)
        core = [c for c in core if len(c) >= 8]
        bad = []
        if len(core) >= 3 and len(set(core)) <= len(core) // 2:
            bad.append(f'重复句 {len(core) - len(set(core))}/{len(core)}')
        db_ = om.db()
        for x in sents:
            named = [w for w in _SLUG2SYS.values() if w in x]
            if not named: continue
            for i in re.findall(r'\[([a-z]+/[A-Za-z0-9_/.\-]+)\]', x):
                if i.startswith('claim/'): continue
                sy = _obj_sys(i, db_)            # v17 实逮: "29号机组变桨系统存在润滑断供问题 [mechanism/WTG29/lub_starvation]" — 对象是主轴承机制, 句子说变桨
                if sy and sy not in named:
                    bad.append(f'句写{"/".join(named[:2])}却引{sy}对象 {i[:40]}'); break
            if len(bad) >= 3: break
        return bad

    def _sys_bad(t):
        """问题点名系统 X, 答案引用的判级/机制对象却全属别的系统 → 答非所问. 返回 [(id, 所属系统)] 的错配清单; [] = 无问题."""
        if not _qsys:
            return []
        db_ = om.db(); cited = re.findall(r'\[\s*([a-z]+/[A-Za-z0-9_/.\-]+)\s*\]', t or '')
        got = []
        for i in dict.fromkeys(cited):
            if i.startswith('claim/'):
                c = next((x for x in contract_claims() if x.get('claim_id') == i[6:]), None)
                if c:
                    cs = set(c.get('scope', {}).get('systems') or []) | {s_ for s_ in [_sys_of((c.get('display') or {}).get('title_zh', '')[:60])] if s_}
                    got.append((i, _qsys if _qsys in cs else (sorted(cs)[0] if cs else None)))
            else:
                got.append((i, (db_.get(i, {}).get('props') or {}).get('system')))
        got = [(i, sy) for i, sy in got if sy]
        if got and not any(sy == _qsys for _, sy in got):
            return got
        return []
    def _must_bad(t):
        return [c for c in _must if f"[claim/{c['claim_id']}]" not in (t or '')]
    _t_gate0 = time.time()
    for _fx in range(3):
        if _fx >= 1 and (time.time() - _t_gate0) / _fx > 60:   # 每轮改写 >60 s 的慢档 (27B/32B) 只给一次改写机会: 校闸不过就不出文, 不烧 10 分钟
            _step('校闸', f'慢档改写超时预算 (单轮 {(time.time() - _t_gate0) / _fx:.0f}s), 不再改写')
            break
        if _fx >= 1 and budget_s and (time.time() - _t_start) > budget_s:   # 有预算的档: 超预算后不再改写 (改写一次仍不过 = 交给升档/不出文)
            _step('校闸', f'超预算 {time.time() - _t_start:.0f}s > {budget_s}s, 不再改写')
            break
        _step('校闸', f'第 {_fx + 1} 次')
        final = canon_cites(final)
        ok, bad = _ground(final)
        cb = _claim_bad(final); sb = _sys_bad(final); mb = _must_bad(final); qb = _quality_bad(final); ib = _id_bad(final); lb = _level_bad(final)
        if _valid(final) and ok and not cb and not sb and not mb and not qb and not ib and not lb:
            return dict(answer=final, gate='接地闸 通过', rounds=len(msgs), facts=facts)
        if lb and not (mb or sb or cb or qb or ib):
            fix = ('你上一轮答案引用判级对象的句子没写出它的判级词: ' + '; '.join(lb) + '。每个 [verdict/…] 所在句必须原样写出该对象的判级 (优秀/良好/报警/危险/不可判/定论/候选…), 不许把别的判级安在它头上; 重新输出完整答案, 不要解释。')
        elif ib and not (mb or sb or cb or qb):
            fix = ('你上一轮答案引用了本体里不存在的对象 id: ' + ', '.join(ib) + '。只能引用工具返回或预取里出现过的完整 id, 原样照抄; 删掉编造的引用, 重新输出完整答案, 不要解释。')
        elif qb and not (mb or sb or cb):
            fix = ('你上一轮答案质量不合格: ' + '; '.join(qb) + '。重新输出: 每句一个不同的结论, 不要复述同一句; 引用的判级对象必须与句子说的系统一致 (偏航句只引 yaw 判级); 没有证据就写"无直接证据"一次即可, 不要解释。')
        elif mb:
            fix = ('契约里已有针对本问题的已审级结论, 回答必须引用并转述它 (本体证据只能作补充, 与契约冲突时以契约为准): ' +
                   '; '.join(f"[claim/{c['claim_id']}]【{c['verdict']}】{(c.get('display') or {}).get('title_zh', '')[:80]}" for c in mb) +
                   '。重新输出完整答案, 每句末尾带[对象id], 不要解释。')
        elif sb:
            fix = (f'问题问的是「{_qsys}」系统, 你上一轮引用的对象全属别的系统 ({"; ".join(f"{i}→{sy}" for i, sy in sb[:4])})。'
                   f'重新输出完整答案: 只写「{_qsys}」系统的证据与结论 (证据里没有就说没有), 每句末尾带[对象id], 不要解释。')
        elif cb:
            fix = ('你上一轮答案引用契约结论有误: ' + '; '.join(cb) +
                   '。重新输出完整答案: 引用 [claim/编号] 时原样写出该条裁决词, 删掉契约里没有的编号, 不要解释。')
        elif lang == 'en':
            if not _valid(final):
                fix = (f'Re-answer the original question: {q}. Use only the evidence above, write '
                       'in English only, and end every conclusion sentence with a source tag like '
                       '[object/id]. Give the answer directly, with no process explanation.')
            else:
                fix = (f'Your previous answer used numbers {bad["numbers"]} or unit ids '
                       f'{bad["turbines"]} that are not grounded in the tool evidence. Rewrite '
                       'the full answer in English only, keep grounded content, and remove or correct '
                       'unsupported values. Do not explain the correction process.')
        elif not _valid(final):
            fix = (f'重新回答原问题「{q}」: 基于上面工具返回的证据完整作答, '
                   '每个结论句末尾用[对象id]标注来源。直接给答案, 不要解释过程。')
        else:
            fix = (f"你上一轮答案中数字 {bad['numbers']} / 台号 {bad['turbines']} 不在工具证据里。"
                   '重新输出完整答案: 保留有证据的内容, 删去或修正越界的数字, 不要解释。')
        msgs.append(dict(role='user', content='/no_think ' + fix))
        rsp = _chat(model, msgs, tools=[], npred=_npred_for(q, model))
        final = (rsp.get('content') or '').strip()
        msgs.append(rsp)
    final = canon_cites(final)
    ok, bad = _ground(final)
    cb = _claim_bad(final); sb = _sys_bad(final); mb = _must_bad(final); qb = _quality_bad(final); ib = _id_bad(final); lb = _level_bad(final)
    if ok and _valid(final) and not cb and not sb and not mb and not qb and not ib and not lb:
        return dict(answer=final, gate='接地闸 通过', rounds=len(msgs), facts=facts)
    why = (f'有效={bool(_valid(final))}, 越界={bad if not ok else "无"}, 契约引用={cb or "无误"}, 系统聚焦={("答非所问: " + str(sb[:3])) if sb else "无误"}, '
           f'契约优先={("未引用 " + ", ".join(c["claim_id"] for c in mb)) if mb else "无误"}, 质量={qb or "无误"}, 编造id={ib or "无"}, 判级词={lb or "无误"}')
    blocked = (BLOCKED_EN if lang == 'en' else BLOCKED_ZH).format(why=why)
    print(f'[fast_agent] 校闸未过 ({why[:160]}) 末稿前 600 字 (只进服务端 stderr, 供诊断): {final[:600]!r}', file=sys.stderr, flush=True)
    return dict(answer=blocked, gate='校闸未过 · 不出正文 (' + why + ')', rounds=len(msgs), facts=facts,
                blocked=True, draft_len=len(final or ''))


# ── 自动升档 (用户令 2026-09-07 "复杂题 8B 拦了自动升 a3b 重答") ──
# 8B 在全场类复杂题上会写退化稿 (实测 "无直接证据[对象id]"), 闸拦得对但用户拿不到答案; a3b (30B 参数 / 3B 激活) 单轮 ~20 s 且质量接近 27B.
# 规则: 默认档被闸拦下 → 用升档目标重答一次; 升档也拦 → 照旧不出文, 但把两次拦截原因都带回.
ESCALATE = {'qwen3:8b': 'qwen3:32b'}   # ★2026-09-22 用户人工测试 7 实逮: 原写 qwen3.8:27b, 本机**没这个 tag**
#   ⇒ 升档调用 digest=None、ask 终态 error（用户看到"复核中"卡住）。改为本机实装的 32B（18.8 GB 权重, 预算见 BUDGET_S）。


def installed_models(timeout=5):
    """本机 Ollama 已装模型名集合（拿不到就返回空集 —— 绝不影响主流程）。"""
    import json as _j
    import urllib.request as _u
    try:
        with _u.urlopen('http://127.0.0.1:11434/api/tags', timeout=timeout) as r:
            d = _j.loads(r.read().decode('utf-8'))
        return {m.get('name', '') for m in (d.get('models') or [])}
    except Exception:
        return set()


def escalate_target(model):
    """升档目标解析: ①`configs/models.json` 的 escalate_qa（且**确已安装**）→ ②本文件表 → ③None。

    为什么要"确已安装": 档位写了一个本机没装的模型时，升档必然失败，而用户只看到一句"复核中"。
    配置可换模型，但换错必须**响亮回落**，不能静默把答案卡死。
    """
    try:
        import json as _j
        from src import paths as _P
        _cand = (((_j.loads(_P.config('models.json').read_text(encoding='utf-8')).get('tiers') or {})
                  .get('escalate_qa') or {}).get('model'))
        inst = installed_models()
        if _cand and (not inst or _cand in inst):
            return _cand
        if _cand:
            print(f'[escalate] 配置档位 {_cand} 本机未安装, 回落到 {ESCALATE.get(model)}', flush=True)
    except Exception:
        pass
    return ESCALATE.get(model)


def interim_text(q, first_gate='', nxt=''):
    """先快后准 (用户裁 2026-09-07): 默认档被拦后 30 s 内先给用户看的"初答" —— 不含被拦的错稿, 只列契约里与本题最相关的结论 (机器直出, 契约来源),
    并明说正在复核. 升档模型出来后整段替换。"""
    refs = contract_refs(q, k=3)
    lines = [f"- 【{c['verdict']}】{(c.get('display') or {}).get('title_zh', '')[:110]} [claim/{c['claim_id']}]" for c in refs]
    return ('【初答 · 复核中】默认档 (8B) 的回答未通过校闸' + (f' ({first_gate[:60]})' if first_gate else '')
            + f', 已交更强模型 ({nxt or "升档档"}) 复核, 约 1–3 分钟后本段会被替换。\n'
            '契约里与本问题最相关的已审级结论 (机器直出, 非模型生成):\n' + ('\n'.join(lines) if lines else '- (契约里没有直接相关的结论)'))


def answer_escalate(q, model='qwen3:8b', max_rounds=5, on_step=None, lang='zh', on_partial=None):
    r = answer(q, model, max_rounds=max_rounds, on_step=on_step, lang=lang)
    nxt = escalate_target(model)
    if not r.get('blocked') or not nxt:
        r.setdefault('model', model); return r
    if on_partial:
        try: on_partial(interim_text(q, r['gate'], nxt))
        except Exception: pass
    if on_step:
        try: on_step('升档', f'{model} 校闸未过 → {nxt} 重答')
        except Exception: pass
    r2 = answer(q, nxt, max_rounds=max_rounds, on_step=on_step, lang=lang)
    r2['model'] = nxt; r2['escalated_from'] = model; r2['first_gate'] = r['gate']
    if r2.get('blocked'):
        r2['gate'] = f"两档均未过: {model} ({r['gate'][:60]}) / {nxt} ({r2['gate'][:60]})"
    return r2


if __name__ == '__main__':
    import time
    if sys.argv[1:2] == ['--selftest']:
        _selftest()
        sys.exit(0)
    q = sys.argv[1] if len(sys.argv) > 1 else '29号机组现在什么状态、为什么、该怎么办?'
    mdl = sys.argv[2] if len(sys.argv) > 2 else 'qwen3:8b'
    t0 = time.time()
    r = answer(q, mdl)
    print(f"[{mdl}] {time.time()-t0:.0f}s · {r['gate']} · {r['rounds']} 轮\n")
    print(r['answer'])