"""The model's half of the write path: tools that *offer*, never tools that act. Each whitelisted action gets one ``propose_`` tool. Calling it validates the parameters server-side, stores a proposal and sends the employee a confirmation card — and returns to the model a plain statement that a card was sent. There is no tool here that submits anything, so a model that has decided to "just do it" has nothing to reach for. The tool name carries the same message as its behaviour: a model reading its own tool list cannot come away believing it can act unilaterally. """ from __future__ import annotations import json import logging from typing import Any from sqlalchemy.ext.asyncio import AsyncSession from app.assistant import confirm_card, corroborate, messages from app.assistant.actions import AssistantAction, all_actions, get_action from app.assistant.offer_delivery import FeishuOfferDelivery, OfferDelivery from app.assistant.proposals import create_proposal from app.core.exceptions import AppException from app.core.locale import Lang from app.core.rbac import is_active_employee from app.core.txn import best_effort from app.identity.person_models import IdentityPerson logger = logging.getLogger(__name__) _PREFIX = "propose_" def tool_name_for(action: AssistantAction) -> str: return f"{_PREFIX}{action.key}" def action_for_tool(name: str) -> AssistantAction | None: if not name.startswith(_PREFIX): return None return get_action(name[len(_PREFIX) :]) TOOL_ACTION_SEARCH = "search_helix_actions" TOOL_ACTION_PROPOSE = "propose_helix_action" _ACTION_SEARCH_LIMIT = 6 def propose_tools() -> list[dict[str, Any]]: """Whitelisted actions in the kernel's tool shape. Two shapes, one path. The curated actions get a named tool each, because naming them is what makes the model reach for the right one without searching. Everything else the platform can do is reachable through the search/propose pair — the same two-stage design the read bridge uses, and for the same reason: 150+ schemas do not fit in a prompt. """ generic = [ { "name": TOOL_ACTION_SEARCH, "description": ( "Find a Helix flow that can actually DO what the employee is " "asking for (create an email alias, request a seat, apply for " "access...). Search this BEFORE offering to file a support " "ticket: a ticket asks a human to do it by hand, while a flow " "found here submits the real request. Returns tool names and " "their parameter schemas." ), "input_schema": { "type": "object", "properties": { "query": { "type": "string", "description": "What the employee wants to do, in a few words.", }, }, "required": ["query"], "additionalProperties": False, }, }, { "name": TOOL_ACTION_PROPOSE, "description": ( "Offer one of the flows returned by " f"{TOOL_ACTION_SEARCH}. IMPORTANT: this does NOT submit " "anything — it sends the employee a confirmation card they " "must read and approve. Fill 'arguments' ONLY with values the " "employee actually gave you; never invent a plausible one to make " "the call succeed, or the card looks complete while containing " "words they never said. Missing a value? Ask for it instead." ), "input_schema": { "type": "object", "properties": { "action": { "type": "string", "description": f"Tool name exactly as {TOOL_ACTION_SEARCH} returned it.", }, "arguments": { "type": "object", "description": "Parameters for that tool, per its schema.", "additionalProperties": True, }, }, "required": ["action", "arguments"], "additionalProperties": False, }, }, ] return generic + [ { "name": tool_name_for(action), "description": ( f"{action.localized_intent('en')} " "IMPORTANT: this does NOT submit anything. It sends the employee " "a confirmation card they must read and approve. Only call it once " "the employee has ASKED you to do it and gave you every required " "value: a question about how something works is not a request to " "do it, and a value they never said must be asked for, never " "invented." ), "input_schema": action.schema, } for action in all_actions() ] #: A support ticket asks a human to do by hand what the platform may already #: do properly, so it is the one action that has to prove it is the last resort. #: #: Production, 2026-08-17: told 「帮我创建 shuai4@pingcap.com 的别名，用于项目对外 #: 联系」 — every value present — the model searched, saw the alias flow, and #: filed a ticket anyway. Four rounds of prompt wording did not change that; a #: named ``propose_create_support_ticket`` in the tool list is simply the path #: of least resistance. So the server decides instead of asking. _TICKET_ACTION = "create_support_ticket" class _TurnFlowMemory: """What this turn already learned about the real flows, one turn only. Built inside :func:`make_dispatch`, whose closure IS a single turn, so there is no lifetime to manage and nothing to reset. It exists because the interesting evidence is not "does a flow exist in the catalogue" — the old guard asked exactly that, and answered it badly: it returned nothing at all when no embedding backend was configured, and the lexical matcher it could have fallen back to offers 「椅子坏了」 → ``complete_api_key_rotation`` on CJK bigram overlap. The interesting evidence is what the model itself just did: production, 2026-08-18 16:19:54, ONE turn recorded ``["propose_helix_action", "propose_create_support_ticket"]`` — the model reached for the real API-key flow, was refused for a missing ``valid_from``, and filed a ticket instead, in the same turn. Nothing in the old guard could see that, because it only asked the catalogue. """ __slots__ = ("rejections", "searched", "tickets_refused") def __init__(self) -> None: #: Flow → the validation message it was refused with. A flow that failed #: on a MISSING FIELD is one question away from succeeding; it is not #: evidence that no flow fits. self.rejections: dict[str, str] = {} #: Flow names any search surfaced this turn. self.searched: list[str] = [] #: How many ticket attempts have already been turned away. self.tickets_refused = 0 def _refuse_ticket_if_a_flow_fits(memory: _TurnFlowMemory) -> str | None: """Turn away the FIRST ticket attempt when a real flow is in evidence. Three branches, strongest evidence first, and they answer different questions — which is the whole point. The old guard had none of them: it re-ran a catalogue match, so a flow the model had already reached for and merely under-filled looked identical to no flow at all, and with no embedding backend configured (production's state) it waved every ticket through. Refused exactly ONCE per turn. Not a deadlock: if the model reads the refusal, re-reads the candidates and still concludes a human has to do this by hand, the second attempt goes through. A guard that could never be satisfied would eventually deny a legitimate escalation, which is a worse outcome than a ticket that should have been a flow — and the reflex this exists to break only needs one forced reconsideration. """ if memory.tickets_refused: logger.info( "assistant: ticket allowed after a refusal — the model re-affirmed", ) return None if memory.rejections: # A flow was reached for and refused on its PARAMETERS. That is not # "no flow fits", it is "one question away". memory.tickets_refused += 1 logger.info( "assistant: ticket refused, these flows only lacked a value: %s", sorted(memory.rejections), ) return json.dumps( { "error": "the_flow_you_tried_only_needed_a_value", "flows_you_already_reached_for": memory.rejections, "note": ( "You already found the right flow this turn; it was refused " "for a missing or mis-formatted value, not because it was " "the wrong flow. ASK the employee for that value — one " "short question — then call the flow again. A ticket here " "hands a human the same question." ), }, ensure_ascii=False, ) if memory.searched: # The model searched and was shown candidates. Refusing names them, so # the refusal is actionable rather than a dead end. memory.tickets_refused += 1 candidates = list(dict.fromkeys(memory.searched))[:3] logger.info("assistant: ticket refused, search had shown: %s", candidates) return json.dumps( { "error": "a_real_flow_exists", "use_one_of": candidates, "note": ( "These came back from your own search this turn. Call " "propose_helix_action with whichever matches, asking the " "employee for any value you are missing. If none of them " "actually covers what they asked for, say so and file the " "ticket again — it will go through." ), }, ensure_ascii=False, ) # Never searched. The old guard re-ran a catalogue match here and named the # result — but the lexical matcher, which is what runs whenever no embedding # backend is configured, offers 「椅子坏了」 → ``complete_api_key_rotation`` # on CJK bigram overlap alone. Naming a nonsense flow is worse than naming # none: it invites the model to offer something absurd to the employee. So # this branch asserts nothing about the catalogue and asks for the one thing # that would produce real evidence. memory.tickets_refused += 1 logger.info("assistant: ticket refused, the model had not searched") return json.dumps( { "error": "search_before_filing_a_ticket", "search_with": TOOL_ACTION_SEARCH, "note": ( "You have not looked for a real flow this turn. Most requests " "have one, and it submits the actual request instead of asking " f"a human to do it by hand. Call {TOOL_ACTION_SEARCH} with what " "the employee wants. If nothing there covers it, file the " "ticket again — it will go through." ), }, ensure_ascii=False, ) def derived_actions_display_suffix() -> str: from app.assistant.derived_actions import DISPLAY_SUFFIX # noqa: PLC0415 return DISPLAY_SUFFIX def make_dispatch( *, person: IdentityPerson, chat_id: str, open_id: str, lang: Lang, trace_id: str, conversation_id: Any | None = None, cards_sent: list[str] | None = None, corroborator: Any | None = None, dry_run: bool = False, simulated_actions: list[dict[str, Any]] | None = None, delivery: OfferDelivery | None = None, ): """Build the dispatcher the conversation kernel calls for a propose tool. Bound to one person and one chat at construction time, so the identity the proposal is created for comes from the signed Feishu envelope rather than from anything the model produced. ``delivery`` is how the finished offer reaches the employee, and it is the ONLY thing that varies between the chat and the web page — see :mod:`app.assistant.offer_delivery`. Omitted, it stays what it always was: a card pushed to ``open_id``. Everything above it (validation, the active-employee gate, corroboration, the ticket guard, the stored proposal) is shared, deliberately, because those are the checks that decide what an employee may submit and a second copy of them is a second answer. """ delivery = delivery or FeishuOfferDelivery(open_id=open_id) # One turn's worth of memory: this closure is built once per turn. memory = _TurnFlowMemory() async def _dispatch( name: str, tool_input: dict[str, Any], db: AsyncSession, _person: IdentityPerson, ) -> str: if name == TOOL_ACTION_SEARCH: from app.assistant import ( action_search, # noqa: PLC0415 — cycle derived_actions, # noqa: PLC0415 — cycle ) query = str((tool_input or {}).get("query") or "") # Semantic first: a synonym table cannot cover 「坐席」/「license」/ # 「授权」/a typo, and discovery failing reads to the employee as the # flow not existing. Lexical stays as the net for when no embedding # backend is configured. ranked = await action_search.semantic_search( db, query, limit=_ACTION_SEARCH_LIMIT, ) if ranked is None: found = derived_actions.search(query, limit=_ACTION_SEARCH_LIMIT) else: found = [ derived_actions.DerivedTool( name=n, description=a.localized_intent("en"), schema=a.schema, ) for n in ranked if (a := derived_actions.get_derived(n)) is not None ] if not found: # Nothing close enough by meaning — try the words themselves # before concluding the platform cannot do it. found = derived_actions.search(query, limit=_ACTION_SEARCH_LIMIT) # Remembered so the ticket guard below can ask "what did YOU just # find" instead of re-running a catalogue match that may disagree # with what the model was shown. memory.searched.extend(t.name for t in found if t.name != _TICKET_ACTION) return json.dumps( { "actions": [ {"action": t.name, "what_it_does": t.description, "schema": t.schema} for t in found ], "note": ( "Call " + TOOL_ACTION_PROPOSE + " with one of these. If none " "fit, THEN offer a support ticket." ), }, ensure_ascii=False, ) if name == TOOL_ACTION_PROPOSE: requested = str((tool_input or {}).get("action") or "") action = get_action(requested) if action is None: return json.dumps( {"error": "unknown_action", "detail": requested}, ensure_ascii=False, ) tool_input = (tool_input or {}).get("arguments") or {} if not isinstance(tool_input, dict): return json.dumps({"error": "arguments must be an object"}, ensure_ascii=False) else: action = action_for_tool(name) if action is None: return json.dumps({"error": f"unknown tool {name}"}, ensure_ascii=False) # Read access is deliberately wider than write access: a leaver working # out their notice and a pre-hire whose Feishu account is already open # both get to ASK the assistant things. Neither may have a request # SUBMITTED for them — every domain service enforces # ``is_active_employee``, so a card offered here would be approved by the # employee and then refused at execute time, leaving them a red card and # no explanation of why. Refused before the card is built instead. if not is_active_employee(person): logger.info( "assistant: proposal refused, %s is not an active employee (%s)", person.id, person.employment_status, ) return json.dumps( { "error": "not_an_active_employee", "employment_status": str(person.employment_status or ""), "note": ( "This employee cannot have requests submitted in their " "name — a joiner's account activates on their first day, " "and a confirmed leaver loses submit rights immediately. " "Do NOT offer any flow or file a ticket for them: tell " "them plainly which of the two applies and that IT can " "help directly in the meantime." ), }, ensure_ascii=False, ) try: params = action.normalise(tool_input or {}) except AppException as exc: # Hand the validation message back so the model can ask the # employee for the missing or wrong field instead of retrying blind. # Recorded ONLY when this turn's own search surfaced the flow. The # first shape recorded every failed validation, so a flow the model # merely guessed at — 「椅子坏了」 matched ``complete_api_key_rotation`` # on CJK bigram overlap — made the ticket guard answer "you already # found the right flow, go and ask for the missing value", and the # assistant went off asking an employee with a broken chair for an # API key's start date. The guard may only assert what it knows. # Credited when the platform SHOWED the model this flow — either # this turn's own search surfaced it, or it is one of the curated # actions the prompt names outright. Both mean "advertised", which is # the distinction that matters: what must not be credited is a flow # the model produced from a loose lexical guess through # ``propose_helix_action``. ``create_asset_request`` is only reachable # as a named tool (it is not in the derived catalogue at all), so # keying on the search alone left one of the three curated flows # permanently unable to earn this branch. # ``action_for_tool(name) is not None`` was TAUTOLOGICAL here — the # branch above already returned for an unresolvable name — and # ``get_action`` falls back to the derived catalogue, so it admitted # all 60 derived flows rather than the 3 the prompt names. That put # the guessed-flow hole back through a different entrance: a model # emitting ``propose_`` earned "you already found the # right flow". Membership of the curated registry is the actual # question, so it is the one asked. curated_keys = {curated.key for curated in all_actions()} advertised = action.key in memory.searched or action.key in curated_keys if action.key != _TICKET_ACTION and advertised: memory.rejections[action.key] = exc.message return json.dumps( { "error": "invalid_parameters", "detail": exc.message, "note": ( "If this IS the flow the employee asked for, it is only " "missing or mis-formatted a value: ask them for it and " "call this same flow again — filing a support ticket " "instead hands a human the question you were one " "sentence away from answering. If it is NOT the right " "flow, say so and look for the right one; do not fill " "this one in to make it succeed." ), }, ensure_ascii=False, ) # An id the model did not read somewhere cannot be shown to the # employee under a name the model also wrote — the card would be # self-consistent and wrong. Send it back to look the value up. if unproven := corroborate.unclaimed_labels(params, corroborator): return json.dumps( { "error": "unverified_reference", "detail": ( "The name you gave for these does not match anything " f"you looked up in this conversation: {', '.join(unproven)}. " "The card shows the NAME and submits the ID, so they have " "to be the same row. Run the tool that lists them, take " "the id and the name from the SAME row, and try again." ), }, ensure_ascii=False, ) if action.key == _TICKET_ACTION: refusal = _refuse_ticket_if_a_flow_fits(memory) if refusal is not None: return refusal # The administrator's simulator deliberately runs through this exact # resolver, guard and validation path. It must never create a proposal # or contact Feishu, though: a preview is evidence, not an action. if dry_run: preview = { "action_key": action.key, "params": params, "would_offer_confirmation": True, } if simulated_actions is not None: simulated_actions.append(preview) return json.dumps( { "status": "confirmation_preview", "dry_run": True, "note": ( "This is an administrator preview. The employee would " "see a confirmation card for this action, but NO card " "has been sent and nothing has been submitted. State " "what would happen; do not claim a card was delivered." ), }, ensure_ascii=False, ) # Storing the proposal is the only step here that writes, and a failed # write deactivates the WHOLE transaction — the recovery reply below # reads the Feishu integration row through this same session, so # swallowing the error without isolating it raises # ``PendingRollbackError`` from inside the handler: the employee is # told nothing, and the model gets an exception instead of its "use # the portal" instruction. A bare ``db.rollback()`` would keep the # session usable but expire the caller's ``person``, whose attributes # the turn reads after this returns (``MissingGreenlet``) — so the # SAVEPOINT is the right primitive, and the commit stays outside it. issued = None async with best_effort( db, label=f"assistant proposal {action.key}", log=logger, ) as step: issued = await create_proposal( db, person=person, action_key=action.key, params=params, conversation_id=conversation_id, trace_id=trace_id, ) if not step.ok or issued is None: logger.error( "assistant: proposing %s failed for person=%s: %s", action.key, person.id, step.reason(), ) await delivery.notice(db, text=messages.offer_failed(lang)) return json.dumps( { "error": "proposal_failed", "note": ( "The employee has already been told to use the portal. " "Do not claim anything was submitted." ), }, ensure_ascii=False, ) await db.commit() try: url = await confirm_card.confirm_url( db, token=issued.token, digest=issued.proposal.params_hash, ) delivered = await delivery.offer( db, action=action, params=params, digest=issued.proposal.params_hash, url=url, lang=lang, proposal_id=str(issued.proposal.id), ) except Exception: # Past the commit the proposal exists, so a failure here is a # DELIVERY failure — reporting it as "nothing was submitted" would # be a lie the confirmation link then contradicts. Both steps are # read-only + outbound, so the session is still usable. logger.exception( "assistant: delivering the %s card failed for person=%s", action.key, person.id, ) delivered = False if not delivered: # Deterministic fallback rather than leaving the wording to a model # that might announce a submission that never happened. await delivery.notice(db, text=messages.offer_failed(lang)) return json.dumps( { "error": "card_not_delivered", "note": ( "The employee has already been told to use the portal. " "Do not claim anything was submitted." ), }, ensure_ascii=False, ) if cards_sent is not None: cards_sent.append(action.key) return json.dumps( { "status": "awaiting_confirmation", "note": ( "A confirmation card is now in front of the employee and it " "states everything: what will be submitted, that nothing is " "submitted until they confirm, and when the offer expires. " "Reply with an EMPTY string — narrating the card is a second " "message saying what they are already looking at." ), }, ensure_ascii=False, ) return _dispatch