"""Proposable writes, derived from the platform's own MCP registry. The read bridge already borrows the MCP registry so the assistant learns new *reads* the day they ship. Writes were the asymmetry: each one needed a hand-written catalogue entry, so the assistant could only offer the three someone had got round to writing — and fell back to "I'll file you a ticket" for everything else, including flows the platform fully supports. An employee asking for an email alias got a support ticket while ``create_email_alias`` sat in the registry, unreachable. Three of the four responsibilities a hand-written action owns are already carried by the tool itself: * ``schema`` — the tool's own ``parameters``. * ``normalise`` — FastMCP validates and coerces against that schema on ``run()``, and the domain service validates the business rules behind it. * ``execute`` — ``tool.run()``, with the caller the server resolved. Only the **confirmation card** genuinely needs authoring, because its whole job is that the employee understood what they are approving. So that is the only thing this module leaves open: a derived action renders a generic card from the schema, and :mod:`app.assistant.actions` may override any tool with a curated one. Missing polish makes a card plainer — it no longer makes the flow unavailable. Safety is unchanged from the read bridge, and is not weakened by deriving: * the model may only ever *propose*; execution needs the employee's confirm against a hash of the parameters they were shown; * :data:`app.mcp.annotations.DESTRUCTIVE_TOOLS` is excluded outright; * :data:`OPT_OUT` removes tools that are technically employee-callable but wrong to trigger from a chat sentence, and any tool whose own description declares an admin-only authz rule is dropped by the same pass; * RBAC still runs inside the tool, against the person the server resolved, at execute time — deriving cannot widen what anyone may do. """ from __future__ import annotations import logging import re import uuid from dataclasses import dataclass from datetime import UTC, date, datetime, timedelta from functools import lru_cache from typing import Any from sqlalchemy.ext.asyncio import AsyncSession from app.assistant import card_labels from app.assistant.actions import AssistantAction, SummaryRow from app.core.locale import DEFAULT_LANG, Lang from app.identity.person_models import IdentityPerson logger = logging.getLogger(__name__) #: Every language a card is rendered in, typed so the comprehensions below #: stay inside ``Lang`` instead of widening to ``str``. _LANGS: tuple[Lang, ...] = ("zh", "en", "ja") #: Mutating tools that an employee may well be allowed to call, but which must #: not be reachable from a chat sentence. RBAC is not the reason for any of #: these — judgement about the surface is. OPT_OUT = frozenset( { # ── Acts in someone else's name ──────────────────────────────────── # The confirming employee is not the subject, so "you are about to submit # this" would be false on the card. # ── Admin-only, so offering it would be a card that refuses ─────── # (访问对象闭环 PR1/PR2, added to OPT_OUT 2026-09-05.) These four # call ``_require_global_admin`` / ``_require_asset_admin`` in # their own body. Derived-proposable is computed from "an # employee could call this", which they cannot — so without this # the assistant would put a card in front of an employee that # then refuses on submit. That is worse than not offering it: # the refusal reads as a platform bug, and the employee has no # way to know the action was never theirs to take. "update_data_access_grant_lifecycle_settings", "extend_data_access_grant", "revoke_data_access_grant", "update_api_key_idle_alert_config", # Same shape, different author (#3457's 备机登记 tool, 2026-09-05): # global-admin only in its own body. Fixed here rather than left # red — the gate is one list, and a half-fixed gate fails for # everybody. "register_onboarding_hire_device", "submit_buyback_on_behalf", # ── 不可逆:做完就没有回头路 ──────────────────────────────────── # ``disband_ticket_chat`` 删掉一张已结工单的飞书群,而**聊天记录随房间 # 一起消失**。这一档从来不上聊天面(见模块头的三档划分):一句话就能 # 触发的不可逆动作,和「确认一下参数」不是同一类风险 —— 员工点错的 # 代价是一段再也找不回来的过程记录。它在群里有自己的入口(「解散群」, # 那里工单终态是硬闸),在门户/后台有按钮,由每日收尾任务兜底。 "disband_ticket_chat", # ── Decides someone else's request ───────────────────────────────── # RBAC already limits these to people who hold the authority; the reason # they are off the chat surface is that a decision made from one sentence, # without the request's evidence in front of the approver, is the governance # hole the review screens exist to close. # # ``approve_permission_request`` / ``reject_permission_request`` # (访问对象闭环 PR2) are the same category: the caller is the # software owner / IT contact deciding somebody ELSE's request # for a permission. RBAC does allow a plain employee to hold that # authority, so they derive as proposable — but this file's own # policy is that an approval made from one sentence, without the # request's evidence in front of the approver, is the governance # hole the review screen exists to close. PR2 built that screen # on purpose; putting a one-tap chat card beside it would undo # the decision. "approve_permission_request", "reject_permission_request", # ``approve_my_api_key_release`` / ``reject_my_api_key_release`` are here # despite the "my" in their names, which is exactly the trap: the caller is # the requester's MANAGER, and approving synchronously revokes the # subordinate's credential. Classifying them from the name — as a first cut # of this file did — put a card titled "同意释放我的 API Key" in front of a # manager for an action that tears down someone else's key. "approve_my_api_key_release", "reject_my_api_key_release", "approve_cloud_grant", "reject_cloud_grant", "reject_cloud_scope_request", "decide_access_certification_item", "decide_cloud_grant_renewal", # A mover review packet asks the NEW MANAGER whether a colleague # still needs each of ~238 entitlements. Every property that makes # it a good MCP tool for a manager driving an AI client with their # own credentials makes it a bad one-line chat trigger: the caller # is not the subject, the decision is about somebody else's access, # and the whole point of the category view is that the counts and # examples are in front of you when you answer. RBAC is not the # reason it is here — the packet's own reviewer gate already holds. "decide_role_change_review_category", "decide_group_join_request", "decide_security_finding", "submit_access_review", "create_access_certification", "create_assessment_cycle", "create_vendor_assessment", "create_inbound_questionnaire", # ── Moves ownership between people ───────────────────────────────── # Same call the curated asset action already made: borrowing and repair are # self-scoped and reversible, while handing an asset or a key to someone # else belongs on a screen with the thing in front of you. "create_asset_transfer", "accept_asset_transfer", # Names a colleague and grants them visibility of your key's spend. Picking # the right person out of a chat sentence is the failure mode; the roster # screen shows who is already on it. "add_api_key_member", # Operate on one *named* employee, chosen by the caller — HR/IT maintenance, # not self-service, and none of them say "admin" in a way the marker rule # would catch. "hr_refresh_person_from_hris", "sync_person_to_identity_target", "sweep_leaver_cloud_assignments", "transfer_apikey_assignment_owner", "transfer_my_api_key", # ── Administrative bookkeeping ───────────────────────────────────── # Month-end close and onboarding checklists are operated from their own # cockpits, where the surrounding state is what makes the edit meaningful. "finance_close_set_task_status", "finance_close_set_task_comment", "delete_onboarding_ack_item", # ── Siblings that inherit their authz by reference ───────────────── # Named explicitly as well as pattern-matched. A description saying only # "same contract as the other lifecycle tools" declares nothing to read, so # the family's other members went one way and these went the other. Belt # and braces where being wrong means an employee is offered a card that # reboots production. "server_power_off", "server_restart", "cost_backfill", # ── IT operating its own queues and configuration ────────────────── # Read out the full proposable list and judged each by hand: no # description-derived rule catches these, because they declare no role and # some read as self-service from the name. # advance_onboarding_now force-flip a process to live (Day-1 rescue) # complete_provisioning_task ticking off a queue, not doing a thing # create_provisioning_bundle platform configuration # set_onboarding_hardware_allocation the admin twin of the manager-scoped # ``set_my_team_joiner_…``, which stays "advance_onboarding_now", "complete_provisioning_task", "create_provisioning_bundle", "set_onboarding_hardware_allocation", # ── Spends real model budget, or is an admin console action ──────── "intelligence_execute_recommendation", "intelligence_update_recommendation", # ── Not a flow an employee can be shown ──────────────────────────── # ``presign_upload_put`` is plumbing (it hands back an S3 URL), and # ``list_approval_flow_stages`` is a READ that happens to be marked # mutating for its cache write. A confirmation card for either asks the # employee to approve something that is not an action. "presign_upload_put", "list_approval_flow_stages", # ── Puts a duty or a seat onto a named colleague ─────────────────── # Both read as self-service from the name — one even starts with "my" — # and both act on someone else. ``handover_ai_tool_application`` moves an # approved seat to ``successor_person_id``; ``reassign_my_handover`` pushes # the caller's offboarding-handover duty onto ``new_person_id``, who finds # out by being assigned it. Neither belongs to the confirming employee, so # the card's promise — "this is what YOU are submitting" — is not true. "handover_ai_tool_application", "reassign_my_handover", # ── Platform surfaces whose description states no role ───────────── # The marker rule below reads the declared authz; these declare none, so # they have to be named. Each is IT/HR/audit plane: the cost of guessing # wrong on a chat card is an employee being offered 「重启云服务器」 or an # audit evidence bundle. RBAC would refuse at execution — that refusal # arriving AFTER a confirmation tap is the defect. "annotate_domain", "annotate_cdn_domain", "auditor_run_evidence_query", "draft_assessment_narratives", "classify_external_account", "cost_recompute_allocation", "mark_onboarding_blocker_resolved", "triage_cloud_efficiency_recommendation", # Each of these spends real model budget per call. A costly generation is # not something to trigger from a sentence that might have been a typo. "intelligence_generate_platform_analysis", "intelligence_send_recommendation", "cloud_scope_generate_ai_analysis", # ── Gate their OWN body, so no description can reveal it ─────────── # Both call ``_require_global_admin`` as their first statement while # declaring no role in their description, so neither the marker list nor # the pattern rule can see it — the only way to find these was to read the # 61 tool bodies. Procurement master data is also not "something an # employee does for themselves" under any reading: the card rendered # 「新增供应商」 with rows for 银行账号 and SWIFT dictated from a chat # sentence, and RBAC refused only AFTER the tap. "create_procurement_supplier", "create_procurement_order", } ) #: Tools whose own description declares an admin-only authz rule. #: #: Derived rather than hand-listed, for the reason this whole module exists: a #: hand list of admin tools rots against a registry that changes weekly, and the #: first cut of ``OPT_OUT`` proved that classifying a tool from its NAME gets it #: wrong. Platform configuration is not "something an employee does for #: themselves", and a chat card is a poor surface for it — the surrounding state #: is what makes such an edit meaningful. _ADMIN_ONLY_MARKERS = ( "admin only", "admin-only", # Production, 2026-08-17: ``calculate_annual_leave_balance`` says "HR-only" # and slipped through a list that only knew the word "admin". A privileged # surface does not have to use the word. "hr only", "hr-only", "it admin only", "global_admin only", "finance only", "finance-only", "auditor only", ) #: The phrasings tool authors actually use, which the substring list above #: cannot cover. Audited against the live registry: ``Restricted to #: global_admin``, ``Requires IT admin role``, ``Requires the 'hr' or #: 'global_admin' role``, ``RBAC: can_manage_cloud``, ``Requires cloud-manage #: authority``, and a bare leading ``Admin:``. Eighteen admin-plane writes were #: proposable because the list only knew the literal words "admin only". #: #: The capability-function alternative is its own branch, because requiring a #: trailing ``role`` / ``authority`` word missed the commonest shape of all. #: ``complete_access_certification`` declares 「Requires can_manage_security #: (global_admin / auditor)」 and stayed proposable, even though the two sibling #: campaign tools were opted out by hand — and the ``cost_*_resource_creator`` #: writes declare 「RBAC: ``can_view_finance`` OR ``can_read_cloud``」, a #: finance/cloud plane that spends a rate-limited provider API call per tap. #: Naming the ``can_*`` predicate IS the authz declaration; the sentence around #: it varies too much to enumerate. _ADMIN_ONLY_PATTERNS = re.compile( r"restricted\s+to\s+[^.\n]*(?:admin|hr|auditor|finance)" r"|requires\s+(?:the\s+)?[^.\n]{0,40}?" r"(?:admin|auditor|hr|finance|cloud-manage|can_manage)" r"[^.\n]{0,20}?(?:\brole\b|authority)" # Trailing ``[^.\n]*`` deliberately runs to the end of the clause: the # owner-alternative test below reads the MATCHED text, so a gate written # 「Requires can_manage_cloud, or the scope owner」 has to carry its own # escape hatch inside the match or it would be excluded by half a sentence. r"|(?:requires|rbac\s*:)[^.\n]{0,40}?`*\bcan_[a-z_]+[^.\n]*" r"|\brbac\s*:[^.\n]*(?:admin|can_manage|auditor)" # ``Same RBAC + provider coverage as ``server_power_on``.`` A tool that # inherits its authz BY REFERENCE has no colon and names no role, so every # branch above misses it — and these come in families. ``server_power_on`` # was excluded while ``server_power_off`` and ``server_restart`` stayed # proposable, which is the exact 「重启云服务器」 card the opt-out list # says it exists to prevent. r"|same\s+rbac\b" # ``Manage-cloud role required.`` — the same sentence, written backwards. r"|(?:role|permission|authority)\s+required" r"|^\s*admin\s*:", re.IGNORECASE | re.MULTILINE, ) #: An authz clause that names the OWNER (or the caller) as an alternative is #: not an admin-only gate — 「the CURRENT scope owner OR a cloud admin」 is a #: self-service action with an admin escape hatch, and the owner is usually an #: ordinary employee. Without this, the attestation flows that most need to be #: one tap in chat would be the ones excluded. #: #: Tested against the MATCHED CLAUSE, never the whole description. Scanning the #: full text let ``set_sso_app_owner`` — 「Requires IT admin role」 — stay #: proposable because a later sentence explains what an owner *is*. #: A sentence that describes what the tool is NOT, or points elsewhere. The #: marker in it is about a *different* surface, so it says nothing about who #: may call this one. _NOT_THIS_TOOL = re.compile( r"\bis not\b|\bnot a valid\b|\binstead\b|\bvia the\b|\buse\b\s+``", re.IGNORECASE, ) _OWNER_ALTERNATIVE = re.compile( r"\bowner\b|\bthe caller\b|\byour own\b|\bself-scoped\b", re.IGNORECASE, ) #: Reviewed exceptions to the marker rule above: the phrase appears, but scoped #: to *part* of the tool rather than to calling it at all. _ADMIN_MARKER_REVIEWED_KEEP = frozenset( { # Employees maintain their own catalogue entry; only the ownership and # identity FIELDS are admin-gated, and the service enforces that per field. "update_my_software_catalog_entry", } ) def _is_admin_only(description: str) -> bool: lowered = (description or "").lower() for marker in _ADMIN_ONLY_MARKERS: at = lowered.find(marker) if at == -1: continue # Judge the SENTENCE the marker sits in, not the whole description. # ``respond_inventory_item`` — a flow every employee hits during a # 盘点 sweep — was excluded because one sentence explains what the tool # is NOT: 「…is an asset-admin-only action done via the asset transfer # endpoints」. The regex branch below already scopes itself this way; # the substring pass did not. start = max(lowered.rfind(".", 0, at), lowered.rfind("\n", 0, at)) + 1 stop = min( x for x in ( lowered.find(".", at), lowered.find("\n", at), len(lowered), ) if x != -1 ) sentence = description[start:stop] if _NOT_THIS_TOOL.search(sentence) or _OWNER_ALTERNATIVE.search(sentence): continue return True match = _ADMIN_ONLY_PATTERNS.search(description) if match is None: return False return not _OWNER_ALTERNATIVE.search(match.group(0)) #: Rendered instead of a value that would be empty on the card. An employee #: must never be asked to approve a blank row. _EMPTY = "—" _MAX_VALUE_CHARS = 300 #: Appended when a value does not fit. 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 employee cannot be shown all #: of a 4000-character field on a phone, but they can be told that is what #: happened. _CLIPPED: dict[Lang, str] = { "zh": "…（全文将按原样提交，共 {n} 字）", "en": "… (submitted in full, {n} characters)", "ja": "…（全文がそのまま送信されます・{n} 文字）", } def _clip(text: str, lang: Lang = DEFAULT_LANG) -> str: """Shorten for the card, and say so.""" if len(text) <= _MAX_VALUE_CHARS: return text marker = (_CLIPPED.get(lang) or _CLIPPED[DEFAULT_LANG]).format(n=len(text)) return text[:_MAX_VALUE_CHARS] + marker #: Parameters that say WHO a request is for. The confirming employee is the #: only correct answer for these on a self-service surface, so the server fills #: them and they are removed from the schema the model sees — a value it cannot #: supply is a value it cannot get wrong. #: #: Production, 2026-08-17: the model set ``target_email`` to the ALIAS being #: created rather than the requester's mailbox. The card looked plausible, the #: service rejected it at confirm time, and the employee got "Submitting this #: request failed" after pressing the button. SELF_BOUND_FIELDS: dict[str, str] = { "target_email": "work_email", "person_id": "id", "requester_email": "work_email", "user_email": "work_email", "employee_email": "work_email", } #: Plumbing the employee neither decides nor should read. Generated by the #: server from the proposal, which also makes it correct: the key is stable per #: proposal, so a double-confirm cannot become a double submission. SERVER_GENERATED_FIELDS = frozenset({"idempotency_key", "request_uuid", "client_token"}) def _server_generated(schema: dict[str, Any]) -> list[str]: props = schema.get("properties") or {} return [name for name in props if name in SERVER_GENERATED_FIELDS] def _self_bound(schema: dict[str, Any]) -> list[str]: props = schema.get("properties") or {} return [name for name in props if name in SELF_BOUND_FIELDS] def _strip_self_bound(schema: dict[str, Any]) -> dict[str, Any]: """The model's copy of the schema, without the fields the server owns. A CONDITIONALLY bound field stays visible, because the server only owns it some of the time: stripping ``target_email`` outright left the model no way to name a group, so the group half of ``create_email_alias`` — which the tool's own description documents — could not be proposed at all. It keeps the field and is told when to leave it alone; ``_fill_self_bound`` still overwrites it in the self case, so the 2026-08-17 incident stays fixed. """ bound = set(_self_bound(schema)) | set(_server_generated(schema)) conditional = {n for n in bound if n in _CONDITIONAL_SELF_BOUND} bound -= conditional if not bound and not conditional: return schema props = {k: v for k, v in (schema.get("properties") or {}).items() if k not in bound} for name in conditional: other, self_values = _CONDITIONAL_SELF_BOUND[name] spec = dict(props.get(name) or {}) spec["description"] = ( f"{spec.get('description', '')} Leave this out when " f"'{other}' is {' or '.join(sorted(self_values))} — the server " "fills the employee's own address. Supply it only for the other " "cases, using the value the employee gave you." ).strip() props[name] = spec # A conditionally bound field is OPTIONAL to the model: the server supplies # it in the self case. ``_make_normalise`` re-requires it for the other # cases, where only the employee knows the value. required = [ r for r in (schema.get("required") or []) if r not in bound and r not in conditional ] return {**schema, "properties": props, "required": required} #: A self-bound field whose meaning depends on ANOTHER field. ``target_email`` #: is the requester's mailbox only when ``target_kind`` is a user; #: ``create_email_alias`` also accepts ``group``, and binding it regardless #: produced a card reading 「目标邮箱 = 你的公司邮箱」 beside 「目标类型 = 邮件组」 #: — self-contradicting, in front of an execution that could not succeed. The #: unconditional bind was itself a fix (the model once set this to the ALIAS #: being created); the answer is to keep the bind and scope it, not to drop it. _CONDITIONAL_SELF_BOUND: dict[str, tuple[str, frozenset[str]]] = { "target_email": ("target_kind", frozenset({"user"})), } #: Characters that occupy no width. A name made of them is not a name. _INVISIBLE = "\u200b\u200c\u200d\ufeff\u2060\u00a0\u3000" def _visible(text: str) -> bool: return bool(text.strip(_INVISIBLE).strip()) def _binds_to_self(name: str, params: dict[str, Any]) -> bool: """Whether this field really refers to the confirming person, here.""" rule = _CONDITIONAL_SELF_BOUND.get(name) if rule is None: return True other, self_values = rule supplied = params.get(other) # Absent means the flow's default, which is the self case for every rule # here — a group alias always says so explicitly. return supplied is None or str(supplied).strip().lower() in self_values def _fill_self_bound( schema: dict[str, Any], params: dict[str, Any], person: IdentityPerson, proposal_id: Any | None = None, ) -> dict[str, Any]: filled = dict(params) for name in _server_generated(schema): filled[name] = f"assistant-{proposal_id}" if proposal_id else str(uuid.uuid4()) for name in _self_bound(schema): if not _binds_to_self(name, filled): continue value = getattr(person, SELF_BOUND_FIELDS[name], None) if value is not None: filled[name] = str(value).strip().lower() if "email" in name else str(value) return filled _SELF_LABELS: dict[str, dict[Lang, str]] = { "email": { "zh": "你的公司邮箱", "en": "your work mailbox", "ja": "あなたの会社メール", }, "person": {"zh": "你本人", "en": "you", "ja": "あなた"}, } def _pick_self_label(field: str, lang: Lang) -> str: table = _SELF_LABELS["email" if "email" in field else "person"] return table.get(lang) or table[DEFAULT_LANG] #: Discriminators whose canonical values this module knows, because it #: declares them itself in ``_CONDITIONAL_SELF_BOUND``. Nothing else. _CASE_NORMALISED_FIELDS: frozenset[str] = frozenset( other for other, _ in _CONDITIONAL_SELF_BOUND.values() ) def _tidy_value(name: str, value: Any) -> Any: """Cheap, unambiguous shape fixes on values the model wrote. Only where the correction cannot be wrong: a *suffix* or *domain* field is the part AFTER the ``@``, so a leading ``@`` is always a mistake. Production sent ``alias_suffix="@pingcap.com"``, which the service concatenates into ``shuai4@@pingcap.com``. """ if isinstance(value, str) and name.endswith(("_suffix", "_domain")): return value.strip().lstrip("@") # ``target_kind="USER"`` bound the employee's mailbox correctly and drew a # correct card, then failed AFTER the tap on ``pattern=r"^(user|group)$"``. # Case is not something the employee chose — but this correction is only # safe where the canonical values are KNOWN, which is why it is limited to # the discriminators declared in ``_CONDITIONAL_SELF_BOUND``. # # Applying it by suffix (``*_kind``/``*_type``) reached seven fields and # broke two of them outright: ``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. It also silently downcased free text that auditors # read verbatim (``submit_ai_tool_application.data_type`` = "PII, Customer # SSN"). FastMCP flattens these Literals to plain ``"type": "string"``, so # the schema cannot tell us the canonical casing — which is exactly why # guessing it is not allowed. if isinstance(value, str) and name in _CASE_NORMALISED_FIELDS: return value.strip().lower() # A ``\n`` the model wrote as two characters. Same repair the curated # actions apply, and for the same reason: a derived card renders free text # the same way, so it showed the same visible escapes. if isinstance(value, str): from app.assistant.actions import ( # noqa: PLC0415 — cycle unescape_literal_newlines, ) return unescape_literal_newlines(value) return value #: Property key carrying a date field's direction, set by the MCP tool that #: owns the field (``FutureDate`` / ``PastDate`` in ``app/mcp/tools.py``). #: Read here rather than hand-listed: a list of "date fields" in the assistant #: would rot on the first new flow, and the constraint belongs with the field. _DATE_DIRECTION_KEY = "x_helix_date" _ISO_DATE = re.compile(r"^\d{4}-\d{2}-\d{2}$") #: The domain schemas allow a day of slack on a "not in the past" date for #: timezone skew between client and server; matched here so propose time and #: execute time agree. _PAST_DATE_SLACK = timedelta(days=1) def _check_date_directions( schema: dict[str, Any], cleaned: dict[str, Any], today: date, ) -> None: """Refuse a date pointing the wrong way, while the model can still fix it. An LLM has no clock: asked for 「有效期到 9 月 20 号」 it reaches for the year its training ended, and production got ``2024-09-20``. The domain schema rejects that too — but only at execute time, which is AFTER the employee has tapped confirm, and being asked to approve a request that cannot run is the exact failure this whole validation layer exists to prevent. The message names today so the model can correct itself rather than retry the same wrong year. """ from app.assistant.actions import _invalid # noqa: PLC0415 — cycle props = schema.get("properties") or {} for field, raw in cleaned.items(): spec = props.get(field) or {} direction = spec.get(_DATE_DIRECTION_KEY) if not direction or not isinstance(raw, str): continue text = raw.strip() if not text: # Empty means "not filled", which the required-fields check above has # already ruled on. Treating it as a malformed DATE told the model to # go and ask the employee for an optional birthday it had correctly # left blank — and the refusal message wrapped it as "this flow is the # right one, it is only missing a value". continue if not _ISO_DATE.match(text): raise _invalid( f"'{field}' must be an ISO date, YYYY-MM-DD. Today is {today.isoformat()}.", ) try: given = date.fromisoformat(text) except ValueError: raise _invalid( f"'{field}' is not a real date: {text!r}. Today is {today.isoformat()}.", ) from None if direction == "future" and given < today - _PAST_DATE_SLACK: raise _invalid( f"'{field}' is {text}, which is in the past — today is " f"{today.isoformat()}. Re-read what the employee said and " "resolve it against today's real date (a month and day with no " "year means the next time that day comes round). Do NOT file a " "support ticket over this; fix the date and offer the flow " "again.", ) # NO slack in this direction. The slack on the future branch mirrors a # real domain rule ("not more than a day in the past", for timezone # skew); there is no matching rule here to agree with — the only # past-direction field in the catalogue is a date of BIRTH, and nobody's # is tomorrow. Adding a day here just widened the one check that exists. if direction == "past" and given > today: raise _invalid( f"'{field}' is {text}, which is in the future — today is " f"{today.isoformat()}, and this field takes a past date.", ) #: Suffix of the companion field the model must fill next to every ``*_id``. #: The employee cannot read ``plan_id=7f3a…`` and cannot be asked for one #: either, so the model has to name what it resolved — which both makes the #: card checkable and proves it looked the id up instead of inventing it. DISPLAY_SUFFIX = "_display" def _id_fields(schema: dict[str, Any]) -> list[str]: """Every reference field, singular AND plural. ``_ids`` was missing, so seven flows put a raw UUID list on the card with no name beside it and nothing to corroborate: ``request_email_group.additional_owner_ids`` asked the employee to approve making 「4f2c-aaaa-0001, 9b71-bbbb-0002」 co-owners of a mail group, and ``respond_server_owner_confirmation.server_ids`` asked them to attest to machines identified only by UUID. Exactly the card this module's docstring calls 「correct on its face and does something else」. """ props = schema.get("properties") or {} return [ name for name in props if (name.endswith("_id") or name.endswith("_ids")) and not name.endswith(DISPLAY_SUFFIX) ] def _is_plural_id(name: str) -> bool: return name.endswith("_ids") def _augment_with_display(schema: dict[str, Any]) -> dict[str, Any]: """Ask the model for a human name beside every id it fills.""" ids = _id_fields(schema) if not ids: return schema props = dict(schema.get("properties") or {}) required = list(schema.get("required") or []) for name in ids: display = f"{name}{DISPLAY_SUFFIX}" if _is_plural_id(name): props[display] = { "type": "array", "items": {"type": "string"}, "description": ( f"One human-readable name per entry in {name}, IN THE SAME " "ORDER and the same length. The employee reads these on the " "confirmation card instead of the ids — a list of raw ids is " "something nobody can check before approving it." ), } else: props[display] = { "type": "string", "description": ( f"The human-readable name behind {name}, exactly as it appears in " "the catalogue you looked it up in (e.g. 'Figma 专业版'). The " "employee reads THIS on the confirmation card — never ask them " "for the id itself." ), } if name in required: required.append(display) elif _is_plural_id(name): # An OPTIONAL list of ids still puts raw UUIDs on the card when the # model omits the names — which is the whole defect. Six of the # seven plural params are optional, so tying the display's # requiredness to the id's left the fix reaching exactly one flow. # ``_make_normalise`` demands it only when the ids are present. props[display]["description"] += " REQUIRED whenever you fill the id list at all." return {**schema, "properties": props, "required": required} def _strip_display(params: dict[str, Any]) -> dict[str, Any]: """Companion labels are for the card; the tool only ever gets the ids.""" return {k: v for k, v in params.items() if not k.endswith(DISPLAY_SUFFIX)} @dataclass(frozen=True) class DerivedTool: name: str description: str schema: dict[str, Any] def _title_for(name: str, description: str) -> dict[Lang, str]: """Card heading, per language. The lexicon carries the employee-facing flows. Anything it does not cover falls back to a humanised TOOL NAME rather than the description's first sentence: "Add an alias to an email mailbox (target_kind='user') or..." is prose written for a model, and putting it in the heading slot is how a card starts looking machine-generated. """ del description # heading comes from the lexicon or the name, never prose return { lang: card_labels.title_label(name, lang) or card_labels.humanise(name) for lang in _LANGS } def _intent_for(name: str) -> dict[Lang, str]: """One localized sentence, built from the heading — never the description. The tool description is written for a model and reads like it ("Add an alias to an email mailbox (target_kind='user')..."). Production, 2026-08-17: that string landed verbatim on a Chinese confirmation card. The heading already says what this is, so the intent line only has to say what pressing confirm does. """ return { lang: _INTENT_TEMPLATE[lang].format( what=card_labels.title_label(name, lang) or card_labels.humanise(name), ) for lang in _LANGS } _INTENT_TEMPLATE: dict[Lang, str] = { "zh": "确认后将以你的名义提交「{what}」，走和网页申请完全一样的流程。", "en": "Confirming submits “{what}” in your name, through the same flow the web form uses.", "ja": "確認すると、あなたの名義で「{what}」を送信します（Web と同じフロー）。", } def _field_label(prop_name: str, spec: dict[str, Any]) -> dict[Lang, str]: """Label for one parameter row, per language. Parameter names repeat across the registry, so one lexicon entry serves every flow that takes that field — which is why this scales without a hand-written card per action. Unknown names degrade to a humanised schema title ("Alias local part"), which reads plain rather than broken. """ schema_title = str(spec.get("title") or "").strip() or None return {lang: card_labels.field_label(prop_name, schema_title, lang) for lang in _LANGS} def _render_value(value: Any, lang: Lang) -> str: """Values get the lexicon too: ``user`` on a card should read 用户邮箱.""" if value is None or value == "": return _EMPTY if isinstance(value, bool): return "true" if value else "false" if isinstance(value, (list, tuple)): joined = ", ".join(card_labels.value_label(str(v), lang) for v in value) return _clip(joined, lang) or _EMPTY return _clip(card_labels.value_label(str(value), lang), lang) def _make_summarise(schema: dict[str, Any]): props: dict[str, Any] = dict(schema.get("properties") or {}) bound = _self_bound(schema) plumbing = set(_server_generated(schema)) def _summarise(params: dict[str, Any], lang: Lang) -> list[SummaryRow]: """Every stored parameter appears — including ones not in the schema. The card's contract is that what the employee approved is what runs, so a value present in the proposal but absent from this list would be an unreviewed input. Schema order first (it reads in the order the flow thinks about them), then any extras. """ rows: list[SummaryRow] = [] for name in bound: # Named as the relationship rather than as an address: the server # binds it to the confirming person, so "your work mailbox" is both # truer and easier to check than a string of characters. # # Unless it is NOT bound this time — a conditionally bound field # holds a value the employee gave, and printing 「你的公司邮箱」 # over it made the card contradict its own neighbouring row. rows.append( SummaryRow( label=_field_label(name, dict(props.get(name) or {})), value=( _pick_self_label(name, lang) if _binds_to_self(name, params) else _render_value(params.get(name), lang) ), ), ) for name, spec in props.items(): if name in bound or name in plumbing or name.endswith(DISPLAY_SUFFIX): continue if name in params: # An id is unreadable and uncheckable. Show the name the model # resolved it from, under the id's own label, and keep the raw # value off the card entirely. display = params.get(f"{name}{DISPLAY_SUFFIX}") value = ( _render_value(display, lang) if (name.endswith("_id") or name.endswith("_ids")) and display else _render_value(params.get(name), lang) ) rows.append( SummaryRow( label=_field_label(name, spec if isinstance(spec, dict) else {}), value=value, ), ) for name, value in params.items(): if name not in props: rows.append( SummaryRow( label={"zh": name, "en": name, "ja": name}, value=_render_value(value, lang), ), ) return rows return _summarise def _is_blank(value: Any) -> bool: """Whether a supplied value amounts to nothing. Whitespace counts. ``" "`` is not a date, not a name and not a reason; the only thing it can do is get past a check written as ``== ""``. """ if value is None: return True # ``_visible`` rather than ``strip()``: zero-width and BOM characters strip # to nothing visible but are truthy, and this file already learned that once # (see ``_visible`` and its use on ``*_display`` lists). A required field of # ``"\u200b"`` otherwise renders as a blank row on the card and fails at # execute time — the same shape as the whitespace case, one character set over. return isinstance(value, str) and not _visible(value) def _make_normalise(schema: dict[str, Any]): props: dict[str, Any] = dict(schema.get("properties") or {}) required: set[str] = set(schema.get("required") or []) def _normalise(raw: dict[str, Any]) -> dict[str, Any]: """Drop unknown keys and reject missing required ones at propose time. FastMCP validates again at execute time, but failing here is what keeps a half-filled card from ever reaching the employee: being asked to approve a request that cannot run is worse than being asked one more question. """ from app.assistant.actions import _invalid # noqa: PLC0415 — cycle owned = set(_self_bound(schema)) | set(_server_generated(schema)) # A CONDITIONALLY bound field is only server-owned some of the time, so # it must survive this filter — the decision needs the sibling field, # which is in ``raw``. ``_fill_self_bound`` still overwrites it in the # self case, so a guessed value cannot reach the card there. owned -= {name for name in _CONDITIONAL_SELF_BOUND if not _binds_to_self(name, raw or {})} # Server-owned fields are DROPPED, not kept: a stored value the model # guessed would be rendered on the card and then overridden at execute # time, so the employee would approve one destination and another would # be used. Nothing stored, nothing to contradict. cleaned = ( {k: _tidy_value(k, v) for k, v in (raw or {}).items() if k in props and k not in owned} if props else {k: v for k, v in (raw or {}).items() if k not in owned} ) # ``str.strip()`` in the emptiness test, not just ``== ""``. A value of # " " passed this check AND the date check below (which skips an empty # ``strip()``), so a required date of one space reached the card and # failed at execute time — after the employee had tapped confirm, which # is the outcome this whole layer exists to prevent. missing = [k for k in required if k not in owned and _is_blank(cleaned.get(k))] if missing: raise _invalid( f"missing required field(s): {', '.join(sorted(missing))}. ASK " "the employee for these — one short question — then offer this " "same flow again. Do NOT switch to a support ticket: a missing " "detail is a question, not a reason to hand the request to a " "human who would have to ask it anyway.", ) # UTC plus the same one-day slack the domain schemas allow, which is # what absorbs the timezone gap between the employee's calendar and the # server's — the prompt is where "today" is stated per employee. _check_date_directions(schema, cleaned, datetime.now(UTC).date()) # A plural id list without its names would render raw UUIDs on the # card. Demanded here rather than in ``required`` because the ids # themselves are optional: the pair is all-or-nothing. for name in list(cleaned): if not _is_plural_id(name) or name.endswith(DISPLAY_SUFFIX): continue ids = cleaned.get(name) if not isinstance(ids, (list, tuple)) or not ids: continue shown = cleaned.get(f"{name}{DISPLAY_SUFFIX}") names = list(shown) if isinstance(shown, (list, tuple)) else [shown] # Every id needs a name of its own. Requiring only a truthy # CONTAINER let ``[""]`` and ``["Alice", ""]`` through: the first # rendered a blank row, the second rendered 「Alice, 」 beside two # ids — and in both cases an id nobody named went to the service. # ``str(n).strip()`` alone treats ``0`` and ``False`` as 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. Zero-width and BOM characters strip to # nothing visible but are truthy, so they are stripped explicitly. if len(names) != len(ids) or not all(isinstance(n, str) and _visible(n) for n in names): raise _invalid( f"'{name}{DISPLAY_SUFFIX}' must give ONE non-empty name per " f"entry in '{name}', in the same order — the employee " "cannot check a list of raw ids.", ) # Conditionally self-bound fields: optional in the schema because the # server owns them in the self case, but mandatory in the others — # nobody but the employee knows which group the alias points at, and # letting it through would fail at execute time, after the tap. for name, (other, self_values) in _CONDITIONAL_SELF_BOUND.items(): if name not in props: continue sibling = cleaned.get(other) if sibling is not None and not isinstance(sibling, (str, int, bool)): # ``target_kind=["user"]`` made ``_binds_to_self`` stringify a # list and conclude "not self" while ``_render_value`` unwrapped # it and printed 用户邮箱 — so the card asserted a personal alias # pointing at a colleague's mailbox. Refuse it here rather than # let the tool reject it after the employee already tapped. raise _invalid( f"'{other}' must be a single value, not {type(sibling).__name__}.", ) if _binds_to_self(name, cleaned): continue if not str(cleaned.get(name) or "").strip(): raise _invalid( f"'{name}' is required when '{other}' is " f"{cleaned.get(other)!r} — ask the employee for it.", ) return cleaned return _normalise def _make_execute(tool_name: str, schema: dict[str, Any]): async def _execute( db: AsyncSession, *, person: IdentityPerson, params: dict[str, Any], proposal_id: uuid.UUID, trace_id: str, ) -> dict[str, Any]: """Run the tool as the employee, exactly as the read bridge calls one. ``db`` is unused on purpose: the tool opens and commits its own session, the same way the HTTP surface does, so the domain keeps its own transaction, idempotency and audit behaviour untouched. """ from app.mcp.context import mcp_trace_id, mcp_user_id # noqa: PLC0415 from app.mcp.server import mcp_server # noqa: PLC0415 tool = mcp_server._tool_manager._tools.get(tool_name) # noqa: SLF001 if tool is None: raise RuntimeError(f"tool {tool_name} is no longer registered") user_token = mcp_user_id.set(person.id) try: trace_token = mcp_trace_id.set(uuid.UUID(str(trace_id))) except (ValueError, AttributeError, TypeError): trace_token = mcp_trace_id.set(uuid.uuid4()) try: result = await tool.run( _strip_display( _fill_self_bound(schema, dict(params), person, proposal_id), ), ) finally: mcp_user_id.reset(user_token) mcp_trace_id.reset(trace_token) return {"tool": tool_name, "result": _payload(result)} return _execute #: Enough for any tool's own response. Past it we are storing a page of text in #: a JSON column, not an identifier. _PAYLOAD_TEXT_LIMIT = 2000 def _payload(result: Any) -> Any: """Keep the tool's own return value, structured, when it has one. ``Tool.run`` is called with ``convert_result=False``, so ``result`` is the tool function's actual Python return value — for the write tools, a plain dict such as ``{"request_id": …, "request_number": …}``. Stringifying that threw away the only durable trace of *which row was created*: the id survived as text inside a repr, truncated, unqueryable, and impossible to read back without parsing a format nobody promised to keep. Non-dict returns are still stringified — there is no id to preserve in a bare string or a list, and a JSON column is not the place to smuggle one. """ if isinstance(result, dict): return result return str(result)[:_PAYLOAD_TEXT_LIMIT] @lru_cache(maxsize=1) def derived_catalogue() -> dict[str, AssistantAction]: """Every mutating MCP tool the assistant may offer, keyed by tool name. Built from ``mutates=`` — the same ground truth the read bridge inverts — so a flow becomes offerable on the commit that ships its tool, and stops being offerable on the commit that marks it destructive. """ from app.mcp import tools as _tools # noqa: F401, PLC0415 — force registration from app.mcp.annotations import DESTRUCTIVE_TOOLS, mutating_tools_from_source from app.mcp.server import mcp_server # noqa: PLC0415 flags = mutating_tools_from_source() catalogue: dict[str, AssistantAction] = {} for name, tool in mcp_server._tool_manager._tools.items(): # noqa: SLF001 if flags.get(name) is not True: continue # read-only or unknown → not a write to propose if name in DESTRUCTIVE_TOOLS or name in OPT_OUT: continue description = (getattr(tool, "description", "") or "").strip() if _is_admin_only(description) and name not in _ADMIN_MARKER_REVIEWED_KEEP: continue schema = getattr(tool, "parameters", None) or {} if not (schema.get("properties") or {}): # Nothing to show on a card. A confirmation with no parameters # cannot tell the employee what they are approving. continue catalogue[name] = AssistantAction( key=name, title=_title_for(name, description), intent=_intent_for(name), # All three read the SAME derived schema. Augmenting the raw # schema for the validator while the model was shown the stripped # one made them disagree: ``_augment_with_display`` appends # ``person_id_display`` to ``required`` for a self-bound # ``person_id`` that ``_strip_self_bound`` already removed, so the # validator demanded a field the model was never shown and the flow # became permanently unproposable. Latent only because today's one # self-bound ``*_id`` happens to be optional. schema=(model_schema := _augment_with_display(_strip_self_bound(schema))), normalise=_make_normalise(model_schema), # ``summarise`` deliberately gets the UN-stripped schema: the card # renders the self-bound fields as 「你的工作邮箱」, and it can only # do that if they are still in the schema it was built from. Do not # "make this consistent" with the two above — that silently drops # those rows from every card. summarise=_make_summarise(_augment_with_display(schema)), execute=_make_execute(name, schema), ) logger.info("assistant: %d write tools derived as proposable", len(catalogue)) return catalogue def get_derived(key: str) -> AssistantAction | None: return derived_catalogue().get(key) #: Words employees use for the same thing. Production, 2026-08-17: an employee #: asked for a 「邮**箱**别名」 and the catalogue title says 「邮**件**别名」 — one #: character apart, zero results, and the assistant fell back to a support #: ticket for a flow the platform has. Discovery failing looks exactly like the #: flow not existing. _SYNONYMS: dict[str, tuple[str, ...]] = { "邮箱": ("邮件", "mail", "email"), "邮件": ("邮箱", "mail", "email"), "别名": ("alias",), "邮件组": ("邮箱组", "group"), "坐席": ("席位", "seat", "license", "账号"), # The reverse direction was missing, so an employee who typed the English # word got nothing from the lexical fallback — 「坐席」 reached "license" # but "license" reached nothing. Semantic search covers this when the # embedding backend is up; the fallback is what runs when it is not. "license": ("坐席", "席位", "seat", "授权", "licence"), "licence": ("坐席", "席位", "seat", "license"), "seat": ("坐席", "席位", "license"), "席位": ("坐席", "seat"), "权限": ("access", "permission"), "申请": ("request", "apply"), "创建": ("create", "add", "新增"), "新增": ("创建", "create", "add"), "设备": ("电脑", "asset", "device", "笔记本"), "电脑": ("设备", "laptop", "asset"), # The generic approval submitter is intentionally described in English, # but employees ask for this flow in Chinese. Without this bridge the # no-embedding fallback misses a live buyback path and the model invents a # policy answer instead of asking which device to repurchase. "回购": ("buyback", "repurchase", "offboarding_buyback"), "买断": ("buyback", "repurchase", "offboarding_buyback"), "域名": ("domain", "dns"), "云": ("cloud",), "退订": ("cancel", "取消"), "取消": ("cancel", "退订"), } def _cjk_grams(text: str) -> set[str]: """Character bigrams for CJK, which has no spaces to split on. Without this, 「创建邮箱别名」 is one token that must appear verbatim in a title — so a phrase an employee actually types matches nothing. """ chars = [c for c in text if "\u4e00" <= c <= "\u9fff"] return {"".join(pair) for pair in zip(chars, chars[1:], strict=False)} | set(chars) def _expand(query: str) -> set[str]: lowered = query.lower().replace("，", " ").replace(",", " ") terms = {t for t in lowered.split() if len(t) > 1} terms |= _cjk_grams(lowered) for term in list(terms): terms.update(_SYNONYMS.get(term, ())) return {t for t in terms if t} def search(query: str, *, limit: int = 6) -> list[DerivedTool]: """Rank proposable writes against a free-text query. Lexical, but in the languages employees actually type: the haystack carries every localized title (not only the default one), CJK is matched on character bigrams rather than on whitespace tokens, and a small synonym map bridges the words people use to the words the catalogue happens to use. """ terms = _expand(query) if not terms: return [] scored: list[tuple[int, DerivedTool]] = [] for name, action in derived_catalogue().items(): titles = " ".join(action.localized_title(lang) for lang in _LANGS) haystack = f"{name} {titles}".lower() score = sum((3 if t in name else 0) + (2 if t in haystack else 0) for t in terms) if score: scored.append( ( score, DerivedTool( name=name, description=action.localized_intent(DEFAULT_LANG), schema=action.schema, ), ), ) scored.sort(key=lambda pair: (-pair[0], pair[1].name)) return [tool for _, tool in scored[:limit]]