"""Which real business object an assistant action created — declared, never guessed. An action's job ends when the domain service returns. What the employee asked is a different question: *did it actually happen*. Answering that needs a pointer from the proposal to the row the domain really created, and the pointer has to be right — a wrong one reports somebody else's status back to the employee as if it were theirs. The tempting shortcut is to scan the returned JSON for something that looks like an id. It does not work here, and the reason is not style. The platform's own write tools return five different shapes for the same idea: request_software_addon → {"id": …} request_software_seat → {"allocation_id": …, "approval_request_id": …} create_email_alias → {"request_id": …, "request_number": …} request_cloud_grant_renewal → {"request_group_id": …, "grant_count": 7} create_support_ticket → {"id": …, "ticket_number": …} A "first key ending in _id" heuristic has to choose between ``allocation_id`` and ``approval_request_id`` on row two with nothing to choose on; and row four is not one object at all — ``request_group_id`` covers ``grant_count`` separate grants, so "is it effective yet" means something different there. Asking the model to read the result and report the id is worse still: that is precisely the step where a confident wrong answer turns into a durable database row. So every covered action **declares** where its object is. An action with no declaration creates no link and records *why* (:data:`MISS_NO_LOCATOR`). The miss is a stored fact rather than an absent row, because "we linked nothing" and "we never tried" have opposite fixes and are otherwise indistinguishable from the outside. What this module deliberately does not do: interpret status. It answers *which row*, and stops. Reading that row's real state — and re-checking that this person may see it — belongs to the outcome coordinator, against the domain's own tables. """ from __future__ import annotations import re from dataclasses import dataclass from typing import Any #: Why no object was linked. Stored, not logged-and-forgotten: an assistant #: funnel that cannot separate "this action has no declaration yet" from "the #: declaration is wrong" cannot be acted on. MISS_NO_LOCATOR = "no_locator" MISS_NO_RESULT = "no_result" MISS_ID_KEY_ABSENT = "id_key_absent" MISS_ID_BLANK = "id_blank" #: The id is not a shape an identifier takes. Today every one of these comes #: from our own domain services and is a UUID or a short key, so this cannot #: fire — which is exactly why it is checked rather than assumed. The id is #: interpolated into a URL path, and one containing ``/`` or ``..`` composes a #: link that walks out of the resource it is supposed to point at. MISS_ID_UNSAFE = "id_unsafe" #: The link write itself blew up. Distinct from the misses above: those #: mean the contract is incomplete, this means the contract was fine and #: the database was not — opposite fixes. MISS_LINK_ERROR = "link_error" @dataclass(frozen=True) class OutcomeLocator: """Where one action's business object is, in that action's own result.""" #: Stable name for the kind of row. Not a table name — several actions #: land in ``employee_requests`` and the projection treats them alike. resource_type: str #: The exact key. One key, declared, no fallbacks: a locator that tries #: ``id`` then ``request_id`` would silently keep working while pointing #: at the wrong column after a tool changed shape. id_key: str #: Which resolver reads this row's live state. resolver_key: str #: The human-facing number, when the tool returns one. Employees say #: "REQ-1042", not a UUID. number_key: str | None = None #: Server-composed deep link template. ``{id}`` / ``{number}``. The model #: never authors a URL. #: #: Every value here is checked against the frontend's own route table by #: ``tests/test_mobile/test_deep_link_contract.py``, and it has to be: the #: first version of this table invented ``/me/requests``, ``/me/software``, #: ``/me/cloud`` and ``/support/tickets/:id``, **none of which are routes**. #: Nothing else goes red for that — the backend just stores a string — and #: the employee, who is reading this in Feishu on a phone, gets a 404. #: Note the app mixes two prefixes on purpose: support is under ``/me/``, #: the self-service pages under ``/my/``. deep_link: str | None = None #: Set when one id stands for several rows (cloud renewal). The count is #: carried so the projection can say "5 of 7 renewed" instead of #: pretending a batch is a single object. count_key: str | None = None @dataclass(frozen=True) class LocatedObject: resource_type: str resource_id: str resolver_key: str number: str | None = None deep_link: str | None = None covers: int | None = None # ── Resource types ─────────────────────────────────────────────────── # # Deliberately coarser than the tables behind them. ``employee_request`` is one # type covering alias / group-create / group-join because the projection reads # them the same way; splitting it would multiply resolvers without changing a # single answer an employee gets. RES_EMPLOYEE_REQUEST = "employee_request" RES_SUPPORT_TICKET = "support_ticket" RES_API_KEY_REQUEST = "api_key_request" RES_SOFTWARE_SEAT = "software_seat_allocation" RES_SOFTWARE_ADDON = "software_addon_request" RES_VPN_APPLICATION = "vpn_application" RES_DATA_ACCESS_REQUEST = "data_access_request" RES_CLOUD_GRANT_RENEWAL = "cloud_grant_renewal_group" RES_USER_PERMISSION = "user_permission" #: Every action the assistant may offer that creates something worth following. #: #: Keyed by ``action_key`` — the curated action name, which for derived actions #: is the MCP tool name. Adding an entry is the whole cost of bringing a new #: flow into the lifecycle; leaving one out degrades to "submitted, unlinked", #: which is honest. LOCATORS: dict[str, OutcomeLocator] = { # ── support ────────────────────────────────────────────────────── "create_support_ticket": OutcomeLocator( resource_type=RES_SUPPORT_TICKET, # ``ticket_id``, not ``id``. Two code paths answer to this one action # key and they do not agree: the MCP tool returns ``id`` (via # ``_serialize_support_ticket``), while the curated action in # :mod:`app.assistant.actions` returns ``ticket_id`` — and # ``get_action`` prefers curated, so ``ticket_id`` is what actually # arrives. The guard test resolves through ``get_action`` for exactly # this reason; a locator checked against the tool alone would have # missed on every support ticket, which is the highest-volume action # the assistant has ever proposed in production. id_key="ticket_id", number_key="ticket_number", resolver_key="support_ticket", deep_link="/me/support/{id}", ), # ── self-service requests (one unified ``employee_requests`` table) ── # # Five actions, one row shape. ``EmployeeRequest`` was renamed out of # ``asset_employee_requests`` precisely because it stopped being about # assets, so splitting these into five resource types would multiply # resolvers without changing a single answer an employee gets. "create_asset_request": OutcomeLocator( resource_type=RES_EMPLOYEE_REQUEST, id_key="request_id", number_key="request_number", resolver_key="employee_request", deep_link="/my/requests", ), "create_email_alias": OutcomeLocator( resource_type=RES_EMPLOYEE_REQUEST, id_key="request_id", number_key="request_number", resolver_key="employee_request", deep_link="/my/requests", ), "request_email_group": OutcomeLocator( resource_type=RES_EMPLOYEE_REQUEST, id_key="request_id", number_key="request_number", resolver_key="employee_request", deep_link="/my/requests", ), "request_join_email_group": OutcomeLocator( resource_type=RES_EMPLOYEE_REQUEST, id_key="request_id", resolver_key="employee_request", deep_link="/my/requests", ), "submit_computer_buyback": OutcomeLocator( resource_type=RES_EMPLOYEE_REQUEST, # The curated action delegates to ``submit_approval_request``, whose # ``ApprovalBackedEmployeeRequestResponse`` carries ``id`` — so the id # here comes from the *inner* tool, not from the buyback wrapper. id_key="id", number_key="request_number", resolver_key="employee_request", deep_link="/my/requests", ), # ── software ───────────────────────────────────────────────────── "request_software_seat": OutcomeLocator( resource_type=RES_SOFTWARE_SEAT, # ``allocation_id``, not the ``approval_request_id`` sitting beside it # in the same dict: the allocation is the thing the employee wanted, # and the approval is a step on its way. The resolver reaches the # approval *through* the allocation, so a seat granted without one # still resolves instead of looking permanently unlinked. id_key="allocation_id", resolver_key="software_seat", deep_link="/me", ), "request_permission": OutcomeLocator( resource_type=RES_USER_PERMISSION, # ``request_permission`` returns the USER-PERMISSION row itself # (``UserPermissionResponse``), not a request wrapper — the grant and # the ask for it are one row in this domain. The tool re-reads it after # creating it and raises rather than falling back to "some row", so an # ``id`` that arrives here is the one it just made. id_key="id", resolver_key="user_permission", deep_link="/me", ), "request_software_addon": OutcomeLocator( resource_type=RES_SOFTWARE_ADDON, id_key="id", resolver_key="software_addon", deep_link="/me", ), # ── api keys ───────────────────────────────────────────────────── "request_api_key": OutcomeLocator( resource_type=RES_API_KEY_REQUEST, id_key="id", number_key="request_number", resolver_key="api_key_request", deep_link="/me/api-keys", ), # No asset-transfer entry: ``create_asset_transfer`` is in # ``derived_actions.OPT_OUT`` — it hands an asset to a *named colleague*, # and picking the right person out of a chat sentence is the failure # mode. The assistant cannot offer it, so a locator for it would be # coverage on paper only. "release_my_api_key": OutcomeLocator( resource_type=RES_API_KEY_REQUEST, # Same ``ApiKeyAssignmentResponse`` as ``request_api_key`` — releasing a # key moves the SAME assignment row through ``pending_release`` to # ``revoked``. One resolver serves both directions, which is why the # release needed no new spec: what the employee is following is the # same object either way. id_key="id", number_key="request_number", resolver_key="api_key_request", deep_link="/me/api-keys", ), # ── network / data / cloud ─────────────────────────────────────── "submit_vpn_application": OutcomeLocator( resource_type=RES_VPN_APPLICATION, # Curated again, and again it disagrees with its MCP twin: the tool # returns ``id``, the curated action returns ``application_id``. id_key="application_id", resolver_key="vpn_application", deep_link="/my/requests", ), "submit_data_access_personal_grant": OutcomeLocator( resource_type=RES_DATA_ACCESS_REQUEST, id_key="request_id", resolver_key="data_access_request", deep_link="/my/data-access/requests", ), "request_cloud_grant_renewal": OutcomeLocator( resource_type=RES_CLOUD_GRANT_RENEWAL, id_key="request_group_id", # One id, many grants. Carried so the projection can never call a # batch finished because its first row is. count_key="grant_count", resolver_key="cloud_grant_renewal", deep_link="/my/cloud/grants", ), } #: What an identifier may look like. Deliberately narrow: every id this #: contract handles is a UUID or a short opaque key, and anything with a path #: separator in it is not an id — it is a path. #: Must START with an alphanumeric, so ``..`` and ``.`` are not identifiers — #: a dot-only value composes ``/support/tickets/..``, which resolves to the #: list page and quietly answers a question about the wrong thing. _ID_SHAPE = re.compile(r"[A-Za-z0-9][A-Za-z0-9._:-]{0,63}") def unwrap_result(result: Any) -> dict[str, Any] | None: """Return the action's own payload dict. Two shapes reach here and **both are ours**, so this is unwrapping, not sniffing: a curated action returns its payload directly, while a derived action returns ``{"tool": …, "result": …}`` because the wrapper adds the tool name. Anything else is a shape we do not produce, and gets no guess. """ if not isinstance(result, dict): return None if "tool" in result and "result" in result: inner = result.get("result") return inner if isinstance(inner, dict) else None return result def source_tool_of(result: Any) -> str | None: """The MCP tool that actually produced this payload, when one did. Usually the same as ``action_key`` — a derived action IS its tool. It differs exactly where a curated action delegates: ``submit_computer_buyback`` runs ``submit_approval_request``, and without this the record would name the wrapper while the id came from something else. That is the pair you want in front of you when a locator turns out to point at the wrong row. """ if isinstance(result, dict) and "tool" in result and "result" in result: tool = result.get("tool") return str(tool)[:64] if tool else None return None def locate(action_key: str, result: Any) -> tuple[LocatedObject | None, str | None]: """Find the business object one executed action created. Returns ``(object, None)`` or ``(None, miss_reason)``. Never raises: a locator miss must not turn a successful submission into a failed one — the domain row exists either way, and losing the link is strictly better than telling the employee their request failed when it did not. """ locator = LOCATORS.get(action_key) if locator is None: return None, MISS_NO_LOCATOR payload = unwrap_result(result) if payload is None: return None, MISS_NO_RESULT if locator.id_key not in payload: return None, MISS_ID_KEY_ABSENT raw = payload.get(locator.id_key) resource_id = "" if raw is None else str(raw).strip() if not resource_id: return None, MISS_ID_BLANK if not _ID_SHAPE.fullmatch(resource_id) or ".." in resource_id: # Refused rather than escaped. Escaping would keep a nonsense id and # point the employee at a page that cannot exist; refusing records that # the contract met something it does not understand, which is the fact # somebody needs. return None, MISS_ID_UNSAFE number_raw = payload.get(locator.number_key) if locator.number_key else None number = None if number_raw is None else str(number_raw).strip() or None covers = None if locator.count_key is not None: count_raw = payload.get(locator.count_key) if isinstance(count_raw, int): covers = count_raw deep_link = None if locator.deep_link: deep_link = locator.deep_link.format(id=resource_id, number=number or "") return ( LocatedObject( resource_type=locator.resource_type, resource_id=resource_id, resolver_key=locator.resolver_key, number=number, deep_link=deep_link, covers=covers, ), None, )