ON-5590: Sub-sector impact mapping and clearer prioritization explanations

hiap-meed/.env.example

@@ -26,6 +26,9 @@ HIAP_MEED_POLICY_SIGNALS_DATA_SOURCE=mock
 # default number of ranked actions returned when request cityDataList item has no topN override
 HIAP_MEED_TOP_N=20
 
+# guarded future activity-data-level mapping step inside Impact
+ACTIVITY_DATA_LEVEL_MAPPING=false
+
 # alignment other-preference LLM mapping runtime config
 # model used by alignment free-text -> co-benefit mapping
 HIAP_MEED_ALIGNMENT_OTHER_PREFERENCE_MODEL=

hiap-meed/README.md

@@ -41,6 +41,7 @@ HIAP_MEED_LEGAL_DATA_SOURCE=mock
 HIAP_MEED_ACTION_DATA_SOURCE=mock
 HIAP_MEED_POLICY_SIGNALS_DATA_SOURCE=mock
 HIAP_MEED_TOP_N=20
+ACTIVITY_DATA_LEVEL_MAPPING=false
 HIAP_MEED_ALIGNMENT_OTHER_PREFERENCE_MODEL=
 HIAP_MEED_FREE_TEXT_EXCLUSIONS_ENABLED=false
 HIAP_MEED_FREE_TEXT_EXCLUSIONS_MODEL=
@@ -64,6 +65,7 @@ Variables:
 - `HIAP_MEED_ACTION_DATA_SOURCE`: action catalog source (`mock` or `api`)
 - `HIAP_MEED_POLICY_SIGNALS_DATA_SOURCE`: policy-signal input source (`mock` or `api`)
 - `HIAP_MEED_TOP_N`: default number of ranked actions to return per city (default `20`)
+- `ACTIVITY_DATA_LEVEL_MAPPING`: guarded future Impact mapping switch; `false` keeps true subsector-only matching, `true` calls the current stub and still returns subsector-only results
 - `HIAP_MEED_ALIGNMENT_OTHER_PREFERENCE_MODEL`: OpenAI model used only by the deprecated legacy free-text co-benefit mapping helper
 - `HIAP_MEED_FREE_TEXT_EXCLUSIONS_ENABLED`: if `true`, the exclusion preview endpoint calls OpenAI to resolve clear free-text action exclusions
 - `HIAP_MEED_FREE_TEXT_EXCLUSIONS_MODEL`: OpenAI model used for preview free-text action exclusion matching
@@ -181,11 +183,22 @@ Score normalization policy:
 Impact block behavior (implemented):
 
 - Impact reads action emissions targeting from `emissions`, including:
+  - `sector_number`
+  - `subsector_number` (**list** of subsector integers, even when only one subsector is covered)
   - `gpc_reference_number` (**list** of GPC refs in the mock/API schema)
   - `impact_text` (`very low`, `low`, `medium`, `high`, `very high`)
 - Impact reads city emissions from the frontend request:
+  - `requestData.cityDataList[].cityEmissionsData.gpcData[*].activities[*].activityType`
   - `requestData.cityDataList[].cityEmissionsData.gpcData[*].activities[*].totalEmissions`
-  - The service sums activity emissions per GPC key before scoring.
+  - The service normalizes each outer GPC key to `sector.subsector` and sums activity emissions at that true subsector level before scoring.
+- `gpc_reference_number` remains stored as passive reference data, but the active Impact join now uses `sector_number + subsector_number[]`.
+- `coBenefits[*]` now only carry co-benefit impact metadata (`impact_numeric`, optional relationship/text/methodology). They do not carry sector or GPC targeting fields.
+- `ACTIVITY_DATA_LEVEL_MAPPING=false` keeps the new true subsector matching path.
+- `ACTIVITY_DATA_LEVEL_MAPPING=true` calls the current activity-data stub, logs `not implemented`, and still returns the same subsector-level result.
+- Negative `V.*` AFOLU inventory values remain valid request data, but Impact does not treat them as reducible emissions.
+  - Matching for Impact uses strictly positive city emissions only.
+  - The reduction-share denominator also uses strictly positive city emissions only.
+  - This is an intentional product rule so Impact stays in `0..1` and measures reducible emissions rather than existing removals.
 - Impact computes canonical score as:
   - `0.80 * reduction_share_of_city_emissions + 0.20 * timeline_score`
   - Timeline mapping: `<5 years -> 1.0`, `5-10 years -> 0.5`, `>10 years -> 0.0`, missing or unknown timeline `-> 0.5`
@@ -383,7 +396,7 @@ Example response:
           "evidence_summary": {
             "impact": {
               "impact_block_score": 0.88,
-              "matched_city_gpc_refs_count": 2
+              "matched_city_subsector_keys_count": 1
             }
           },
           "explanations": {}
@@ -455,6 +468,7 @@ What each request run folder contains:
 
 - `summary.jsonl`: one JSON line per high-level pipeline event for that request
 - `NNN_<step>.json`: concise per-step detail files (fetch, filter, score, response summary)
+- Impact step details include true subsector matching diagnostics plus the `ACTIVITY_DATA_LEVEL_MAPPING` flag/stub metadata.
 - `response_full.json`: full API response payload for that request
 - `input_snapshot.json`: reproducibility-critical request inputs
 - `manifest.json`: run-level index of generated files, key counts, and pointers for top-ranked rows vs full evidence files

