"""Run one assistant turn: history → model → reply → durable transcript. The model kernel is the one that already ships (``app.domains.support.copilot``: pluggable Bailian/Anthropic, employee-scoped read-only tools, grounded-or-escalate). This module does not reimplement any of that; it supplies the channel-side concerns a chat needs and a web form does not — remembered context, one turn at a time per chat, a holding note when an answer is slow, and a transcript an administrator can read back. """ from __future__ import annotations import asyncio import contextlib import json import logging import re import time import uuid from dataclasses import dataclass, field, replace from datetime import UTC, datetime from typing import Any from sqlalchemy import select from sqlalchemy.exc import IntegrityError from sqlalchemy.ext.asyncio import AsyncSession from app.assistant import messages as canned from app.assistant import session as session_store from app.assistant.models import ( CHANNEL_FEISHU, CHAT_KIND_DM, OUTCOME_ANSWERED, OUTCOME_FAILED, OUTCOME_UNAVAILABLE, ROLE_ASSISTANT, ROLE_USER, AssistantConversation, AssistantMessage, ) from app.assistant.prompt import feishu_system_prompt from app.assistant.reply_hygiene import add_missing_destination, to_feishu_text from app.core.locale import Lang, resolve_lang from app.identity.person_models import IdentityPerson logger = logging.getLogger(__name__) #: How long to stay silent before telling the employee we are still working. #: A turn that calls tools routinely runs several seconds, so the original 4s #: threshold fired before nearly every answer and read as a verbal tic rather #: than reassurance. Only a turn slow enough that the silence would otherwise #: look like a broken bot earns an interruption. HOLDING_NOTE_AFTER_SECONDS = 15.0 #: Attempts at claiming a transcript sequence. Two concurrent turns need one #: retry between them; more than a couple means something other than a race. _SEQ_RETRIES = 3 # A model may repeat the confirmation-card wording from history even when it # did not call the proposal tool in the current turn. That wording is a claim # about an external side effect, so it must be backed by ``cards_sent`` rather # than trusted as ordinary model prose. _CARD_SENT_CLAIM_MARKERS = ( "已经把确认卡片发给你", "确认卡片已发", "sent you a confirmation card", "confirmation card is now in front of you", "確認カードを送信しました", ) def _claims_card_sent_without_delivery(reply: str) -> bool: folded = (reply or "").casefold() return any(marker.casefold() in folded for marker in _CARD_SENT_CLAIM_MARKERS) @dataclass(frozen=True) class TurnResult: outcome: str reply: str tool_calls: list[str] provider: str | None latency_ms: int #: The stored assistant message, so a rating can be attached to the exact #: answer that earned it. message_id: str | None = None #: Knowledge-base sources the turn actually read, captured from the tool #: results rather than from whatever the model chose to repeat. docs: list[dict[str, str]] = field(default_factory=list) #: The same answer with its Markdown intact, for the one surface that can #: render it. ``reply`` is flattened for Feishu *text* messages, which show #: every marker literally; a card element is ``{"tag": "markdown"}`` and #: renders bold and lists properly, so flattening first threw away #: formatting the card could have shown. reply_rich: str = "" #: The turn already answered with a card, so ``reply`` is context for the #: transcript and the model's memory — not a message to send. Sending it #: too would put a narration underneath the card the employee is reading. card_only: bool = False @dataclass(frozen=True) class SimulationResult: """A non-mutating preview of the same Feishu employee assistant path. This deliberately does not create a conversation, proposal, card or history entry. Administrators need to inspect the decision path without changing what the employee will later see in their real chat. """ reply: str tool_calls: list[str] docs: list[dict[str, str]] policy_conflicts: list[dict[str, str]] proposed_actions: list[dict[str, Any]] provider: str | None available: bool turn_failed: bool #: Every tool result of this turn, verbatim, and only when the caller asked #: for it (``capture_evidence=True``). Off by default because the admin #: preview does not need it and a turn's raw tool output can run to #: hundreds of kilobytes. #: #: The evaluation harness needs it for one check no model can perform: a #: URL or an id that appears in the reply and in NOTHING the turn read was #: not retrieved, it was invented. Answering that requires the bytes the #: model actually saw, not the tidied ``docs`` list. evidence: list[str] = field(default_factory=list) #: Wall-clock of the model/tool loop, milliseconds. Recorded so a run can #: report latency alongside quality — an answer that takes ninety seconds #: is a different product from the same answer in eight. latency_ms: int | None = None #: Bounds on captured evidence. A bridged listing tool can return a megabyte, #: and this list is serialised into an evaluation row that an administrator #: later opens in a browser. _EVIDENCE_MAX_BLOBS = 40 _EVIDENCE_MAX_CHARS = 20_000 #: Turns of supplied history a simulation will replay. The kernel caps what it #: sends the model anyway (``_MAX_HISTORY``); this stops an API caller from #: making the request body itself the attack. _MAX_SETUP_TURNS = 20 _MAX_SETUP_TURN_CHARS = 4_000 def _clean_setup_turns(turns: list[dict[str, str]] | None) -> list[dict[str, str]]: """Validate caller-supplied history into the kernel's message shape. Anything that is not a ``user``/``assistant`` message with text is dropped rather than passed through: the value lands directly in the provider request, and a ``system`` role smuggled in here would be an instruction the channel's own prompt never sanctioned. """ cleaned: list[dict[str, str]] = [] for turn in (turns or [])[:_MAX_SETUP_TURNS]: if not isinstance(turn, dict): continue role = str(turn.get("role") or "").strip().lower() if role not in (ROLE_USER, ROLE_ASSISTANT): continue content = str(turn.get("content") or "").strip() if not content: continue cleaned.append({"role": role, "content": content[:_MAX_SETUP_TURN_CHARS]}) return cleaned _SOFTWARE_ROLE_WORDS = re.compile( r"负责人|谁负责|owner|it\s*联系人|it\s*contact|support\s*contact", re.IGNORECASE, ) _SOFTWARE_ROLE_QUERY = re.compile( r"^\s*(?P.+?)(?:这个|该)?(?:软件|应用|app|application)" r"(?:的)?(?:负责人|owner|it\s*联系人|it\s*contact|support\s*contact)", re.IGNORECASE, ) _SOFTWARE_ROLE_QUERY_SHORT = re.compile( r"^\s*(?:谁是|谁负责|who\s+owns?\s+|owner\s+of\s+)" r"(?P.+?)(?:的)?(?:负责人|owner|it\s*联系人|it\s*contact)?\s*[吗?？]?$", re.IGNORECASE, ) _SOFTWARE_ROLE_QUERY_SUFFIX = re.compile( r"^\s*(?P.+?)\s*(?:的)?" r"(?:负责人|owner|it\s*联系人|it\s*contact|support\s*contact)\s*(?:是谁|是誰|who)?\s*[吗?？]?$", re.IGNORECASE, ) # A conversation can contain an assistant answer written before a policy was # corrected. Policy questions must not inherit that prose as evidence on the # next turn: the current workflow and current published sources are the only # admissible facts. Keep the classifier deliberately narrow; it changes # history handling only for an executable asset-buyback intent, not every # mention of a computer. _BUYBACK_DIRECT_TERMS = ( "回购", "买断", "内购", "buy back", "buyback", "buy out", "buyout", "repurchase", ) _BUYBACK_COMPUTER_TERMS = ("电脑", "笔记本") # Informational questions need evidence even when a model decides it can answer # from general IT knowledge. This is deliberately a cue for *pre-fetching*, # not a substitute for action discovery: the normal tool surface remains # available for requests that turn out to be executable workflows. _KB_INFORMATIONAL_WORDS = re.compile( r"如何|怎么|怎样|为什么|是什么|流程|政策|指南|手册|设置|配置|使用|发送|" r"how\s+to|what\s+is|policy|guide|manual|setup|configure", re.IGNORECASE, ) @dataclass(frozen=True) class BuybackGrounding: """Current workflow evidence injected before a buyback answer. The approval service is the operational source of truth for its own calculation. A mirrored Feishu article can be older than that service; we cite only a KB source that independently supports the same 36-month lifecycle, and never let a conflicting legacy document override it. """ system_context: str docs: list[dict[str, str]] conflicting_docs: list[dict[str, str]] @dataclass(frozen=True) class KnowledgeGrounding: """Server-fetched evidence for an ordinary informational question.""" system_context: str docs: list[dict[str, str]] def _public_source_links(docs: list[dict[str, str]]) -> list[dict[str, str]]: """Persist only the citation contract rendered back to employees. Article ids, source types and embedding freshness are internal retrieval metadata. Keeping them in the transcript couples the employee-facing card to implementation details and made pre-fetched and tool-fetched sources serialize differently. """ return [ {"title": str(doc.get("title") or doc.get("url") or ""), "url": str(doc.get("url") or "")} for doc in docs if doc.get("url") ] def _is_buyback_question(question: str) -> bool: # Word containment is linear and preserves the intentionally narrow # classifier without applying an ambiguous regex to user-controlled text. normalized = " ".join((question or "").casefold().split()) if any(term in normalized for term in _BUYBACK_DIRECT_TERMS): return True return "购买" in normalized and any( term in normalized for term in _BUYBACK_COMPUTER_TERMS ) def _is_kb_informational_question(question: str) -> bool: return bool(_KB_INFORMATIONAL_WORDS.search(question or "")) async def _ground_kb_question( db: AsyncSession, question: str, ) -> KnowledgeGrounding: """Load KB evidence before the model can choose to skip retrieval. A prompt instruction alone is not a control plane: production showed the model returning "no company guide" without a single KB call, while the exact article was published and semantic-searchable. The service owns the retrieval for informational requests, then tells the model exactly what it may use (or that none was found). This leaves actions available, but makes an unsupported absence claim impossible. """ from app.domains.support.kb_service import SupportKbService # noqa: PLC0415 articles: list[Any] = list( await SupportKbService(db).search_published(question, limit=3) ) docs = [ { "id": str(article.id), "title": article.title, "url": article.external_url or "", "source": article.source, "updated_at": str(article.updated_at or ""), } for article in articles ] if not articles: evidence = "- No matching published KB article was found." else: evidence = "\n".join( f"- {article.title}: {article.body[:1200]}\n Source: {article.external_url or 'Helix KB'}" for article in articles ) return KnowledgeGrounding( docs=docs, system_context=( "\n\nKB EVIDENCE PREFETCH — THIS TURN\n" "The server searched the published company KB before this answer. " "Use the returned article(s) as the source for any how-to or policy " "claim. If one is listed, do NOT say that the company has no guide; " "answer from it and cite its source. If none is listed, say only that " "the published KB search found no matching article — never infer a " "company practice from general knowledge.\n" f"{evidence}\n" ), ) def _supports_current_buyback_lifecycle(body: str) -> bool: """Whether a mirrored policy supports, rather than contradicts, 36 months.""" folded = re.sub(r"\s+", "", body or "").casefold() return ( ("36个月" in folded or "36months" in folded) and "满2年" not in folded and "满二年" not in folded and "电脑采购年限>2年" not in folded ) def _contradicts_current_buyback_lifecycle(body: str) -> bool: folded = re.sub(r"\s+", "", body or "").casefold() return "满2年" in folded or "满二年" in folded or "电脑采购年限>2年" in folded async def _ground_buyback_policy(db: AsyncSession) -> BuybackGrounding: """Load current workflow policy plus compatible KB evidence once per turn. This is a preflight, not a model suggestion: it gives Feishu and the administrator preview the same facts and prevents stale conversation text from becoming a policy source. No write or external side effect occurs. """ from app.approval.buyback_price import DEPRECIATION_LIFETIME_MONTHS # noqa: PLC0415 from app.domains.support.kb_service import SupportKbService # noqa: PLC0415 articles: list[Any] = list( await SupportKbService(db).search_published("电脑回购", limit=12) ) compatible = [ article for article in articles if _supports_current_buyback_lifecycle(article.body) ] conflicting = [ article for article in articles if _contradicts_current_buyback_lifecycle(article.body) ] def serialize(article) -> dict[str, str]: return { "id": str(article.id), "title": article.title, "url": article.external_url or "", "source": article.source, "updated_at": str(article.updated_at or ""), } docs = [serialize(article) for article in compatible[:3]] evidence = ( "\n".join(f"- {doc['title']}: {doc['url'] or 'no external link'}" for doc in docs) or "- No compatible KB article is currently available." ) return BuybackGrounding( docs=docs, conflicting_docs=[serialize(article) for article in conflicting[:3]], system_context=( "\n\nCURRENT BUYBACK POLICY — VERIFIED BEFORE THIS TURN\n" f"The live Helix approval calculation uses a {DEPRECIATION_LIFETIME_MONTHS}-month " "(3-year) lifecycle: before that it calculates the buyback quote from the " "asset record; at or after that it applies the approved floor schedule. " "This workflow rule is authoritative for eligibility and price. Do not repeat " "or rely on any earlier chat statement saying 2 years, even if it appears in " "conversation history or a legacy document. State the current rule, show the " "employee's eligible device choices when available, and ask which device they " "want before proposing the approval.\n" "Compatible KB evidence (cite only these when mentioning policy):\n" f"{evidence}\n" "This preflight is the only KB evidence allowed for this buyback turn; do not " "search the knowledge base again, because that search can surface withheld " "legacy articles.\n" "If no compatible KB article is listed, say the workflow rule is confirmed but " "the policy document needs administrator review; do not invent any additional policy." ), ) @dataclass(frozen=True) class Preflight: """Everything the server looked up BEFORE the model saw the question. Three groundings used to be inlined, near-identically, at two call sites, and a third caller (the web page) had none of them at all. That asymmetry is what this dataclass exists to make impossible: a channel gets the preflight by asking for it, not by remembering to copy twenty lines. Each piece is a preflight rather than a prompt instruction for the same reason, stated once here instead of three times: a system prompt is a request. Production showed the model answering 「知识库中没有…」 without having called the KB tool once, for a question the published article answered. The server does the retrieval and then tells the model what it may use — which leaves every action available, and makes an unsupported absence claim impossible. """ #: Appended to the channel's system prompt, in the order the groundings ran. system_context: str #: Sources the employee may be shown, deduplicated by URL. docs: list[dict[str, str]] #: Buyback only: published articles that CONTRADICT the live workflow rule. #: Withheld from the model and surfaced to administrators. policy_conflicts: list[dict[str, str]] #: The server searched the KB itself, so the turn cited it even if the model #: called no tool. Reported in ``tool_calls`` — a transcript that omits it #: makes a correctly grounded citation look fabricated. kb_prefetched: bool #: A live policy question is grounded from the CURRENT sources every time. #: An assistant message written before a policy was corrected is not an #: authority, so the thread is dropped for this turn only. drop_history: bool #: The individual context blocks, for the evaluation harness's evidence #: capture. Same content as ``system_context``, unconcatenated. evidence_blocks: list[str] #: 编号预检索实际跑过的只读工具名。和 ``kb_prefetched`` 同一个道理:服务端 #: 替模型查掉的东西必须出现在这一轮的 ``tool_calls`` 里 —— 否则管理员看到的 #: 是一条报出了账号负责人、却没有任何来源的回答,读起来像凭空编的。 identifier_tools: list[str] async def build_preflight( db: AsyncSession, question: str, *, person_id: Any = None, ) -> Preflight: """Ground a question against the platform before the model reads it. Never raises. Each grounding is independent and a failure in one costs its own evidence and nothing else — the assistant staying available on a KB outage is worth more than any single citation. """ from app.assistant import flow_grounding as flows # noqa: PLC0415 from app.assistant import identifier_grounding as ids # noqa: PLC0415 docs: list[dict[str, str]] = [] conflicts: list[dict[str, str]] = [] blocks: list[str] = [] is_buyback = _is_buyback_question(question) kb_prefetched = False grounded_identifiers = False identifier_tools: list[str] = [] if is_buyback: try: buyback = await _ground_buyback_policy(db) docs.extend(buyback.docs) conflicts.extend(buyback.conflicting_docs) blocks.append(buyback.system_context) kb_prefetched = True except Exception: # noqa: BLE001 - keep the assistant available on KB trouble logger.exception("assistant: buyback policy preflight failed person=%s", person_id) elif _is_kb_informational_question(question): try: kb = await _ground_kb_question(db, question) docs.extend(kb.docs) blocks.append(kb.system_context) kb_prefetched = True except Exception: # noqa: BLE001 - a KB outage must not break the assistant logger.exception("assistant: KB preflight failed person=%s", person_id) # 「这件事该走哪个流程、找谁」. Pre-fetched rather than left to the model to # discover, and NOT gated on the KB coming back empty: the transcript that # motivated this had the assistant answer 「知识库中没有…」 for a question the # approval register could answer outright. # 问题里带了编号(云账号号码、实例 id、IP、域名、资产序列号)时,先按提问者 # 本人的权限把它查成行。见 ``identifier_grounding`` 的模块头:2026-09-07 员工 # 问「380838443567 这个云账号谁负责」,拿到的是两条**审批流程**的负责人。 if person_id is not None: try: asker = await db.get(IdentityPerson, str(person_id)) if asker is not None: resolved = await ids.build_identifier_grounding( question, person=asker, ) if resolved is not None: blocks.append(resolved.system_context) grounded_identifiers = True identifier_tools.extend( dict.fromkeys(hit["tool"] for hit in resolved.hits), ) except Exception: # noqa: BLE001 - grounding must never break a turn logger.exception( "assistant: identifier preflight failed person=%s", person_id, ) # # 编号查到了行,而问句**纯粹**在问「这个东西归谁」时,就不再附流程清单: # 那正是那次错答的成因 —— 手边有一份名字很像的现成清单时,模型会从清单里挑, # 而不去读已经摆在眼前的那一行。 # # 但混着问的不撤(``asks_how_a_process_works``):「这个账号怎么申请权限」 # 里,那一行是「哪个账号」的答案,流程清单是「怎么申请」的答案 —— 撤掉它 # 就把员工真正问的那一半扔了。 withhold_flows = grounded_identifiers and not flows.asks_how_a_process_works( question, ) if flows.looks_like_a_how_do_i_question(question) and not withhold_flows: try: flow = await flows.build_flow_grounding(db, question) if flow is not None: blocks.append(flow.system_context) except Exception: # noqa: BLE001 - grounding must never break a turn logger.exception("assistant: flow preflight failed person=%s", person_id) return Preflight( system_context="".join(blocks), docs=docs, policy_conflicts=conflicts, kb_prefetched=kb_prefetched, drop_history=is_buyback, evidence_blocks=[block for block in blocks if block], identifier_tools=identifier_tools, ) def _software_role_product(question: str) -> str | None: """Extract a product only for an explicit ownership/contact question. This is intentionally a narrow classifier. It is not an attempt to understand arbitrary language; it exists to prevent the LLM from choosing a sibling data source (for example SSO integrations) for a question whose answer is a canonical software-catalogue relationship. """ text = (question or "").strip() if not text or not _SOFTWARE_ROLE_WORDS.search(text): return None for pattern in ( _SOFTWARE_ROLE_QUERY, _SOFTWARE_ROLE_QUERY_SHORT, _SOFTWARE_ROLE_QUERY_SUFFIX, ): match = pattern.search(text) if match: product = re.sub(r"(?:这个|该|的)$", "", match.group("product")).strip() return product or None return None def _normalise_catalogue_name(value: str) -> str: """Match the catalogue's case/spacing-insensitive product identity.""" return re.sub(r"[^\w\u4e00-\u9fff]", "", value or "").casefold() async def _forced_software_role_lookup( *, db: AsyncSession, person: IdentityPerson, question: str, trace_id: str, called: list[str], ) -> str | None: """Return a grounded answer for software owner/contact questions. Ownership is a relational lookup, not a generative answer. The model may still handle the many other assistant intents, but this class gets a server-side result and therefore cannot hallucinate a colleague or turn a non-empty catalogue result into "not found" after a second tool call. """ product = _software_role_product(question) if not product: return None from app.assistant import mcp_bridge # noqa: PLC0415 from app.mcp.context import mcp_trace_id, mcp_user_id # noqa: PLC0415 try: user_token = mcp_user_id.set(person.id) trace_token = mcp_trace_id.set(uuid.UUID(trace_id)) try: raw = await mcp_bridge._call_tool( # noqa: SLF001 — audited in-process lookup { "name": "list_software_catalog", "arguments": {"search": product, "page": 1, "page_size": 10}, }, person=person, ) finally: mcp_trace_id.reset(trace_token) mcp_user_id.reset(user_token) except Exception: # noqa: BLE001 — normal model path remains available logger.exception("assistant: forced software role lookup failed") return None called.append("list_software_catalog") try: payload = json.loads(raw) except (TypeError, ValueError): return "软件目录查询失败，暂时无法确认负责人。" if payload.get("truncated"): return "软件目录返回了不完整结果，暂时无法安全确认负责人。" items = payload.get("items") or [] if not items: return f"在软件目录中未找到「{product}」的应用记录，因此暂时没有可确认的负责人信息。" # The catalogue also matches descriptions. "Netsuite" therefore returns # the exact Netsuite row plus Salesforce, whose description mentions the # NetSuite integration. For a relationship question, name identity wins # over full-text relevance; never ask the employee to disambiguate a row # whose name already exactly matches what they typed. needle = _normalise_catalogue_name(product) name_matches = [ item for item in items if needle and ( _normalise_catalogue_name(str(item.get("name") or "")) == needle or _normalise_catalogue_name(str(item.get("name") or "")) in needle or needle in _normalise_catalogue_name(str(item.get("name") or "")) ) ] if len(name_matches) == 1: items = name_matches if len(items) > 1: names = "、".join(str(item.get("name") or "") for item in items[:5]) return f"软件目录中找到多个匹配项：{names}。请告诉我你指的是哪一个。" item = items[0] role_is_contact = bool(re.search(r"it\s*联系人|it\s*contact|support\s*contact", question, re.I)) role = "it_contact_person" if role_is_contact else "owner_person" role_label = "IT 联系人" if role_is_contact else "业务负责人" person_data = item.get(role) or {} mention = person_data.get("mention") display_name = person_data.get("display_name") quality = (item.get("data_quality") or {}).get("it_contact" if role_is_contact else "owner") if mention and display_name: named = mention elif display_name and quality == "canonical_person": # Defensive fallback for an older tool payload that has a canonical # person but no mention marker. Never invent an id or mention token. named = display_name else: named = "未关联到具体员工" answer = f"{item.get('name') or product} 的{role_label}是 {named}。" if not role_is_contact: it_contact = item.get("it_contact_person") or {} if it_contact.get("mention"): answer += f" IT 联系人是 {it_contact['mention']}。" elif item.get("it_contact"): answer += f" IT 联系部门是 {item['it_contact']}。" if item.get("version"): answer += f" 当前版本：{item['version']}。" if quality in {"legacy_text_only", "broken_person_reference"}: answer += " 目录中的负责人字段未关联到在职员工，不能把原始文本当作人员身份。" return answer async def simulate_employee_turn( db: AsyncSession, *, person: IdentityPerson, question: str, lang: Lang | None = None, setup_turns: list[dict[str, str]] | None = None, capture_evidence: bool = False, ) -> SimulationResult: """Preview a brand-new Feishu employee turn through the production path. The previous KB debug page called the generic web-copilot prompt with an administrator identity and a KB-only instruction. That could never be an answer trace for the employee assistant: it omitted real flows such as device buyback, used different personal data, and let the model pick a different search query. This function uses the Feishu prompt, read bridge, action discovery and proposal validation used in production. The only substitution is a dry-run proposal dispatcher, so it cannot send a card or create a request while an admin is testing it. ``setup_turns`` is everything said BEFORE ``question``, as ``[{"role": "user"|"assistant", "content": str}, ...]``. Supplied rather than read from the session store because a simulation has no chat thread: it is how a caller reproduces 「这台」 — a follow-up whose referent lives in the previous message. It mirrors ``run_turn``'s history handling exactly, including the buyback exception (a live policy question is re-grounded from scratch, never from what an earlier turn happened to say), so a multi-turn preview and a multi-turn production chat take the same path. ``capture_evidence`` additionally returns every tool result verbatim. Off by default: the admin preview does not need it, and it is not free. """ from app.assistant import ( corroborate, # noqa: PLC0415 mcp_bridge, # noqa: PLC0415 ) from app.assistant import propose_tools as proposals_tools # noqa: PLC0415 from app.domains.support.copilot import ( # noqa: PLC0415 KB_TOOL_NAME, CopilotResult, run_copilot_turn, ) # The employee channel resolves its copy from the recipient, not from the # administrator viewing this page. Keep that parity even when the caller # does not supply a language explicitly. lang = lang or resolve_lang(person) trace_id = str(uuid.uuid4()) bridged_calls: list[str] = [] cards_sent: list[str] = [] proposed_actions: list[dict[str, Any]] = [] seen = corroborate.Corroborator() kb_docs: list[dict[str, str]] = [] policy_conflicts: list[dict[str, str]] = [] evidence: list[str] = [] started_at = time.monotonic() pre = await build_preflight(db, question, person_id=person.id) kb_docs.extend(pre.docs) policy_conflicts.extend(pre.policy_conflicts) # Same rule as ``run_turn``: a live policy question is grounded from the # current KB every time and never from what an earlier turn said, so its # history is dropped. Everything else keeps it, which is the whole point of # accepting ``setup_turns``. prior_turns = [] if pre.drop_history else _clean_setup_turns(setup_turns) propose_dispatch = proposals_tools.make_dispatch( person=person, chat_id="admin-simulation", # Never used in dry-run mode. A non-empty sentinel makes the tool # surface identical to the employee channel without being routable. open_id="dry-run", lang=lang, trace_id=trace_id, cards_sent=cards_sent, corroborator=seen, dry_run=True, simulated_actions=proposed_actions, ) bridge_dispatch = mcp_bridge.make_dispatch(person=person, called=bridged_calls) extra_tools = [*proposals_tools.propose_tools(), *mcp_bridge.bridge_tools()] async def extra_dispatch( name: str, tool_input: dict[str, Any], session: Any, caller: Any, ) -> str: if name in (mcp_bridge.TOOL_SEARCH, mcp_bridge.TOOL_CALL): return str(await bridge_dispatch(name, tool_input, session, caller)) return str(await propose_dispatch(name, tool_input, session, caller)) def observe(name: str, raw: str) -> None: seen.observe(raw) if capture_evidence and raw: # Bounded per blob AND in total: a bridged listing tool can return # a megabyte, and this list is serialised into an evaluation row. if len(evidence) < _EVIDENCE_MAX_BLOBS: evidence.append(raw[:_EVIDENCE_MAX_CHARS]) if name != KB_TOOL_NAME: return try: for item in json.loads(raw) or []: if not isinstance(item, dict): continue article_id = str(item.get("id") or "") if article_id and not any(d.get("id") == article_id for d in kb_docs): kb_docs.append( { "id": article_id, "title": str(item.get("title") or ""), "url": str(item.get("url") or ""), "source": str(item.get("source") or ""), "updated_at": str(item.get("updated_at") or ""), } ) except (TypeError, ValueError): logger.debug("assistant: unusable KB result while simulating") try: forced = await _forced_software_role_lookup( db=db, person=person, question=question, trace_id=trace_id, called=bridged_calls, ) if forced is not None: result = CopilotResult( available=True, reply=forced, tool_calls=[], provider="helix-catalogue", ) else: result = await run_copilot_turn( db, person, [*prior_turns, {"role": ROLE_USER, "content": question}], extra_tools=extra_tools, extra_dispatch=extra_dispatch, system_prompt=feishu_system_prompt(lang, person=person) + pre.system_context, observer=observe, ) except Exception: # noqa: BLE001 - preview must never disrupt the admin page logger.exception("assistant: employee simulation failed person=%s", person.id) return SimulationResult( reply=canned.failed(lang), tool_calls=bridged_calls, docs=kb_docs, policy_conflicts=policy_conflicts, proposed_actions=proposed_actions, provider=None, available=True, turn_failed=True, evidence=evidence, latency_ms=int((time.monotonic() - started_at) * 1000), ) tool_calls = [name for name in (result.tool_calls or []) if name != mcp_bridge.TOOL_CALL] if pre.kb_prefetched: tool_calls.insert(0, KB_TOOL_NAME) tool_calls.extend(pre.identifier_tools) tool_calls.extend(bridged_calls) if capture_evidence: # The server-side preflights are evidence too, and they are the ONLY # evidence for a turn the model answered without calling anything — # which is the common shape for a policy question. Omitting them made # every legitimately pre-grounded citation look fabricated. evidence.extend(block[:_EVIDENCE_MAX_CHARS] for block in pre.evidence_blocks) # 🔴 The same last-pass repairs the employee's message goes through. # # ``run_turn`` flattens Markdown for a Feishu *text* message and appends the # destination link when the reply gestures at "the IT system" without one. # The simulation skipped both, so the harness judged raw model output — # ``**bold**``, headings, and the ``[[helix_mention:…]]`` marker — rather # than the delivered text, and never exercised the destination guarantee at # all. This module's own docstring says the ONLY substitution is the dry-run # dispatcher; this makes that true. delivered = add_missing_destination( to_feishu_text((result.reply or "").strip()), lang, ).strip() return SimulationResult( reply=delivered, tool_calls=tool_calls, docs=kb_docs, policy_conflicts=policy_conflicts, proposed_actions=proposed_actions, provider=result.provider, available=result.available, turn_failed=result.turn_failed, evidence=evidence, latency_ms=int((time.monotonic() - started_at) * 1000), ) async def run_turn( db: AsyncSession, *, person: IdentityPerson, chat_id: str, question: str, lang: Lang, trace_id: str, on_slow: Any = None, open_id: str = "", card_context: Any = None, extra_system: str = "", chat_kind: str = CHAT_KIND_DM, ) -> TurnResult: """One question in, one answer out, transcript written either way. ``on_slow`` is an awaitable factory invoked once if the turn is still running after :data:`HOLDING_NOTE_AFTER_SECONDS`. ``extra_system`` is appended to the channel prompt for this turn only — the group-chat conduct block is its one producer today. It is deliberately a string rather than a second prompt: the assistant in a room must be the same assistant as in a private chat, knowing one extra fact. When ``open_id`` is supplied the kernel additionally gets the whitelisted *propose* tools and the read-only MCP bridge. The propose tools only ever send a confirmation card, and the bridge can only reach tools the platform itself marks read-only — so the model can look things up and offer to act, but has nothing to reach for that would act on its own. """ from app.assistant import ( # noqa: PLC0415 corroborate, mcp_bridge, room_history, # noqa: PLC0415 ) from app.assistant import propose_tools as proposals_tools # noqa: PLC0415 from app.domains.support.copilot import ( # noqa: PLC0415 KB_TOOL_NAME, CopilotResult, run_copilot_turn, ) # Captured before anything can fail. A failed flush inside a tool expires # every instance in the session, so ``person.id`` read AFTER that point needs # IO and raises — including from the ``logger.exception`` line in the error # handler below, which is how the handler became the thing that crashed. person_id = person.id # Captured before the thread is read, so a 「新话题」 that lands while this # turn is running cannot be undone by this turn's own write-back. epoch = await session_store.current_epoch(person_id=person_id, chat_id=chat_id) prior = await session_store.load_history( db, person_id=person_id, chat_id=chat_id, ) # Continuity is decided from the thread as it was BEFORE this turn: a turn # that fails must not strand itself in a brand-new conversation row, and a # turn that succeeds must not re-read the history it just wrote. continuing = bool(prior) # A current policy question must be grounded again. In particular, a # historical assistant message is not an authority that can carry a stale # "two years" rule into a new buyback turn. Follow-up device selection # still has its normal history path because it does not match this policy # intent classifier. pre = await build_preflight(db, question, person_id=person_id) history = ( [{"role": ROLE_USER, "content": question}] if pre.drop_history else [ *prior, {"role": ROLE_USER, "content": question}, ] ) # Resolved BEFORE the dispatcher is built, and reused by ``_record`` below. # ``_open_conversation_id`` used to answer this by taking the most recent row # for the chat — but the row for THIS turn does not exist yet when the # dispatcher is built, so a proposal made in the first turn after the Redis # thread expired was filed against the previous, already-closed conversation # (or NULL for a first-ever thread). Joining the transcript to the action is # most of what 「可回溯」 means, and it was joining to the wrong one. conversation = await _current_conversation( db, person=person, chat_id=chat_id, continuing=continuing, chat_kind=chat_kind, ) from app.core.coordinators.assistant_support_coordinator import ( # noqa: PLC0415 ensure_active_assistant_request, ) service_request = await ensure_active_assistant_request( db, conversation=conversation, ) # Snapshot before the model/tool loop. A failed tool flush rolls the # session back and expires ORM instances; reading ``service_request.id`` # afterwards would trigger async lazy I/O in the failure path and suppress # the canned reply plus transcript this function promises to preserve. service_request_id = service_request.id # Committed before the model runs, not merely flushed. A failed turn rolls # this session back (see below), and a rolled-back conversation row leaves # ``_record`` inserting messages against a parent that no longer exists — # FK 1452, which turns "the model failed" into "the turn crashed and left no # transcript". The row has to exist for every outcome, which is what # ``_record``'s own docstring promises: a transcript that only records # successes cannot answer "the assistant did not answer me". await db.commit() # The ID, not the instance. A failed turn rolls this session back, and a # rollback EXPIRES every instance in it — so ``_record`` reading # ``conversation.turn_count`` afterwards triggers a lazy refresh and dies # with ``MissingGreenlet`` inside the very error path meant to keep the # employee informed. Read once here, while the instance is live. conversation_id = conversation.id extra_tools = None extra_dispatch = None bridged_calls: list[str] = [] cards_sent: list[str] = [] # What the model actually read this turn. A ``*_id`` it puts on a card has # to have come from here; see ``corroborate``. seen = corroborate.Corroborator() if open_id: propose_dispatch = proposals_tools.make_dispatch( person=person, chat_id=chat_id, open_id=open_id, lang=lang, trace_id=trace_id, cards_sent=cards_sent, corroborator=seen, # Without this every chat-created proposal stored # ``conversation_id = NULL``, so the admin console could show the # transcript OR the action but never join them — which is most of # what「可回溯」 is supposed to mean. ``escalation.py`` was the only # caller that passed it. conversation_id=conversation_id, ) bridge_dispatch = mcp_bridge.make_dispatch( person=person, called=bridged_calls, ) extra_tools = [ *proposals_tools.propose_tools(), *mcp_bridge.bridge_tools(), ] # 🔴 **群里再多一个工具:读这个房间。** # # 这个能力原来只挂在字面命令上(群里发「总结」两个字)。于是员工用一句 # 自然语言问同一件事 —— 「看一下群里的聊天 然后总结一下什么问题」—— # 落到通用问答,而模型手里没有任何读群的手段,于是它如实回答「我无法 # 查看群聊历史」。那句话对**平台**是不准确的,而一个「我做不到」的回答 # 在平台其实做得到的时候比错误答案更贵:员工不会再问第二次。 # # ``chat_id`` 从这一轮的信封进闭包,**不进参数** —— 与身份同一条规矩: # 模型说不出别的房间的名字,也就读不到别的房间。 in_room = chat_kind != CHAT_KIND_DM and bool(chat_id) if in_room: extra_tools.append(room_history.read_room_tool()) async def extra_dispatch( name: str, tool_input: dict[str, Any], session: Any, caller: Any, ) -> str: """Route a caller-supplied tool to whichever half owns it.""" if name == room_history.TOOL_READ_ROOM: if not in_room: # 单聊里这个工具压根没给出去;真到了这里就是模型在猜名字。 return "This tool only exists inside a group room." return await room_history.dispatch_read_room( db, chat_id=chat_id, tool_input=tool_input, ) if name in (mcp_bridge.TOOL_SEARCH, mcp_bridge.TOOL_CALL): return str(await bridge_dispatch(name, tool_input, session, caller)) return str(await propose_dispatch(name, tool_input, session, caller)) kb_docs: list[dict[str, str]] = list(pre.docs) def _note_tool(name: str, raw: str) -> None: """Remember what the tools returned: sources, and every id seen.""" seen.observe(raw) if name != KB_TOOL_NAME: return try: for item in json.loads(raw) or []: url = (item or {}).get("url") if url and not any(d["url"] == url for d in kb_docs): kb_docs.append({"title": item.get("title") or url, "url": url}) except Exception: # noqa: BLE001 # A tool result we cannot read costs the citation, never the turn. # Broad on purpose: this runs inside the provider loop, and # anything escaping is caught by ``run_copilot_turn``'s catch-all, # which throws away the whole answer and returns the canned # "unavailable" fallback. A missing source link is a far cheaper # failure than a lost reply. logger.debug("assistant: unusable KB result while collecting sources") started = time.monotonic() holding = _schedule_holding_note(on_slow) try: forced_reply = await _forced_software_role_lookup( db=db, person=person, question=question, trace_id=trace_id, called=bridged_calls, ) if forced_reply is not None: # A canonical relationship lookup is deterministic. Do not spend a # model hop reinterpreting it — the production failure this guards # against was a non-empty catalogue result being turned into a # false "not found" after several model/tool iterations. result = CopilotResult( available=True, reply=forced_reply, tool_calls=[], provider="helix-catalogue", ) else: result = await run_copilot_turn( db, person, history, extra_tools=extra_tools, extra_dispatch=extra_dispatch, # ``person`` is what anchors "today" to the employee's own # timezone; without it the model dates their request from its # training year. system_prompt=( feishu_system_prompt(lang, person=person) # The card the employee replied to, when they replied # to one. Same shape as the other groundings, and # first among them: it is what makes a bare 「这台」 # resolvable at all, so it must be in scope before # the model reads the question. + ( "\n\n" + card_context.as_prompt_block() if card_context is not None else "" ) # 通道级的追加守则(当前唯一的产出者是群聊:「这句话有旁人 # 在听」)。放在接地证据**之前**:它约束的是这一轮怎么说话, # 而证据约束的是说什么 —— 一段被埋在证据后面的行为守则,在 # 长上下文里最先被忽略。 + (f"\n\n{extra_system.strip()}" if extra_system.strip() else "") + pre.system_context ), observer=_note_tool, ) except Exception: logger.exception("assistant: turn failed person=%s", person_id) # Rolled back, not merely logged. Neither ``run_copilot_turn`` nor its # tool dispatcher rolls back, so a DBAPI error inside any bridged tool # leaves this session needing one — and the very next statement is the # transcript write below. Without this the write raised # ``PendingRollbackError`` straight out of ``run_turn``, past the delivery # branch in ``feishu_events``: the employee got NO message at all (not # even the canned failure), no transcript row, and no audit row. The # sibling paths in ``card_actions`` and ``propose_tools`` both do this, # and their comments name this exact exception. await db.rollback() # A rollback EXPIRES every instance in the session, ``person`` included — # so the next read of ``person.id`` (the history write-back here, the # audit row in ``feishu_events`` after this returns) triggers a lazy # refresh and raises ``MissingGreenlet``. Measured, not assumed: probing # this path showed ``db.new`` already empty after the rollback (pending # objects are discarded, so there is nothing to expunge) and the expired # ``person`` to be the whole remaining hazard. with contextlib.suppress(Exception): await db.refresh(person) result = None finally: _cancel_holding_note(holding) latency_ms = int((time.monotonic() - started) * 1000) if result is None: turn = TurnResult(OUTCOME_FAILED, canned.failed(lang), [], None, latency_ms) elif result.turn_failed: # Configured and on, but this call did not work. Recorded as FAILED # rather than ANSWERED — a provider outage in the helpful-rate # denominator makes the whole metric read low for a reason that has # nothing to do with the answers. Distinct from UNAVAILABLE, which # means the kernel is off: the employee is told to try again, not that # the feature does not exist. turn = TurnResult( OUTCOME_FAILED, canned.failed(lang), [], result.provider, latency_ms, ) elif not result.available: # The kernel is switched off or unconfigured. Say so with a next step # rather than pretending the question was understood. turn = TurnResult( OUTCOME_UNAVAILABLE, canned.kernel_unavailable(lang), [], result.provider, latency_ms, ) else: outcome = OUTCOME_ANSWERED rich = (result.reply or "").strip() reply = to_feishu_text(rich).strip() card_only = bool(cards_sent) if card_only: # The card states what will be submitted, that nothing is submitted # until they confirm, and when it expires. A narration underneath it # is a second message about the thing they are already reading — so # nothing is sent. It is still *recorded*, because a turn stored as # empty vanishes from working memory (``load_history`` drops empty # content) and the model, seeing an unanswered request, offers the # same card again. reply = canned.card_offered(lang) elif _claims_card_sent_without_delivery(reply): # Never let model text manufacture evidence of a card send. In # particular, a user saying "创建吧" can cause the model to echo # the previous card narration without invoking a proposal tool. # There is no ticket until the card is delivered and confirmed. reply = canned.offer_failed(lang) rich = reply outcome = OUTCOME_FAILED elif not reply: # Recorded as FAILED, for the reason the ``turn_failed`` branch above # already argues: an outcome of ``answered`` puts this in the # helpful-rate denominator and makes the metric read low for a reason # that has nothing to do with the answers. The employee is shown the # canned failure either way — the record has to agree with them. reply = canned.failed(lang) outcome = OUTCOME_FAILED # The prompt asks for the link; this guarantees it. A reply that points # at "the IT system" with nowhere to tap is a dead end on a phone. reply = add_missing_destination(reply, lang) # Applied to both renderings or the card — the surface most employees # actually read — would be the one missing the link. rich = add_missing_destination(rich, lang) if rich else rich # Replace the bridge's meta-tool name with the tools it actually ran: # "call_helix_tool" in an audit record answers no question anyone has. tool_calls = [name for name in (result.tool_calls or []) if name != mcp_bridge.TOOL_CALL] if pre.kb_prefetched: tool_calls.insert(0, KB_TOOL_NAME) tool_calls.extend(pre.identifier_tools) tool_calls.extend(bridged_calls) turn = TurnResult( outcome, reply, tool_calls, result.provider, latency_ms, reply_rich=("" if card_only else rich), card_only=card_only, ) # Only a real answer earns a place in working memory: replaying "the # assistant is unavailable" as context would poison every later turn. if turn.outcome == OUTCOME_ANSWERED: await session_store.save_history( db, person_id=person_id, chat_id=chat_id, history=[*history, {"role": ROLE_ASSISTANT, "content": turn.reply}], expect_epoch=epoch, ) # ``kb_docs`` is passed rather than read off ``turn``: the turn only gains # its ``docs`` in the ``replace`` below, which needs the id this call # returns. Reading ``turn.docs`` here stores an empty list on every row. message_id = await _record( db, conversation_id=conversation_id, question=question, turn=turn, trace_id=trace_id, sources=_public_source_links(kb_docs), service_request_id=service_request_id, ) return replace(turn, message_id=message_id, docs=kb_docs) def _schedule_holding_note(on_slow: Any) -> asyncio.Task | None: """Run the holding note as its own task. The callback MUST NOT touch the turn's session: two coroutines driving one ``AsyncSession`` corrupt the connection, and cancelling this task mid-query would leave the turn unable to commit its own transcript. Callers pass a callback that opens its own session — see ``feishu_events._holding_note``. """ if on_slow is None: return None async def _wait_then_notify() -> None: try: await asyncio.sleep(HOLDING_NOTE_AFTER_SECONDS) await on_slow() except asyncio.CancelledError: # normal: the answer arrived first raise except Exception as exc: # noqa: BLE001 — a holding note must not break a turn logger.warning("assistant: holding note failed: %s", exc) return asyncio.create_task(_wait_then_notify()) def _cancel_holding_note(task: asyncio.Task | None) -> None: if task is not None and not task.done(): task.cancel() async def _record( db: AsyncSession, *, conversation_id: uuid.UUID, question: str, turn: TurnResult, trace_id: str, sources: list[dict[str, str]] | None = None, service_request_id: uuid.UUID | None = None, ) -> str | None: """Append the exchange to the durable transcript, and commit it. Written for every outcome, including failures: "the assistant did not answer me" is precisely the report an administrator needs to be able to check, and a transcript that only records successes cannot answer it. """ # Re-loaded rather than carried in: on the failure path the caller's rollback # expired its copy, and a stale instance here reads a ``turn_count`` that is # no longer true (or raises fetching it). conversation = await db.get(AssistantConversation, conversation_id) if conversation is None: logger.error( "assistant: conversation %s vanished before its transcript write", conversation_id, ) return None # Retried, because ``(conversation_id, seq)`` is UNIQUE (migration 519) and # ``seq`` comes from ``turn_count`` under a Redis lock that fails open. The # constraint is right — a silently interleaved transcript is unusable as an # audit record — but an uncaught 1062 here throws the WHOLE turn away: the # exception leaves ``run_turn``, past the delivery branch, and the employee # gets no answer at all. That is the exact failure the previous round fixed # from a different cause, so the constraint must degrade to "re-read the # counter and take the next pair", never to silence. for attempt in range(_SEQ_RETRIES): try: return await _write_exchange( db, conversation=conversation, question=question, turn=turn, trace_id=trace_id, sources=sources, service_request_id=service_request_id, ) except IntegrityError: await db.rollback() refreshed = await db.get( AssistantConversation, conversation_id, populate_existing=True, ) if refreshed is None: logger.error( "assistant: conversation %s vanished mid-write", conversation_id, ) return None conversation = refreshed logger.warning( "assistant: transcript seq collision on %s (attempt %d) — " "another turn ran concurrently, re-reading turn_count", conversation_id, attempt + 1, ) # Out of attempts. The answer still reaches the employee: the caller reads a # ``None`` message id as "no rating buttons", not as "no reply". logger.error( "assistant: gave up writing the transcript for %s after %d attempts", conversation_id, _SEQ_RETRIES, ) return None async def _write_exchange( db: AsyncSession, *, conversation: AssistantConversation, question: str, turn: TurnResult, trace_id: str, sources: list[dict[str, str]] | None, service_request_id: uuid.UUID | None, ) -> str: """One attempt at appending the pair. Raises ``IntegrityError`` on a clash.""" conversation_id = conversation.id base_seq = conversation.turn_count * 2 db.add( AssistantMessage( conversation_id=conversation_id, service_request_id=service_request_id, seq=base_seq, role=ROLE_USER, content=question, trace_id=trace_id, ), ) answer = AssistantMessage( conversation_id=conversation_id, service_request_id=service_request_id, seq=base_seq + 1, role=ROLE_ASSISTANT, # The model's own text, Markdown intact. ``turn.reply`` is the copy # flattened for Feishu *text* messages, and storing that was a lossy # round trip: the answer card renders Markdown, so re-rendering the # rated card from the transcript made tapping 👍 visibly strip the # formatting — it reads as the card breaking on tap. Flattening belongs # at each channel, not in the record. content=turn.reply_rich or turn.reply, tool_calls=turn.tool_calls or None, # Same reasoning as ``content`` above, one field over: the answer card # renders these links, and a 👍 tap REPLACES that card with one rebuilt # from this row. Not stored, the sources vanished the moment the employee # rated the answer — the tap deleted the citations it was approving. # The durable card-rebuild contract is deliberately smaller than the # prefetch/debug payload: a cited source needs only its label and link. # Do not leak article ids, mirror origin, or update timestamps into the # transcript JSON merely because server-side grounding carries them. sources=( [ { "title": source.get("title", ""), "url": source.get("url", ""), } for source in sources ] if sources else None ), outcome=turn.outcome, provider=turn.provider, latency_ms=turn.latency_ms, trace_id=trace_id, ) db.add(answer) conversation.turn_count += 1 conversation.last_message_at = datetime.now(UTC) await db.commit() # Read the id AFTER the commit: a rating button has to point at the stored # row, and the id is what the card carries back. return str(answer.id) async def _current_conversation( db: AsyncSession, *, person: IdentityPerson, chat_id: str, continuing: bool, channel: str = CHANNEL_FEISHU, chat_kind: str = CHAT_KIND_DM, ) -> AssistantConversation: """The open thread for this person and chat, or a new one. Segmented by the same working-memory window the model sees: once the Redis thread has expired, the next question starts a new conversation row, so the stored transcript is divided the way the employee experienced it. ``channel`` is part of the lookup, not decoration: the web page and the Feishu chat are separate threads for the same person, and matching on ``chat_id`` alone would be enough to keep them apart only for as long as nobody chooses a colliding key. """ if continuing: existing = ( ( await db.execute( select(AssistantConversation) .where( AssistantConversation.person_id == person.id, AssistantConversation.channel == channel, AssistantConversation.chat_id == chat_id, ) .order_by(AssistantConversation.last_message_at.desc()) .limit(1), ) ) .scalars() .first() ) if existing is not None: return existing from app.core.coordinators.assistant_support_coordinator import ( # noqa: PLC0415 create_assistant_service_request, ) service_request = await create_assistant_service_request( db, requester_id=person.id, ) conversation = AssistantConversation( person_id=person.id, service_request_id=service_request.id, channel=channel, chat_id=chat_id, chat_kind=chat_kind, ) db.add(conversation) await db.flush() return conversation