feat(search): offer each choice refusal's options as ready requests, and skip qualifiers the answer applies

packages/dfo-data-tools/src/entity_resolution/resolver.py

@@ -365,24 +365,44 @@ def _unless_expired[*Ts, R](deadline: float | None, fn: Callable[[*Ts], R], /, *
     return fn(*args)
 
 
-def _group_node_dcids_job(shared: SharedResolver) -> frozenset[str]:
+@dataclass(frozen=True, slots=True)
+class _GroupNodes:
+    """The dcids `Group.entity` can hold (`dcids`), and each of them with the first
+    registered group name that resolves to its group alone (`names`)."""
+
+    dcids: frozenset[str]
+    names: Mapping[str, str]
+
+
+def _group_nodes_job(shared: SharedResolver) -> _GroupNodes:
     """The dcids `Group.entity` can hold, for every group resolvekit knows: the group's own
     id as a dcid, and each organisation that survives the gate beside it
     (`EntityResolver._assemble_group`). A region the membership policy holds is left out,
-    since the policy answers its members ("Africa (DAC)" resolves to `africa`). Runs on the
-    shared resolver's single worker."""
+    since the policy answers its members ("Africa (DAC)" resolves to `africa`).
+
+    Each dcid takes the first registered name that resolves to its group alone: a name whose
+    top survivor is a group the registry holds (`_group_hit`), which `resolve` answers as
+    that `Group` whatever else survives. A name whose top survivor the registry does not
+    hold is an ordinary hit, which can be ambiguous, so its dcids take a later name or none.
+    Runs on the shared resolver's single worker."""
     context = ResolutionContext(entity_types=ADMITTED_TYPES)
     dcids: set[str] = set()
+    names: dict[str, str] = {}
     for name in shared.resolver.known_groups():
         listed, _, _ = _resolve_text_survivors(shared, name, context)
         survivors = [candidate for candidate, _ in listed]
         if not survivors or survivors[0].entity_type not in _GROUP_TYPES:
             continue
+        alone = _group_hit(shared, survivors[0]) is not None
         nodes = [survivors[0]] + [c for c in survivors[1:] if c.entity_type == _ORG_ENTITY_TYPE]
-        dcids.update(
-            dcid for node in nodes if (dcid := identity_to_dcid(node.entity_id)) is not None
-        )
-    return frozenset(dcid for dcid in dcids if not shared.policy.holds_region(dcid))
+        for node in nodes:
+            dcid = identity_to_dcid(node.entity_id)
+            if dcid is None or shared.policy.holds_region(dcid):
+                continue
+            dcids.add(dcid)
+            if alone:
+                names.setdefault(dcid, name)
+    return _GroupNodes(dcids=frozenset(dcids), names=names)
 
 
 def _entity_from_confirmation(
@@ -467,7 +487,7 @@ def __init__(
         self._cache: TLRUCache[_CacheKey, Resolution] = TLRUCache(maxsize=cache_size, ttu=_ttu)
         self._hits = 0
         self._misses = 0
-        self._group_node_dcids: frozenset[str] | None = None
+        self._group_nodes: _GroupNodes | None = None
 
     @property
     def membership(self) -> MembershipPolicy:
@@ -501,11 +521,21 @@ async def group_node_dcids(self) -> frozenset[str]:
         """The dcids that are the node of a group the resolver knows, such as
         `EuropeanUnion`: every dcid a `Group.entity` can hold. The packs fix the set, so it is
         computed once and no graph is read."""
-        if self._group_node_dcids is None:
-            self._group_node_dcids = await asyncio.get_running_loop().run_in_executor(
-                self._shared.executor, _group_node_dcids_job, self._shared
+        return (await self._group_node_lookup()).dcids
+
+    async def group_node_names(self) -> Mapping[str, str]:
+        """Each dcid of `group_node_dcids` with a registered group name that resolves to its
+        group alone, such as `EuropeanUnion` to "European Union": the text that asks for the
+        group's members, which a pin on the node cannot. A dcid whose group has no such name
+        is left out. Computed once from the packs, with no graph read."""
+        return (await self._group_node_lookup()).names
+
+    async def _group_node_lookup(self) -> _GroupNodes:
+        if self._group_nodes is None:
+            self._group_nodes = await asyncio.get_running_loop().run_in_executor(
+                self._shared.executor, _group_nodes_job, self._shared
             )
-        return self._group_node_dcids
+        return self._group_nodes
 
     async def resolve(
         self,

src/search/engine/assemble.py

@@ -38,11 +38,11 @@
     NoData,
     NoDataDenominator,
     NoDataGate,
+    NoDataOption,
     NoDataPart,
     NoDataReason,
     Operation,
     Ordering,
-    ParsedInput,
     ParsedPart,
     ResolvedQualifier,
     RoleAxis,
@@ -212,10 +212,23 @@ def _surviving_qualifier_warnings(
     """Keep dropped-qualifier warnings that remain true across all regions.
 
     Each warning stays paired with its extracted phrase. One region admitting a phrase
-    makes the request-wide warning false. Compare with the same strip-and-casefold
-    normalisation used by ``admitted_qualifier_phrases``.
+    makes the request-wide warning false. A part's departures read its own region's
+    admission (`unapplied_qualifier_warnings`, `RegionResult.unapplied_qualifier_warnings`).
     """
-    admitted = {phrase for region in regions for phrase in region.admitted_qualifier_phrases}
+    admitted = frozenset(
+        phrase for region in regions for phrase in region.admitted_qualifier_phrases
+    )
+    return unapplied_qualifier_warnings(qualifier_warnings, phrases, admitted=admitted)
+
+
+def unapplied_qualifier_warnings(
+    qualifier_warnings: Sequence[Warning],
+    phrases: Sequence[str],
+    *,
+    admitted: frozenset[str],
+) -> list[Warning]:
+    """The `qualifier_warnings` whose paired phrase `admitted` does not hold, compared with
+    the strip-and-casefold normalisation ``admitted_qualifier_phrases`` uses."""
     return [
         warning
         for warning, phrase in zip(qualifier_warnings, phrases, strict=True)
@@ -465,12 +478,13 @@ def build_no_data_part(
     alternate_match: Spec | None = None,
     elected_stat_vars: list[GraphRef] | None = None,
     entity_terms: list[str] | None = None,
+    entity_groups: Sequence[Sequence[str]] | None = None,
     role_axis: EntityRoleKey | None = None,
     bound_role_dcids: Mapping[EntityRoleKey, Sequence[str]] | None = None,
     ambiguous_candidates: list[str] | None = None,
     ambiguous_entities: list[AmbiguousEntityGroup] | None = None,
     ambiguous_roles: Mapping[str, RoleAxis] | None = None,
-    data_query: ParsedInput | None = None,
+    options: list[NoDataOption] | None = None,
     source_terms: list[str] | None = None,
     ambiguous_sources: list[AmbiguousSourceGroup] | None = None,
     qualifier_terms: list[ResolvedQualifier] | None = None,
@@ -496,13 +510,14 @@ def build_no_data_part(
     code lives on ``NoData.gate``. ``elected_stat_vars`` lets sentence composition name an
     exact series match before a nearest substitute.
 
-    ``entity_terms`` preserves the place wording seen by the engine. ``role_axis`` names an
+    ``entity_terms`` preserves the place wording seen by the engine, and ``entity_groups``
+    groups a ``role_not_resolved`` refusal's terms by the place each names. ``role_axis`` names an
     open origin or destination axis in ``no_observations`` copy. ``bound_role_dcids`` records
     query-bound role values for alternate validation, including when both roles are bound.
     ``ambiguous_candidates`` feeds the wording. ``ambiguous_entities`` carries structured
     graph candidates, and ``ambiguous_roles`` names the flow axis of each ambiguous wording
-    the sentence marks as a slot. ``data_query`` is the part's parsed form for
-    ``entity_ambiguous``, ``source_ambiguous`` and ``source_code_ambiguous``.
+    the sentence marks as a slot. ``options`` are the ready requests of a refusal that
+    asks for a choice (`conjoin._options`).
     ``source_terms`` are the source wordings a ``source_not_resolved`` or
     ``source_ambiguous`` outcome rests on, or the series code a ``source_code_*`` outcome
     rests on, and ``ambiguous_sources`` pairs each ambiguous one with its candidates. For
@@ -590,6 +605,7 @@ def build_no_data_part(
         elected_stat_vars=elected_stat_vars,
         variable_term=variable_text,
         entity_terms=entity_terms,
+        entity_groups=entity_groups,
         role_axis=role_axis,
         bound_role_dcids=bound_role_dcids,
         ambiguous_candidates=ambiguous_candidates,
@@ -631,6 +647,7 @@ def build_no_data_part(
                 reason,
                 variable_term=variable_text,
                 entity_terms=entity_terms,
+                entity_groups=entity_groups,
                 role_axis=role_axis,
                 ambiguous_candidates=ambiguous_candidates,
                 elected_stat_vars=elected_stat_vars,
@@ -653,7 +670,7 @@ def build_no_data_part(
             unbound_qualifiers=[*(qualifier_terms or ()), *(unbound_sources or ())],
             unit_alternatives=unit_alternatives,
             denominator=denominator,
-            data_query=data_query,
+            options=options or [],
         ),
         rendered_sentence=rendered,
         sentence_plan=plan,