hiap-meed/app/modules/prioritizer/api.py

@@ -8,6 +8,7 @@
 from fastapi import APIRouter, Depends, HTTPException
 
 from app.modules.prioritizer.config import resolve_top_n
+from app.modules.prioritizer.internal_models import CityActivityRow, CityEmissionsContext
 from app.modules.prioritizer.models import (
     ExplanationTranslationApiRequest,
     ExplanationTranslationApiResponse,
@@ -22,6 +23,9 @@
     PrioritizerApiResponse,
 )
 from app.modules.prioritizer.orchestrator import run_prioritization
+from app.modules.prioritizer.utils.subsector_mapping import (
+    normalize_gpc_reference_to_subsector_key,
+)
 from app.modules.prioritizer.services.exclusion_resolution import (
     resolve_exclusion_preview_with_diagnostics,
 )
@@ -53,26 +57,45 @@ def _error_payload(request_id: str, message: str) -> dict[str, str]:
     return {"request_id": request_id, "error": message}
 
 
-def _extract_city_emissions_by_gpc_ref(city_input: FrontendCityInput) -> dict[str, float]:
-    """
-    Build city emissions totals keyed by GPC reference.
+def _extract_city_emissions_context(city_input: FrontendCityInput) -> CityEmissionsContext:
+    """Build normalized subsector totals plus preserved activity rows."""
+    emissions_by_subsector_key: dict[str, float] = {}
+    activity_rows: list[CityActivityRow] = []
 
-    The frontend request carries emissions per GPC key under
-    `cityEmissionsData.gpcData[*].activities[*].totalEmissions`. This helper
-    sums activity totals per key and ignores missing totals (`None`).
-    """
-    emissions_by_gpc_ref: dict[str, float] = {}
-
-    # Sum activity emissions for each GPC key from frontend request schema.
+    # Normalize each GPC bucket to the active `sector.subsector` join key.
     for gpc_ref, gpc_entry in city_input.cityEmissionsData.gpcData.items():
+        sector_subsector_key = normalize_gpc_reference_to_subsector_key(gpc_ref)
         gpc_total = 0.0
         for activity in gpc_entry.activities:
-            if activity.totalEmissions is None:
-                continue
-            gpc_total += activity.totalEmissions
-        emissions_by_gpc_ref[gpc_ref] = gpc_total
+            if activity.activityType is None:
+                logger.warning(
+                    "City activity row missing activityType locode=%s gpc_reference_number=%s",
+                    city_input.locode,
+                    gpc_ref,
+                )
+            if activity.totalEmissions is not None:
+                gpc_total += activity.totalEmissions
+            activity_rows.append(
+                CityActivityRow(
+                    gpc_reference_number=gpc_ref,
+                    sector_subsector_key=sector_subsector_key,
+                    activity_type=activity.activityType,
+                    activity_value=activity.activityValue,
+                    activity_unit=activity.activityUnit,
+                    total_emissions=activity.totalEmissions,
+                    total_emissions_unit=activity.totalEmissionsUnit,
+                    data_source=activity.dataSource,
+                    notation_key=activity.notationKey,
+                )
+            )
+        emissions_by_subsector_key[sector_subsector_key] = (
+            emissions_by_subsector_key.get(sector_subsector_key, 0.0) + gpc_total
+        )
 
-    return emissions_by_gpc_ref
+    return CityEmissionsContext(
+        emissions_by_subsector_key=emissions_by_subsector_key,
+        activity_rows=activity_rows,
+    )
 
 
 def _normalize_requested_languages(requested_languages: list[str]) -> list[str]:
@@ -246,7 +269,7 @@ def preview_exclusions(
     response_model=PrioritizerApiResponse,
     summary="Run action prioritization synchronously",
     description=(
-        "Ranks actions for one or more cities from the frontend request envelope. "
+        "Ranks actions for one or more cities from the caller request envelope. "
         "When `createExplanations=true`, the backend first generates canonical "
         "English explanations and then translates them into any additionally "
         "requested languages."
@@ -277,7 +300,7 @@ def prioritize(
     ),
 ) -> PrioritizerApiResponse:
     """
-    Prioritize actions from the CityCatalyst frontend request envelope.
+    Prioritize actions from the caller request envelope.
 
     This endpoint is synchronous because the orchestrator and data clients
     are synchronous; FastAPI runs sync routes in a threadpool to avoid
@@ -538,7 +561,7 @@ def _run_for_city_input(
 
     # Create internal request ID used for orchestrator artifacts/tracing.
     internal_request_id = uuid4()
-    city_emissions_by_gpc_ref = _extract_city_emissions_by_gpc_ref(city_input)
+    city_emissions_context = _extract_city_emissions_context(city_input)
     result = run_prioritization(
         locode=city_input.locode,
         weights_override=city_input.weightsOverride,
@@ -549,7 +572,7 @@ def _run_for_city_input(
         city_preference_co_benefit_keys=list(
             city_input.cityStrategicPreferenceCoBenefitKeys
         ),
-        city_emissions_by_gpc_ref=city_emissions_by_gpc_ref,
+        city_emissions_context=city_emissions_context,
         internal_request_id=internal_request_id,
         city_data_api_client=city_data_api_client,
         action_data_api_client=action_data_api_client,