"""Regressions for the review round that followed the Feishu agent's first week. Each test here names the defect it prevents, because every one of them shipped: the reviews found them in code that had already passed a full suite. Grouped by the invariant they defend rather than by module — the point of a card is that what the employee saw is what ran, and three separate files can break that. """ from __future__ import annotations import pathlib import pytest from app.assistant.corroborate import Corroborator, unclaimed_labels from app.assistant.reply_hygiene import add_missing_destination, to_feishu_text from tests.support.card_shape import elements_of class TestAnIdOnACardWasLookedUp: """The card shows a NAME and executes an ID. Only provenance ties them.""" def _seen(self) -> Corroborator: c = Corroborator() c.observe( '[{"id": "7f3a-1", "name": "Zoom Pro"}, ' '{"id": "9c2b-2", "name": "Zoom Basic"}]', ) return c def test_an_invented_id_is_refused(self) -> None: assert unclaimed_labels( {"plan_id": "deadbeef", "plan_id_display": "Zoom Pro"}, self._seen(), ) == ["plan_id"] def test_an_id_read_from_a_lookup_passes(self) -> None: assert unclaimed_labels( {"plan_id": "7f3a-1", "plan_id_display": "Zoom Pro"}, self._seen(), ) == [] def test_a_real_id_under_the_wrong_name_is_refused(self) -> None: """The dangerous case: both halves are real, and the pairing is not. The employee approves 「Zoom Pro」 and gets Basic. Caught only because JSON is read as JSON — the two rows sit 30 characters apart, so any proximity rule would have passed this.""" assert unclaimed_labels( {"plan_id": "9c2b-2", "plan_id_display": "Zoom Pro"}, self._seen(), ) == ["plan_id"] def test_prose_results_still_corroborate(self) -> None: """Not every tool returns JSON, and a tool that returns a sentence must not make its flow unproposable.""" c = Corroborator() c.observe("套餐 Zoom Pro 的编号是 7f3a-1，Zoom Basic 是 9c2b-2") assert unclaimed_labels( {"plan_id": "7f3a-1", "plan_id_display": "Zoom Pro"}, c, ) == [] def test_an_id_that_claims_no_name_is_not_blocked(self) -> None: """``create_procurement_supplier.tax_id`` is a tax registration number the employee dictates — a ``*_id`` that is a VALUE, not a reference. The first rule demanded a lookup for it and made that flow permanently unproposable. Nothing on the card claims a name for it, so nothing on the card can be wrong about it.""" assert unclaimed_labels( {"name": "Acme Ltd", "tax_id": "91310000MA1FL0XXXX"}, Corroborator(), ) == [] def test_a_rejected_call_echoing_the_id_back_is_not_evidence(self) -> None: """NOT valid JSON, on purpose. A JSON error envelope is refused by the structural check anyway, so a JSON fixture passes with the refusal filter deleted and proves nothing. FastMCP's type errors come back as prose, and prose is what reaches the proximity fallback this protects.""" c = Corroborator() c.observe( "Error executing tool list_plans: 1 validation error, " "input_value='MY-FAKE-SENTINEL' for plan Zoom Pro", ) assert unclaimed_labels( {"plan_id": "MY-FAKE-SENTINEL", "plan_id_display": "Zoom Pro"}, c, ) == ["plan_id"] def test_a_real_result_that_merely_says_error_still_counts(self) -> None: """``"error": null`` is a successful listing that happens to carry the key. Discarding it told the model to look up what it had just looked up, and it would retry into the same result.""" c = Corroborator() c.observe('{"error": null, "items": [{"id": "7f3a-1", "name": "Zoom Pro"}]}') assert unclaimed_labels( {"plan_id": "7f3a-1", "plan_id_display": "Zoom Pro"}, c, ) == [] def test_a_nested_row_still_corroborates(self) -> None: """Strictly-same-object matching refused ``{"items": [{"plan": {"id": …}, "name": …}]}`` and bricked any flow whose lookup returns that shape. A record is one LIST ELEMENT.""" c = Corroborator() c.observe('{"items": [{"plan": {"id": "7f3a-1"}, "name": "Zoom Pro"}]}') assert unclaimed_labels( {"plan_id": "7f3a-1", "plan_id_display": "Zoom Pro"}, c, ) == [] def test_a_wrapper_does_not_merge_two_rows_into_one(self) -> None: """``[{"rows": [A, B]}]`` — one outer element holding two records. Read as a single record, row A's id pairs happily with row B's name, which is the exact mispairing this check exists to catch.""" c = Corroborator() c.observe( '[{"rows": [{"id": "7f3a-1", "name": "Zoom Pro"}, ' '{"id": "9c2b-2", "name": "Zoom Basic"}]}]', ) assert unclaimed_labels( {"plan_id": "9c2b-2", "plan_id_display": "Zoom Pro"}, c, ) == ["plan_id"] assert unclaimed_labels( {"plan_id": "7f3a-1", "plan_id_display": "Zoom Pro"}, c, ) == [] @pytest.mark.parametrize( "envelope", [ # The producer is ``mcp_bridge._call_tool``: # ``json.dumps(result, ensure_ascii=False, default=str)[:6000]``. # 29 read tools key their payload on ``items``, 26 on ``data`` — # a bare list is the MINORITY shape, and it is the only one that # survives a repair that closes brackets with a repeated single # character. Round 3 shipped the closer bug precisely because the # fixture was a bare list, and then shipped the FIX with a bare # list too, so the regression test was still blind. Parametrised # over the real shapes so it cannot happen a third time. "bare_list", "items", "data_items", "items_with_meta", ], ) def test_a_truncated_listing_still_gets_the_structural_rule( self, envelope: str, ) -> None: """``mcp_bridge`` hands the model a hard 6000-char slice with no marker. The cut blob does not parse, so without repair the check falls back to proximity matching for exactly the results where it matters most: a listing is truncated BECAUSE it is long, and a long listing is where row A's id and row B's name are plausible neighbours.""" import json rows = [ {"id": f"id-{i:03d}", "name": f"Plan {i}", "note": "x" * 120} for i in range(60) ] document = { "bare_list": rows, "items": {"items": rows, "total": 60}, "data_items": {"data": {"items": rows}}, "items_with_meta": {"items": rows, "meta": {"page": 1, "size": 60}}, }[envelope] full = json.dumps(document, ensure_ascii=False, default=str) cut = full[:6000] assert len(full) > len(cut), "fixture must actually truncate" corroborator = Corroborator() corroborator.observe(cut) # The complete prefix is recovered, so a real pairing still passes … assert unclaimed_labels( {"plan_id": "id-005", "plan_id_display": "Plan 5"}, corroborator, ) == [], f"{envelope}: a correct pairing was refused" # … and the crossed pairing proximity used to admit is refused. assert unclaimed_labels( {"plan_id": "id-006", "plan_id_display": "Plan 5"}, corroborator, ) == ["plan_id"], f"{envelope}: a crossed pairing slipped through" # A row that was cut away entirely was never actually seen. assert unclaimed_labels( {"plan_id": "id-059", "plan_id_display": "Plan 59"}, corroborator, ) == ["plan_id"], f"{envelope}: a cut-away row was accepted" def test_unrecoverable_json_corroborates_nothing(self) -> None: """Refusing beats reaching for a weaker test on the same bytes.""" c = Corroborator() c.observe('[{"id": "7f3a-1", "na') assert unclaimed_labels( {"plan_id": "7f3a-1", "plan_id_display": "Zoom Pro"}, c, ) == ["plan_id"] def test_a_proposal_with_no_ids_is_not_blocked(self) -> None: assert unclaimed_labels({"alias": "shuai4", "reason": "对外联系"}, Corroborator()) == [] class TestHygieneDoesNotDamageTheAnswer: """Stripping Markdown must not rewrite content that only looks like it.""" @pytest.mark.parametrize( "text", [ "计算 2**3 的结果是 8，另外 5**2 是 25", "邮箱 a**@x.com，手机 138**5678", ], ) def test_digit_adjacent_stars_survive(self, text: str) -> None: """``re.S`` with a greedy body let two unrelated ``**`` swallow everything between them — the masked phone number came out as ``1385678``, which reads as a real number and is not one.""" assert to_feishu_text(text) == text @pytest.mark.parametrize( ("raw", "expect"), [ ("**已完成**：3 台", "已完成：3 台"), ("你的**保修期**是 2027-03-01", "你的保修期是 2027-03-01"), ("你有 **12 个** 邮件组", "你有 12 个 邮件组"), ], ) def test_real_emphasis_is_still_stripped(self, raw: str, expect: str) -> None: """Chinese writes bold with no surrounding spaces, so the guard is digit-based rather than word-based — a ``(? None: assert to_feishu_text(text) == text @pytest.mark.parametrize( ("raw", "expect"), [ ("***重要***：请确认", "重要：请确认"), ("line1 **bo\nld** line2", "line1 bo\nld line2"), ], ) def test_triple_stars_and_wrapped_bold_are_stripped( self, raw: str, expect: str, ) -> None: """``***bold italic***`` is standard model output and ``_BOLD`` cannot match it — its content class excludes ``*``, so the inner marker has nothing to pair with and all six asterisks reached the employee. Bold wrapped across a line break failed for the same class of reason.""" assert to_feishu_text(raw) == expect @pytest.mark.parametrize( ("raw", "expect"), [("##保修信息\n内容", "保修信息\n内容"), ("###まとめ\n本文", "まとめ\n本文")], ) def test_a_space_less_cjk_heading_is_still_a_heading( self, raw: str, expect: str, ) -> None: """Requiring the space after ``#`` fixed ``#2946`` but left these markers on screen — Chinese and Japanese are routinely written without it. The predicate is "a space, OR a non-ASCII character".""" assert to_feishu_text(raw) == expect def test_a_heading_still_goes(self) -> None: assert to_feishu_text("## 标题\n正文") == "标题\n正文" def test_being_pointed_at_a_person_does_not_get_a_portal_link(self) -> None: """「IT 系统管理员」 is a human. Appending 「在 Helix 自助提交」 to a sentence that named one is a non-sequitur.""" assert add_missing_destination("请联系 IT 系统管理员", "zh") == "请联系 IT 系统管理员" def test_every_vague_platform_phrase_excludes_people(self) -> None: """All four alternatives, not three. The first cut taught only the Chinese 系统 branch about 管理员; 「IT 那边」 then shipped as the one alternative of four with no people guard, so 「找 IT 那边的同事」 still got 「在 Helix 自助提交」 bolted onto a sentence that named a person. """ for pointed_at_a_person in ( "这个建议找 IT 那边的同事帮你看一下", "请联系 IT 系统管理员", "请找 IT 部门的同事", "please ask the IT team", "IT 部門の担当者にご相談ください", ): assert add_missing_destination(pointed_at_a_person, "zh") == ( pointed_at_a_person ), pointed_at_a_person def test_every_vague_platform_phrase_still_gets_the_link(self) -> None: """And the guard must not disarm the rule for actual places.""" for pointed_at_a_place in ( "请到 IT 那边提交", "请到 IT 系统提交", "raise it in the IT portal", ): assert "helix.pingcap.net" in add_missing_destination( pointed_at_a_place, "zh", ), pointed_at_a_place def test_being_pointed_at_a_place_still_does(self) -> None: out = add_missing_destination("请到 IT 系统提交", "zh") assert "helix.pingcap.net" in out class TestATurnThatSentACardIsRemembered: """Storing "" for a card turn deleted the turn from working memory. These go through ``run_turn``, not through ``TurnResult(...)``. The first cut asserted ``card_only is True`` on a hand-built dataclass and that the stand-in string was non-empty — both stay green with the fix reverted, because neither touches the code that sets them. """ @pytest.mark.asyncio async def test_a_card_turn_records_itself( self, db_session, feishu_person, kernel, cards, sent, enabled_gate, ) -> None: """Trigger: ask for something the platform can do → a card goes out → 「好了吗」. With an empty reply stored, ``load_history`` drops the turn, the model sees two consecutive user messages, and offers the SAME card again.""" from app.assistant import conversation from app.assistant.session import load_history # A VPN application rather than a support ticket. The subject here is # working memory, and a ticket is the one action the server now gates # (first attempt refused unless the model searched — see # ``propose_tools._refuse_ticket_if_a_flow_fits``), which would make this # test fail for a reason that has nothing to do with what it asserts. kernel["state"]["dispatch"] = [ ( "propose_helix_action", { "action": "submit_vpn_application", "arguments": { "vpn_node": "CN4-HK", "usage_description": "需要访问内网 Grafana 排查告警", "resource_checklist": "grafana.internal", }, }, ), ] turn = await conversation.run_turn( db_session, person=feishu_person, chat_id="oc-card-mem", question="打印机坏了帮我报一下", lang="zh", trace_id="t-card-mem", open_id="ou_assistant_open_0001", ) assert cards, "the turn was supposed to send a confirmation card" assert turn.card_only is True # Nothing posted as text under the card the employee is reading. assert sent == [] history = await load_history( db_session, person_id=feishu_person.id, chat_id="oc-card-mem", ) # Survives the round trip through Redis, which filters empty content. assert history[-1]["role"] == "assistant" assert history[-1]["content"].strip() def test_the_stand_in_names_the_life_span(self) -> None: """Working memory (120 min) outlives the offer (60 min). Without the expiry in the text, the model spends the second hour telling the employee to tap a card that is already dead.""" from app.assistant import messages as canned for lang in ("zh", "en", "ja"): assert "60" in canned.card_offered(lang) class TestTheAdminPlaneIsNotProposable: """RBAC refuses these at execute time. A card offering them is still wrong: the refusal arrives after the employee already tapped.""" @pytest.mark.parametrize( "tool", [ "server_power_on", "resend_notification", "trigger_identity_sync_target", "set_next_employee_number", "offboard_jamf_device", "upsert_dynamic_email_group_rule", "set_sso_app_owner", "classify_cloud_iam_user", "auditor_run_evidence_query", # Inherit their authz by reference ("Same RBAC as # ``server_power_on``"), so they declare no role of their own — # and the family's other member was already excluded. "server_power_off", "server_restart", "cost_backfill", # IT operating its own queues, or plumbing that is not an action. "advance_onboarding_now", "complete_provisioning_task", "create_provisioning_bundle", "set_onboarding_hardware_allocation", "intelligence_execute_recommendation", "intelligence_update_recommendation", "presign_upload_put", "list_approval_flow_stages", # Read as self-service from the name; both act on a colleague. "handover_ai_tool_application", "reassign_my_handover", # Declare their gate as a capability PREDICATE rather than as a # role noun — 「Requires can_manage_security (global_admin / # auditor)」 and 「RBAC: ``can_view_finance`` OR ``can_read_cloud``」. # The first tightening pass required a trailing "role"/"authority" # word, so all three stayed proposable while their own siblings # (``create_access_certification``, ``decide_access_certification_item``) # had already been opted out by hand. The two cost tools also spend # a rate-limited provider audit-API call per tap. "complete_access_certification", "cost_resolve_resource_creator", "cost_retry_resource_creator", ], ) def test_it_is_not_offerable(self, tool: str) -> None: from app.assistant.derived_actions import get_derived assert get_derived(tool) is None @pytest.mark.parametrize( "tool", [ "create_email_alias", "create_support_ticket", "keep_my_idle_seat", "cancel_support_ticket", # Owner attestation: the RBAC clause names an admin, but it names # the OWNER first, and owners are ordinary employees. These are # exactly the flows that most deserve to be one tap in chat. "respond_cloud_scope_semiannual_confirm", "respond_server_owner_confirmation", "snooze_cloud_scope_idle_reminder", ], ) def test_self_service_survived_the_tightening(self, tool: str) -> None: from app.assistant.derived_actions import get_derived assert get_derived(tool) is not None def test_no_proposable_tool_gates_itself_on_an_admin_role(self) -> None: """Read the tool BODIES, not their descriptions. ``create_procurement_supplier`` and ``create_procurement_order`` call ``_require_global_admin`` as their first statement while declaring no role in their description — invisible to both the marker list and the pattern rule. Finding them took reading 61 tool bodies by hand, which is not a thing anyone will do again; this does it mechanically. A tool that refuses in its own body cannot be offered: RBAC would reject it AFTER the employee already tapped 「确认」, which is the precise defect ``OPT_OUT`` exists to prevent. """ import ast from app.assistant.derived_actions import derived_catalogue source = ( pathlib.Path(__file__).resolve().parents[2] / "app" / "mcp" / "tools.py" ).read_text() tree = ast.parse(source) # Derived from the file, never hand-listed. The first cut named three # helpers that do not exist (``_require_finance_read``, # ``_require_cloud_manage``, ``_require_auditor``) and missed eight # that do — including ``_require_finance_view`` with 38 call sites. A # scan that silently matches nothing is worse than no scan, because it # reads as coverage. gates = { node.name for node in tree.body if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)) and node.name.startswith("_require_") } assert len(gates) >= 10, ( f"only found {sorted(gates)} — the naming convention changed and " "this scan is no longer looking at anything" ) catalogue = set(derived_catalogue()) # Map REGISTERED tool name -> function, because the two diverge: # ``@mcp_server.tool(name="urge_request")`` decorates # ``async def urge_request_tool``. Matching on the function name alone # silently skipped that tool, and would skip any future admin tool # registered under a name != its function. registered: dict[str, ast.AST] = {} for node in ast.walk(tree): if not isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)): continue name = node.name for decorator in node.decorator_list: if not isinstance(decorator, ast.Call): continue for kw in decorator.keywords: if kw.arg == "name" and isinstance(kw.value, ast.Constant): name = str(kw.value.value) registered[name] = node assert "urge_request" in registered, ( "the decorator's name= is no longer being read; the scan is back " "to matching function names and will skip renamed tools" ) offenders = [] for tool_name, node in registered.items(): if tool_name not in catalogue: continue called = { ( c.func.id if isinstance(c.func, ast.Name) else c.func.attr ) for c in ast.walk(node) if isinstance(c, ast.Call) and isinstance(c.func, (ast.Name, ast.Attribute)) } & gates if called: offenders.append((tool_name, sorted(called))) assert offenders == [], ( "these are offerable but refuse in their own body — add them to " f"OPT_OUT: {offenders}" ) def test_the_catalogue_cannot_grow_unnoticed(self) -> None: """A ceiling, not an exact count — a new self-service flow SHOULD land here on the day it ships, and that is the design. What must not happen quietly is a jump. The enumerated exclusions above only prove the names somebody thought to list: at review time the catalogue held 72 entries and ``server_power_off``, ``server_restart``, ``cost_backfill``, ``advance_onboarding_now`` and seven others were all in it, while every name-based test passed. A bound is what noticed. If this fails because the platform genuinely shipped flows, read the diff of ``derived_catalogue()`` first, confirm every new entry is something an employee does FOR THEMSELVES, then raise the number. Raised 65 -> 68 on 2026-08-26, during the bounty rebuild's review. The three are the business owner's own side of a vulnerability report assigned to them: ``confirm_bounty_report`` (their verdict), ``update_bounty_remediation`` (their fix progress) and ``comment_on_bounty_report`` (asking the security team something), all added by #3184. Each is something an employee does FOR THEMSELVES about work they already own, which is exactly the test set above. This bound had been red since that PR, and the same review found all three were ALSO missing their curated card headings, so a Chinese-speaking owner was handed a card titled "confirm_bounty_report". Raised 68 -> 70 on 2026-08-26 with the approval-flow registry's half-yearly ownership round. The two are ``confirm_my_approval_flow`` (「这条流程还是我负责」) and ``transfer_my_approval_flow`` (「以后归某某 管」) — the exact two answers the half-yearly card asks of an owner about flows already recorded against them, which is the "something an employee does FOR THEMSELVES" test above. Naming somebody ELSE's owner is not here: that stays HTTP + global_admin. Both shipped with curated headings and parameter labels in the same PR, per the lesson the previous paragraph records. Raised 70 -> 71 the same day with ``request_new_approval_flow``: asking IT for an approval flow the company does not have yet. The employee is asking for themselves, and the two things that are NOT self-service — deciding the request and registering the built flow — stay on the HTTP surface behind ``global_admin``. Raised 71 -> 72 on 2026-08-27 with ``describe_my_approval_flow`` (#3283): the owner's one-line 用途说明 for a flow already recorded against them, the same self-scope write as ``PATCH /api/v1/me/approval-flows/{id}/description``. The name reads like a READ, which is the thing to check and not the thing to trust — the classifier does not read names. ``mutates=True`` on its own ``tool_runtime`` call is what put it here, the body calls ``svc.set_description`` and commits, and that service was deliberately split out of ``update_flow`` so the owner path cannot reach name / status / published. Self-scope is enforced in the body (``owner_person_id != rt.user.id`` → 404, never 403), so no RBAC refusal can arrive after the tap. It is the owner's own words about their own row: the test above, satisfied. It shipped WITHOUT its curated heading, so for a day a Chinese-speaking owner was handed a card titled "describe_my_approval_flow" — the same defect the bounty paragraph above records, two paragraphs after it was written down. ``card_labels.TITLES`` now carries it; both parameters (``flow_id``, ``description``) were already in ``FIELDS``. ``test_every_card_has_a_curated_heading`` was red alongside this bound and is the reason to look there FIRST when raising it: the two go red together, because a new proposable flow needs both. And the missing heading cost more than looks. ``search()`` ranks on the tool name plus its LOCALIZED TITLES, so an absent entry left the zh haystack holding "Describe my approval flow" — measured on the catalogue, 「帮我补一句说明」 (a phrase the tool's own description claims to answer), 「流程用途说明」 and 「改一下审批 流程的说明」 all ranked it nowhere, and the assistant would have offered a support ticket for a flow it holds. Discovery failing looks exactly like the flow not existing — which is the sentence this module's own docstring opens with. 2026-08-30: 72 → 75。GitHub ID 自助修改(迁移 579)带来三个员工/ HR Ops 面的写工具 —— ``change_my_github_id`` / ``cancel_my_github_id_change`` / ``decide_github_id_change``。 三条都补了 ``card_labels.TITLES``(否则卡片标题会渲染成工具名, 而中文检索也会因此找不到它们),参数 ``github_id`` 补进 ``FIELDS``。 2026-09-04: 75 → 76。安全宣贯的 ``acknowledge_security_awareness`` (「我已阅读并理解」的助手入口)。它随宣贯主题的 PR2 一起上线,但 **没有配 curated 卡片** —— 三条测试同时红,和上面 GitHub ID 那次 一模一样的形态。已补 ``TITLES`` 与参数 ``campaign_id`` 的 ``FIELDS``。 这个上界的作用正是这个:让「目录悄悄长大」变成一次必须解释的改动。 2026-09-05: 76 → 80。访问对象闭环 PR1/PR2 一共加了 20 个 MCP 写工具, 其中 **11 个**衍生成了可提议动作,而三条测试同时红 —— **和上面两次 一模一样的形态,第三次了**。这一轮的处理不是把上界抬到 87 了事: * **6 个进了 ``OPT_OUT``**。四个(``update_data_access_grant_lifecycle_settings`` / ``extend_data_access_grant`` / ``revoke_data_access_grant`` / ``update_api_key_idle_alert_config``)在自己函数体里 ``_require_global_admin`` / ``_require_asset_admin`` —— 提议给员工 等于给他一张提交时必然被拒的卡。另两个 (``approve_permission_request`` / ``reject_permission_request``) 是「替别人的申请做决定」,落在本文件早就写明的那一类:一句话就批, 而申请的证据不在批的人眼前,正是复核页要堵的治理洞。 * **4 个补了 curated 卡片**(``keep_my_idle_api_key`` / ``renew_my_data_access_grant`` / ``release_my_data_access_grant`` / ``respond_data_access_owner_confirmation``),参数 ``binding_id`` 进了 ``FIELDS``。这四个确实该让持有人自己在聊天里做。 * 顺带修掉一个不是同一批的:``register_onboarding_hire_device`` (#3457)同样是 global-admin only 却可提议 —— 闸是一份清单,修一半 等于对所有人还是红的。 所以上界是 80,不是 87:**差出来的 7 个不是「已解释」,是「本来就不该 在目录里」。** 这条测试真正的价值在这里 —— 它逼出的不是一次数字更新, 是一次「这些动作到底该不该出现在聊天面上」的复核。 2026-09-08: 80 → 81。``open_ticket_chat`` ——「把我这张工单变成一个飞书 群,把负责它的人拉进来」。**这一条确实该在聊天面上**:它是员工替自己 做的事(自己的工单、自己要的那条快路径),函数体里没有任何 ``_require_global_admin`` 之类的门(判据是「求助人 / 这张工单的处理人 / IT 管理员」,而员工走的正是第一条),而且它没有替任何别人做决定 —— 建群的后果是「多了一个房间」,不是「某人的权限变了」。 同 PR 配了 ``TITLES``(zh + en + ja)。参数只有 ``ticket_id``,它早就在 ``FIELDS`` 里 —— 但仍然核对了一遍:少一个参数标签,卡片上就会出现一行 裸的 ``ticket_id``。 2026-09-09:**仍是 81**。``set_my_language`` 当天加进来又当天删掉, 记在这里是因为下一个人很可能会想再加一次:改语言的 HTTP 孪生 (``PATCH /auth/me/profile``)**明确拒绝 PAT**(ADR:被偷走的 PAT 不该 能把人的语言悄悄换成他读不懂的那种、让他错过通知),而 MCP 的鉴权只有 PAT / MCP-OAuth,从来不是浏览器会话。按 ``reveal_my_contact`` 的先例 改成「非 jwt 一律拒绝」也不行 —— 那样它在任何可达上下文里都不会成功, 而**派生出来的卡片会在员工点完确认之后才拒绝**,正是这条上界所属的那 四道闸要防的形状。改语言的两条路是飞书帮助卡(飞书验签的操作者身份) 和网页个人设置页。 """ from app.assistant.derived_actions import derived_catalogue assert len(derived_catalogue()) <= 81 class TestTranscriptToolsStayOffTheChatBridge: def test_the_gap_queue_is_not_reachable_from_chat(self) -> None: """It returns other employees' verbatim questions and answers. Bridged, an admin asking 「哪些问题没解决」 would ship colleagues' free text to the model provider — the exact thing the other three are excluded for.""" from app.assistant.mcp_bridge import readonly_catalogue assert "assistant_unhelpful_answers" not in readonly_catalogue() class TestAProviderOutageIsNeitherAnAnswerNorAShutdown: """``available`` answers "is the copilot on", ``turn_failed`` answers "did this call work". One flag could not carry both, and each channel got the half it reads wrong. Recording an outage as ``answered`` put it in the helpful-rate denominator (Feishu). Recording it as ``available=False`` disabled the web panel's composer — which gates on exactly that flag — so a single transient error locked the employee out until a page reload. The fix keeps the kernel reported as ON and marks the TURN as failed. """ def test_a_transport_failure_still_reports_the_kernel_as_on(self) -> None: from app.domains.support.copilot import CopilotResult # The shape the exception branch of ``run_copilot_turn`` returns. outage = CopilotResult( available=True, turn_failed=True, reply="AI 助手暂时不可用", provider="bailian", ) assert outage.available is True, ( "the web composer gates on `available`; a failed turn must not " "disable it" ) assert outage.turn_failed is True def test_a_healthy_turn_is_not_marked_failed(self) -> None: from app.domains.support.copilot import CopilotResult assert CopilotResult(available=True, reply="hi").turn_failed is False @pytest.mark.asyncio async def test_the_feishu_channel_records_an_outage_as_failed( self, db_session, enabled_gate, feishu_person, sent, kernel, ) -> None: """Not ``answered`` (it would enter the helpful-rate denominator) and not ``unavailable`` (the feature is not switched off).""" from app.assistant.feishu_events import handle_message_event from tests.test_assistant.conftest import ( FakeCopilotResult, message_event, ) kernel["state"]["result"] = FakeCopilotResult( available=True, turn_failed=True, reply="AI 助手暂时不可用", ) outcome = await handle_message_event( db_session, message_event(text="我的电脑保修到什么时候", event_id="e-outage"), trace_id="t-outage", ) assert outcome == "failed", outcome class TestEveryCardSpeaksTheEmployeesLanguage: """The confirmation card is the platform's primary employee-facing write surface. It was rendering 34 of 60 headings as ``Rotate api key`` and 118 parameter rows as ``Monthly Budget Usd`` — to a Chinese reader, on the one screen where they are asked to approve something. Enumerated rather than sampled, because a per-name test only proves the names somebody thought to list, and the catalogue grows on its own. """ def _catalogue(self): from app.assistant.derived_actions import derived_catalogue return derived_catalogue() def test_every_card_has_a_curated_heading(self) -> None: from app.assistant import card_labels # These two have hand-written ``AssistantAction`` entries in # ``actions.py`` that carry their own titles. curated_elsewhere = {"create_support_ticket", "submit_vpn_application"} missing = sorted( k for k in self._catalogue() if k not in card_labels.TITLES and k not in curated_elsewhere ) assert missing == [], ( "these render an English snake-case heading on a zh card — add " f"them to card_labels.TITLES: {missing}" ) def test_every_parameter_row_has_a_label(self) -> None: from app.assistant import card_labels from app.assistant.derived_actions import ( DISPLAY_SUFFIX, SERVER_GENERATED_FIELDS, ) missing = sorted({ field for action in self._catalogue().values() for field in (action.schema.get("properties") or {}) if not field.endswith(DISPLAY_SUFFIX) and field not in SERVER_GENERATED_FIELDS and field not in card_labels.FIELDS }) assert missing == [], ( "these show an English label on a zh card — add them to " f"card_labels.FIELDS: {missing}" ) def test_a_clipped_value_says_it_was_clipped(self) -> None: """Silent truncation made the card lie by omission: a 965-character announcement body rendered as 300 characters ending mid-sentence, and the full 965 were submitted — while the help page promises the card spells out everything that will be submitted.""" from app.assistant.derived_actions import _MAX_VALUE_CHARS, _render_value rendered = _render_value("公司公告正文。" * 140, "zh") assert len(rendered) > _MAX_VALUE_CHARS assert "980" in rendered class TestPluralIdFieldsGetTheSameRule: """``_ids`` was outside the invariant entirely. Seven flows took a list of ids with no companion names, so raw UUIDs went on the card and nothing corroborated them: ``request_email_group.additional_owner_ids`` asked the employee to approve making 「4f2c-aaaa-0001, 9b71-bbbb-0002」 co-owners of a mail group. """ def _seen(self) -> Corroborator: import json c = Corroborator() c.observe(json.dumps({"items": [ {"person_id": "4f2c-0001", "name": "张三"}, {"person_id": "9b71-0002", "name": "李四"}, ]}, ensure_ascii=False)) return c def test_the_model_is_asked_for_one_name_per_id(self) -> None: from app.assistant.derived_actions import get_derived action = get_derived("request_email_group") assert action is not None spec = (action.schema.get("properties") or {}).get( "additional_owner_ids_display", ) assert spec is not None, "plural ids still have no companion label" assert spec.get("type") == "array" def test_names_in_the_wrong_order_are_refused(self) -> None: """Both halves real, the pairing not — the dangerous case.""" assert unclaimed_labels( { "additional_owner_ids": ["9b71-0002", "4f2c-0001"], "additional_owner_ids_display": ["张三", "李四"], }, self._seen(), ) == ["additional_owner_ids"] def test_the_right_order_passes(self) -> None: assert unclaimed_labels( { "additional_owner_ids": ["4f2c-0001", "9b71-0002"], "additional_owner_ids_display": ["张三", "李四"], }, self._seen(), ) == [] def test_a_short_name_list_is_refused(self) -> None: """N names beside M ids: the employee cannot tell which is which.""" assert unclaimed_labels( { "additional_owner_ids": ["4f2c-0001", "9b71-0002"], "additional_owner_ids_display": ["张三"], }, self._seen(), ) == ["additional_owner_ids"] def test_the_card_shows_the_names(self) -> None: from app.assistant.derived_actions import get_derived action = get_derived("request_email_group") rendered = " ".join( r.value for r in action.summarise( { "additional_owner_ids": ["4f2c-0001", "9b71-0002"], "additional_owner_ids_display": ["张三", "李四"], }, "zh", ) ) assert "张三" in rendered assert "4f2c-0001" not in rendered, "raw UUIDs are back on the card" class TestOneRecordStopsAtSiblings: """Round 3's rule shipped with no test at all. ``{"current_plan": {…}, "recommended_plan": {…}}`` is two records in one envelope. Flattening it let the current plan's id pair with the recommended plan's name — and ``current``/``recommended``, ``from``/``to``, ``before``/``after`` are everywhere. """ def _seen(self, blob: str) -> Corroborator: c = Corroborator() c.observe(blob) return c def test_sibling_objects_are_separate_records(self) -> None: blob = ( '{"current_plan": {"plan_id": "AAA", "name": "Zoom Basic"},' ' "recommended_plan": {"plan_id": "BBB", "name": "Zoom Pro"}}' ) assert unclaimed_labels( {"plan_id": "AAA", "plan_id_display": "Zoom Pro"}, self._seen(blob), ) == ["plan_id"] assert unclaimed_labels( {"plan_id": "BBB", "plan_id_display": "Zoom Pro"}, self._seen(blob), ) == [] def test_a_lone_nested_object_is_still_the_same_record(self) -> None: """Refusing this shape is what bricked a class of flows two rounds ago.""" blob = '{"items": [{"plan": {"id": "7f3a-1"}, "name": "Zoom Pro"}]}' assert unclaimed_labels( {"plan_id": "7f3a-1", "plan_id_display": "Zoom Pro"}, self._seen(blob), ) == [] class TestHygieneLeavesPathsAlone: """Round 3 traded a word guard for a digit guard and started eating globs.""" @pytest.mark.parametrize( "text", [ "日志在 /var/log/*.log 和 /tmp/*.tmp 里", # The case round 5's fix exists for, which round 5 did not test: # its closing lookahead admits ANY non-ASCII, and Chinese # enumerates with 、 — so the fix could be reverted and the suite # stayed green (367 passed). "请删除 *.log、*.bak 和 *.tmp 文件", "支持 *.jpg、*.png、*.gif 三种格式", "白名单：*.pingcap.com，黑名单：*.evil.com", "用 user*@pingcap.com 或 admin*@pingcap.com", "变量 a*b 乘以 c*d", "清理 *.tmp 和 *.log", "ls *.py; grep *.md", ], ) def test_globs_and_paths_survive(self, text: str) -> None: """On an IT help desk these are the common case, not the exotic one.""" assert to_feishu_text(text) == text @pytest.mark.parametrize( ("raw", "expect"), [ ("*斜体*测试", "斜体测试"), ("你的*重点*是这里", "你的重点是这里"), ("普通 *emphasis* 文本", "普通 emphasis 文本"), ], ) def test_real_emphasis_is_still_stripped(self, raw: str, expect: str) -> None: assert to_feishu_text(raw) == expect class TestTheDiscriminatorIsNormalisedBeforeTheCard: """``target_kind`` decides whether ``target_email`` means the employee.""" def _alias(self): from app.assistant.derived_actions import get_derived return get_derived("create_email_alias") def _base(self) -> dict: return { "alias_local_part": "shuai4", "alias_suffix": "pingcap.com", "reason": "对外项目联系", } def test_uppercase_still_binds_and_still_executes(self) -> None: """``USER`` drew a correct card and then failed after the tap, because the service pattern is ``^(user|group)$``. Case is not a decision the employee made.""" params = self._alias().normalise({**self._base(), "target_kind": "USER"}) assert params["target_kind"] == "user" def test_a_non_scalar_kind_is_refused_before_the_card(self) -> None: """``["user"]`` made the binding decision say "not self" while the card rendered 用户邮箱 — so the card asserted a personal alias pointing at a colleague's mailbox.""" from app.core.exceptions import AppException with pytest.raises(AppException, match="target_kind"): self._alias().normalise({ **self._base(), "target_kind": ["user"], "target_email": "boss@pingcap.com", }) class TestCaseIsOnlyNormalisedWhereWeKnowTheCanonicalForm: """Round 4 lowercased every ``*_kind``/``*_type`` to fix ONE field. It reached seven and broke two. ``submit_dns_request.record_type`` validates against ``Literal["A","AAAA","CNAME","TXT","NS"]``, so lowercasing left **no input that could succeed**: the card said 「记录类型: cname」 and the tap always failed — the very defect the change was made to prevent, recreated on a different flow. """ def test_an_uppercase_enum_is_left_alone(self) -> None: from app.assistant.derived_actions import get_derived params = get_derived("submit_dns_request").normalise({ "integration_id": "i-1", "integration_id_display": "内网 DNS", "purpose": "对外站点", "zone_name": "pingcap.net", "record_type": "CNAME", "name": "x", "value": "y.example.com", "action": "create", "reason": "测试", }) assert params["record_type"] == "CNAME" def test_free_text_an_auditor_reads_is_left_alone(self) -> None: """``cloud_account_type`` is documented as free text that 「audit reviewers see verbatim in the Feishu approval card」.""" from app.assistant.derived_actions import _tidy_value assert _tidy_value("cloud_account_type", "POC") == "POC" assert _tidy_value("data_type", "PII, Customer SSN") == "PII, Customer SSN" def test_the_one_discriminator_we_declared_is_normalised(self) -> None: from app.assistant.derived_actions import get_derived params = get_derived("create_email_alias").normalise({ "alias_local_part": "shuai4", "alias_suffix": "pingcap.com", "reason": "对外项目联系", "target_kind": "USER", }) assert params["target_kind"] == "user" def test_the_normalised_set_is_derived_not_hand_listed(self) -> None: from app.assistant.derived_actions import ( _CASE_NORMALISED_FIELDS, _CONDITIONAL_SELF_BOUND, ) assert _CASE_NORMALISED_FIELDS == { other for other, _ in _CONDITIONAL_SELF_BOUND.values() } class TestAClaimedNameIsJudgedWhateverItsType: """One square bracket defeated the entire invariant. ``unclaimed_labels`` type-checked the display value and ``continue``d on a mismatch, while the card renders **anything truthy**. Nothing type-checks the field either — ``_make_normalise`` does not, and ``_strip_display`` removes it before FastMCP could. So a fabricated name passed as ``["CEO 张三"]`` was never corroborated and still went on the card. """ def _seen(self) -> Corroborator: import json c = Corroborator() c.observe(json.dumps({"items": [ {"assignment_id": "prod-billing-key", "name": "prod-billing-key"}, ]}, ensure_ascii=False)) return c @pytest.mark.parametrize( "claimed", ["my personal test key", ["my personal test key"], 12345, {"name": "my personal test key"}], ) def test_a_fabricated_name_is_refused_in_any_wrapper(self, claimed) -> None: assert unclaimed_labels( {"assignment_id": "prod-billing-key", "assignment_id_display": claimed}, self._seen(), ) == ["assignment_id"] def test_the_real_name_still_passes(self) -> None: assert unclaimed_labels( { "assignment_id": "prod-billing-key", "assignment_id_display": "prod-billing-key", }, self._seen(), ) == [] def test_one_name_for_many_ids_is_refused(self) -> None: """It says nothing about WHICH id it names.""" assert unclaimed_labels( { "additional_owner_ids": ["id1", "id2"], "additional_owner_ids_display": "财务总监 王五", }, self._seen(), ) == ["additional_owner_ids"] class TestTheRepairCannotStallTheEventLoop: """Round 4 moved the budget so unclosable candidates stopped consuming it — correct in itself, but it removed the only bound on the walk-back, and each step costs an O(n) scan. Measured 265 ms for one blob and ~5 s for one card check, on asyncio, where that blocks every other employee's turn. Employee-triggerable: paste a few KB of ``}]`` into your own support ticket, then ask the assistant about it. """ #: The tail must be an UNTERMINATED STRING, not loose punctuation. My first #: fixture was ``"[" + "}]" * 2900``, which mismatches on the first #: character — every candidate fails in O(1) and the whole blob costs #: 1.6 ms with or without the bound, so the test passed either way. That is #: the same fixture blindness rounds 2–4 each shipped. What actually costs #: is a candidate that ends INSIDE a string: ``_close_open_brackets`` then #: scans to the end before returning None, once per boundary — O(n²). _HOSTILE = '{"items":[{"id":"a","body":"' + "}]" * 2900 def test_a_hostile_blob_is_abandoned_quickly(self) -> None: """Employee-triggerable: paste a few KB of ``}]`` into your own support ticket, then ask the assistant about it. This is asyncio, so the stall blocks every other employee's turn, not just theirs.""" import time from app.assistant.corroborate import _repair_truncated_json started = time.perf_counter() assert _repair_truncated_json(self._HOSTILE) is None elapsed_ms = (time.perf_counter() - started) * 1000 assert elapsed_ms < 15, f"took {elapsed_ms:.0f}ms — the scan is unbounded" def test_a_real_truncation_is_still_repaired(self) -> None: """The bound must not cost the recovery it exists to allow.""" import json from app.assistant.corroborate import _repair_truncated_json rows = [ {"id": f"id-{i:03d}", "name": f"Plan {i}", "detail": "{}" * 60} for i in range(60) ] full = json.dumps({"items": rows}, ensure_ascii=False) cut_at = full.index('"detail"', full.index("id-030")) repaired = _repair_truncated_json(full[: cut_at + 11 + 240]) assert repaired is not None assert len(repaired["items"]) >= 30 def test_the_repair_is_computed_once_per_blob(self) -> None: from app.assistant.corroborate import Corroborator c = Corroborator() c.observe(self._HOSTILE) c.saw_together("a", "b") c.saw_together("c", "d") assert len(c._repaired) == 1 # noqa: SLF001 — the cache IS the contract class TestAnIdListAlwaysCarriesItsNames: """Six of the seven plural params are optional, so tying the display's requiredness to the id's left round 4's fix reaching exactly one flow — omit the names and raw UUIDs go on the card, which was the whole defect.""" def test_ids_without_names_are_refused_before_the_card(self) -> None: from app.assistant.derived_actions import get_derived from app.core.exceptions import AppException with pytest.raises(AppException, match="additional_owner_ids_display"): get_derived("request_email_group").normalise({ "local_part": "it-oncall", "suffix": "pingcap.com", "name": "IT 值班组", "reason": "值班轮转通知", "additional_owner_ids": ["4f2c-0001", "9b71-0002"], }) class TestTheConfirmLinkIsNotACredential: """A Feishu card is forwardable and the signed token travels inside it. The native button is owner-bound (Feishu supplies a verified operator), but the same card carries a ``查看完整内容`` URL whose token *was* the whole authentication. ``load_for_token`` compares ``proposal.person_id`` against the token's own ``sub`` — both minted together — so the "bound to one person" claim was a tautology and anyone the card reached could submit in the sender's name, with the sender's RBAC. """ def test_no_route_accepts_the_token_as_permission_to_act(self) -> None: """The forwarded-card bypass stays closed — by removing the QUESTION. Rounds 6–9 each adjusted who may POST to the token-addressed confirm route and each adjustment produced the next defect, because the question itself was unanswerable: a page opened from a forwardable link, in a browser that provably cannot hold a Helix session, cannot know who is reading it. ``confirm_proposal`` was deleted rather than fixed a fifth time, and it must stay deleted. What this asserts is therefore the PROPERTY, not the absence of every POST. #3413 added ``confirm_own_proposal``, and it is a different shape in the way that matters: it is addressed by proposal id with no token anywhere in it, it requires a signed-in person, and ownership is re-derived from that person server-side. The old route's defect — "holding this string is permission to submit in someone else's name" — has nothing to attach to. So: no route may take a ``token``, and every POST must require a person. A future route that accepted a token again would fail here, which is the regression this class exists to catch. """ import inspect from app.assistant import router as assistant_router from app.core.dependencies import get_current_user assert not hasattr(assistant_router, "confirm_proposal") posts = [ route for route in assistant_router.router.routes if "POST" in getattr(route, "methods", set()) ] for route in posts: endpoint = route.endpoint parameters = inspect.signature(endpoint).parameters assert "token" not in parameters, ( f"{endpoint.__name__} takes a token — that is the deleted route's shape" ) depends = [ p.default.dependency for p in parameters.values() if hasattr(p.default, "dependency") ] assert get_current_user in depends, ( f"{endpoint.__name__} can act without a signed-in person" ) def test_the_render_route_opens_without_a_session(self) -> None: """It must, and round 7 briefly broke this. ``client.ts`` records a production observation (2026-05-21) that the iOS Feishu webview DROPS the refresh-token cookie, so silent re-auth fails there, and ``FeishuMobileGuard`` lists this path as auth-less so the one tap that matters is not an SSO bounce. Requiring a session stranded exactly the population the page serves. The disclosure is bounded by showing a non-owner the card's own truncation instead — see the tests below. """ import inspect from app.assistant.router import show_proposal from app.core.dependencies import get_current_user depends = [ p.default.dependency for p in inspect.signature(show_proposal).parameters.values() if hasattr(p.default, "dependency") ] assert get_current_user not in depends class TestTheAnswerCardCannotBeUsedToPlantALink: """Model output and KB titles are rendered into a ``lark_md`` element. Every other Helix card goes through the notification layer's escaper; this one did not. A document title containing ``](`` closes its own link and injects a second one — a phishing link wearing the assistant's voice, planted by anyone who can publish a KB article or name a ticket. """ def _card(self, **kw): from app.assistant.answer_card import build_answer_card return build_answer_card( reply=kw.pop("reply", "这是一个足够长的、接地气的回答，用来触发卡片渲染。"), message_id="m1", lang="zh", **kw, ) def test_a_title_cannot_close_its_own_link(self) -> None: import json rendered = json.dumps( self._card(docs=[{ "title": "看这里](https://evil.example.com) 点我领取", "url": "https://feishu/doc/2", }]), ensure_ascii=False, ) sources = [ e["content"] for e in elements_of(json.loads(rendered)) if e.get("tag") == "markdown" and "参考文档" in e.get("content", "") ][0] assert sources.count("](") == 1, "a second link was injected" def test_a_url_cannot_close_its_own_link_either(self) -> None: """Round 6 escaped the TITLE and left the URL beside it: the link also ends at the first ``)`` in the href, so ``…/a)[Click here](https://evil.com`` renders as two links and the second points anywhere. Same injection, one field over.""" import json rendered = json.dumps( self._card(docs=[{ "title": "Normal Doc", # NO SPACES. The first version of this payload had them, so # ``_safe_url``'s whitespace rule — which predates the paren # encoding this test exists for — rejected the whole source and # the assertions passed through the ``sources == []`` escape # hatch. Deleting the encoding left the test green. "url": "https://good.com/a)[Click](https://evil.com", }]), ensure_ascii=False, ) sources = [ e["content"] for e in elements_of(json.loads(rendered)) if e.get("tag") == "markdown" and "参考文档" in e.get("content", "") ] assert sources, "the source was dropped, so this proves nothing" assert sources[0].count("](") == 1, "a second link was injected" assert "%29" in sources[0] def test_a_legitimate_bracketed_url_survives(self) -> None: """A wiki path may genuinely contain a bracket — encode, don't drop.""" import json rendered = json.dumps( self._card(docs=[{"title": "正常文档", "url": "https://feishu/wiki/AbC(1)"}]), ensure_ascii=False, ) assert "%281%29" in rendered def test_a_non_http_source_is_dropped(self) -> None: import json rendered = json.dumps( self._card(docs=[{"title": "脚本", "url": "javascript:alert(1)"}]), ensure_ascii=False, ) assert "javascript:" not in rendered def test_inline_html_in_the_reply_is_escaped(self) -> None: import json rendered = json.dumps( self._card(reply="回答里带 标签，长度足够触发卡片渲染的一句话。"), ensure_ascii=False, ) assert "<b>" in rendered class TestEveryIdInAListIsNamed: """Round 5 required a truthy display CONTAINER. One blank entry inside it carried an uncorroborated id onto the card and into the service.""" def _seen(self) -> Corroborator: import json c = Corroborator() c.observe(json.dumps({"items": [ {"person_id": "REAL-1", "name": "Alice"}, {"person_id": "REAL-2", "name": "Bob"}, ]}, ensure_ascii=False)) return c @pytest.mark.parametrize( "display", [["Alice", ""], ["Alice", None], ["Alice", " "]], ) def test_a_blank_name_beside_a_real_one_is_refused(self, display) -> None: """``["Alice", ""]`` rendered 「Alice, 」 while TWO ids executed.""" assert unclaimed_labels( { "additional_owner_ids": ["REAL-1", "SMUGGLED-CEO-ID"], "additional_owner_ids_display": display, }, self._seen(), ) == ["additional_owner_ids"] def test_a_blank_id_under_a_fabricated_name_is_refused(self) -> None: assert unclaimed_labels( {"plan_id": " ", "plan_id_display": "编造的套餐"}, self._seen(), ) == ["plan_id"] @pytest.mark.parametrize( "display", [None, [""], ["Alice"], ["Alice", "Bob", "Carol"]], ) def test_normalise_demands_one_name_per_id(self, display) -> None: from app.assistant.derived_actions import get_derived from app.core.exceptions import AppException raw = { "local_part": "it-oncall", "suffix": "pingcap.com", "name": "IT 值班组", "reason": "值班轮转", "additional_owner_ids": ["A", "B"], } if display is not None: raw["additional_owner_ids_display"] = display with pytest.raises(AppException, match="additional_owner_ids_display"): get_derived("request_email_group").normalise(raw) class TestADisplayNameHasToBeReadable: """"One non-empty name per entry" measured emptiness with ``str(n).strip()``, which calls ``0`` and ``False`` names — and a tool result carrying a ``0``-valued field in the same record (a TTL, a count, a port) then corroborates them. The card renders 「解析记录: 0, 0」 and two ids are recertified under a label as uncheckable as the raw id.""" def _normalise(self, display): from app.assistant.derived_actions import get_derived action = get_derived("recertify_my_dns_records") required = [ k for k in (action.schema.get("required") or []) if k not in ("record_ids", "record_ids_display") ] raw = dict.fromkeys(required, "x") raw["record_ids"] = ["R1", "R2"] raw["record_ids_display"] = display return action.normalise(raw) @pytest.mark.parametrize( "display", [ [0, 0], [False, False], ["\u200b", "\u200b"], # zero-width space ["\ufeff", "\ufeff"], # BOM ["\u3000", "\u3000"], # ideographic space [["www"], ["api"]], [{"n": "www"}, {"n": "api"}], ], ) def test_an_unreadable_name_is_refused(self, display) -> None: from app.core.exceptions import AppException with pytest.raises(AppException, match="record_ids_display"): self._normalise(display) def test_real_names_pass(self) -> None: params = self._normalise(["www 记录", "api 记录"]) assert params["record_ids_display"] == ["www 记录", "api 记录"] class TestTheReadBridgeFindsToolsInChinese: """The registry is English; the employee is not. Production, 2026-08-18: asked 「我要申请 OPEN AI 的坐席」 — a product that IS in the software catalogue, with a working 申请坐席 button on the web — the model searched 「软件目录」, matched nothing, and told the employee OpenAI is not in Helix. 「OpenAI」 matched only the API-key tools, so it offered an API key instead: a confident wrong answer about their own platform. """ def test_the_lexical_net_still_cannot_read_chinese(self) -> None: """Pinning WHY semantic search is required rather than optional. CJK bigrams fixed this for the action catalogue because those titles are localized. Here the registry is written in English, so for a Chinese query with no counterpart in any description there is nothing for a tokenizer to match — 「邮件别名」 finds nothing however it is split. """ from app.assistant.mcp_bridge import _lexical_search assert _lexical_search("邮件别名") == [] assert [t.name for t in _lexical_search("software catalog")][:1] == [ "list_software_catalog", ] def test_a_phrase_written_into_a_description_is_found_lexically_too( self, ) -> None: """The half of the gap a description CAN close, and why it is worth closing twice. Semantic ranking is the general answer, but it depends on an embedding backend that has been down in production for a full day without anyone noticing. Where a description names the words employees actually use (「软件目录」/「负责人」, 「保修」), the lexical fallback reaches the same tool — so the degraded path is not the useless path. """ from app.assistant.mcp_bridge import _lexical_search assert [t.name for t in _lexical_search("软件目录")] == [ "list_software_catalog", ] assert "list_my_assets" in [ t.name for t in _lexical_search("我的电脑保修到什么时候") ] @pytest.mark.asyncio async def test_semantic_ranking_is_used_when_available( self, db_session, monkeypatch, ) -> None: from app.assistant import action_search, mcp_bridge action_search.reset_cache() monkeypatch.setattr( action_search, "semantic_search", lambda *_a, **_kw: _resolved(["list_software_catalog"]), ) found = await mcp_bridge.search(db_session, "软件目录") assert [t.name for t in found] == ["list_software_catalog"] @pytest.mark.asyncio async def test_a_dead_embedding_backend_is_asked_once( self, db_session, monkeypatch, ) -> None: """A failed warm-up used to be retried on every query — a whole-corpus embed with a 32-second timeout, inside a user-facing turn.""" from app.assistant import action_search action_search.reset_cache() calls = {"n": 0} async def _dead(*_a, **_kw): calls["n"] += 1 return None monkeypatch.setattr( "app.domains.support.copilot.embed_texts", _dead, raising=False, ) for _ in range(3): assert await action_search._ensure_vectors( # noqa: SLF001 db_session, action_search.CORPUS_READONLY, ) is False assert calls["n"] == 1, f"embedded {calls['n']} times, should be once" def _resolved(value): import asyncio future: asyncio.Future = asyncio.get_event_loop().create_future() future.set_result(value) return future