fix(spp_scoring,spp_scoring_programs): scoring validation, eligibility enforcement, and access control

spp_programs/__manifest__.py

@@ -4,7 +4,7 @@
     "name": "OpenSPP Programs",
     "summary": "Manage programs, cycles, beneficiary enrollment, entitlements (cash and in-kind), payments, and fund tracking for social protection.",
     "category": "OpenSPP/Core",
-    "version": "19.0.2.0.6",
+    "version": "19.0.2.0.7",
     "sequence": 1,
     "author": "OpenSPP.org",
     "website": "https://github.com/OpenSPP/OpenSPP2",

spp_programs/models/managers/program_manager.py

@@ -3,7 +3,7 @@
 from datetime import datetime, timedelta
 
 from odoo import _, api, fields, models
-from odoo.exceptions import UserError
+from odoo.exceptions import UserError, ValidationError
 
 from odoo.addons.job_worker.delay import group
 
@@ -219,12 +219,59 @@ def _enroll_eligible_registrants(self, states, offset=0, limit=None, do_count=Fa
         _logger.debug("members filtered: %s", members)
         not_enrolled = members.filtered(lambda m: m.state not in ("enrolled", "duplicated", "exited"))
         _logger.debug("not_enrolled: %s", not_enrolled)
-        not_enrolled.write(
+
+        # Run pre-enrollment hooks (e.g., scoring eligibility checks).
+        # Members that fail the hook are moved to not_eligible.
+        hook_failed = self.env["spp.program.membership"]
+        for member in not_enrolled:
+            try:
+                program._pre_enrollment_hook(member.partner_id)
+            except (ValidationError, UserError) as e:
+                _logger.info(
+                    "Pre-enrollment hook rejected registrant %s: %s",
+                    member.partner_id.id,
+                    str(e),
+                )
+                hook_failed |= member
+
+        # Re-check already-enrolled members against the current
+        # eligibility rules. "Verify Eligibility" implies a fresh check;
+        # an enrolled registrant whose data became invalid (e.g. a
+        # required indicator now resolves to a sentinel string) must be
+        # demoted, not silently kept enrolled. We work from `member_before`
+        # (pre-eligibility-manager-filter) so an enrolled member that the
+        # default manager would silently skip is still re-checked here.
+        # See OP#838.
+        already_enrolled = member_before.filtered(lambda m: m.state == "enrolled")
+        re_verify_failed = self.env["spp.program.membership"]
+        for member in already_enrolled:
+            try:
+                program._pre_enrollment_hook(member.partner_id)
+            except (ValidationError, UserError) as e:
+                _logger.info(
+                    "Re-verify rejected enrolled registrant %s: %s",
+                    member.partner_id.id,
+                    str(e),
+                )
+                re_verify_failed |= member
+
+        enrollable = not_enrolled - hook_failed
+        if hook_failed:
+            hook_failed.write({"state": "not_eligible"})
+
+        if re_verify_failed:
+            re_verify_failed.write({"state": "not_eligible"})
+
+        enrollable.write(
             {
                 "state": "enrolled",
                 "enrollment_date": fields.Datetime.now(),
             }
         )
+
+        # Run post-enrollment hooks (e.g., auto-score on enrollment)
+        for member in enrollable:
+            program._post_enrollment_hook(member.partner_id)
         # dis-enroll the one not eligible anymore:
         enrolled_members_ids = members.ids
         members_to_remove = member_before.filtered(
@@ -242,4 +289,4 @@ def _enroll_eligible_registrants(self, states, offset=0, limit=None, do_count=Fa
             program._compute_eligible_beneficiary_count()
             program._compute_beneficiary_count()
 
-        return len(not_enrolled)
+        return len(enrollable)

spp_scoring/__manifest__.py

@@ -1,7 +1,7 @@
 {
     "name": "OpenSPP Scoring",
     "category": "OpenSPP/Targeting",
-    "version": "19.0.2.0.0",
+    "version": "19.0.2.0.3",
     "sequence": 1,
     "author": "OpenSPP.org",
     "website": "https://github.com/OpenSPP/OpenSPP2",
@@ -33,7 +33,9 @@
         "views/scoring_model_views.xml",
         "views/scoring_indicator_views.xml",
         "views/scoring_threshold_views.xml",
+        "views/scoring_invalid_value_views.xml",
         "views/scoring_result_views.xml",
+        "views/res_partner_views.xml",
         # Wizards (must be before menus so actions are available)
         "wizard/batch_scoring_wizard_views.xml",
         # Menus (last, references actions from other files)

spp_scoring/models/__init__.py

@@ -2,9 +2,11 @@
 from . import scoring_indicator
 from . import scoring_value_mapping
 from . import scoring_threshold
+from . import scoring_invalid_value
 from . import scoring_result
 from . import scoring_result_detail
 from . import scoring_engine
 from . import scoring_batch_job
 from . import scoring_indicator_provider
 from . import scoring_data_integration
+from . import res_partner

spp_scoring/models/res_partner.py

@@ -0,0 +1,61 @@
+# Part of OpenSPP. See LICENSE file for full copyright and licensing details.
+"""Extends res.partner with scoring result count and actions."""
+
+from odoo import api, fields, models
+
+
+class ResPartner(models.Model):
+    """Add scoring smart button to registrant form."""
+
+    _inherit = "res.partner"
+
+    scoring_result_ids = fields.One2many(
+        "spp.scoring.result",
+        "registrant_id",
+        string="Scoring Results",
+    )
+    scoring_result_count = fields.Integer(
+        string="# Scores",
+        compute="_compute_scoring_result_count",
+    )
+
+    @api.depends("scoring_result_ids")
+    def _compute_scoring_result_count(self):
+        for partner in self:
+            partner.scoring_result_count = len(partner.scoring_result_ids)
+
+    def action_view_scoring_results(self):
+        """Open scoring results for this registrant."""
+        self.ensure_one()
+        return {
+            "name": self.name,
+            "type": "ir.actions.act_window",
+            "res_model": "spp.scoring.result",
+            "view_mode": "list,form",
+            "domain": [("registrant_id", "=", self.id)],
+            "context": {"default_registrant_id": self.id},
+        }
+
+    def action_score_registrant(self):
+        """Open the batch scoring wizard pre-filled for the selected registrant(s).
+
+        Works for both single-record (form view) and multi-record
+        (list-view bulk selection) calls. The wizard's `registrant_ids`
+        m2m takes precedence over `registrant_domain` (see
+        wizard/batch_scoring_wizard.py:70), so we only need to seed the
+        m2m. The redundant `default_domain` keeps the wizard form's
+        "Domain" field readable when a user opens it on one record.
+        """
+        if not self:
+            return False
+        ctx = {"default_registrant_ids": self.ids}
+        if len(self) == 1:
+            ctx["default_registrant_domain"] = f"[('id', '=', {self.id})]"
+        return {
+            "name": "Score Registrant",
+            "type": "ir.actions.act_window",
+            "res_model": "spp.batch.scoring.wizard",
+            "view_mode": "form",
+            "target": "new",
+            "context": ctx,
+        }

spp_scoring/models/scoring_data_integration.py

@@ -151,8 +151,13 @@ def calculate_score(self, registrant, scoring_model, mode="manual"):
         # Call original method
         result = super().calculate_score(registrant, scoring_model, mode)
 
-        # Store in unified cache if enabled
-        if scoring_model.cache_scores:
+        # Only cache *complete* results — strict-mode required-indicator
+        # failures persist an audit row with is_complete=False and a
+        # zeroed score; publishing that to spp.data.value would let CEL
+        # consumers (e.g. `r.demo_hva_2025_score >= 0.6`) read 0 and
+        # treat the registrant as scored-low rather than not-scored.
+        # See OP#838.
+        if scoring_model.cache_scores and result.is_complete:
             self._store_score_in_cache(result, registrant, scoring_model)
 
         return result

spp_scoring/models/scoring_engine.py

@@ -71,6 +71,12 @@ def calculate_score(self, registrant, scoring_model, mode="manual"):
         # Get sorted indicators (prefetched)
         indicators = scoring_model.indicator_ids.sorted(key=lambda i: i.sequence)
 
+        # Track required-indicator failures separately. In strict mode
+        # these mark the whole result as incomplete (audit row only — no
+        # score is published downstream); other indicator errors are
+        # recorded in error_messages but don't block the score.
+        required_failures = []
+
         # Process each indicator
         for indicator in indicators:
             result = self._calculate_indicator(indicator, registrant)
@@ -79,6 +85,7 @@ def calculate_score(self, registrant, scoring_model, mode="manual"):
             if result.get("error"):
                 errors.append(result["error"])
                 if is_strict_mode and indicator.is_required:
+                    required_failures.append(indicator)
                     continue
 
             total_score += result.get("weighted_score", 0.0)
@@ -115,13 +122,20 @@ def calculate_score(self, registrant, scoring_model, mode="manual"):
         if calculation_method == "cel_formula":
             total_score = self._calculate_cel_total(scoring_model, registrant, inputs_snapshot)
 
-        # Determine classification (thresholds already prefetched)
-        classification = self._get_classification(total_score, scoring_model)
-
-        # Check for errors in strict mode
-        is_complete = True
-        if errors and is_strict_mode:
-            is_complete = False
+        # Strict mode + at least one required-indicator failure → the
+        # result is persisted as an audit row but flagged incomplete.
+        # Score is zeroed and classification cleared so a downstream
+        # consumer that forgets to gate on `is_complete` still gets
+        # nothing meaningful. The cache layer in
+        # scoring_data_integration._store_score_in_cache and the
+        # cached-lookup path in scoring_result.get_latest_score both
+        # filter on is_complete=True. See OP#838.
+        is_complete = not (is_strict_mode and required_failures)
+        if not is_complete:
+            total_score = 0.0
+            classification = {"code": False, "label": False, "color": False}
+        else:
+            classification = self._get_classification(total_score, scoring_model)
 
         # Create result record
         Result = self.env["spp.scoring.result"]
@@ -164,8 +178,9 @@ def _prefetch_model_data(self, scoring_model):
         This ensures indicators, value mappings, and thresholds are loaded
         in a minimal number of queries before processing.
         """
-        # Prefetch indicators with their value mappings
+        # Prefetch indicators with their value mappings + invalid values
         scoring_model.indicator_ids.mapped("value_mapping_ids")
+        scoring_model.indicator_ids.mapped("invalid_value_ids")
 
         # Prefetch thresholds
         list(scoring_model.threshold_ids)
@@ -206,10 +221,25 @@ def _calculate_indicator(self, indicator, registrant):
 
             result["field_value"] = field_value
 
-            # Handle missing values
-            if field_value is None:
+            # Handle missing or falsy values.
+            # For required indicators we treat None, boolean False, and
+            # empty containers/strings as "missing" — but numeric 0 / 0.0
+            # stays a valid value (a real score). The previous
+            # `field_value != 0` short-circuited on boolean False (since
+            # `False == 0` in Python), letting empty Date/Char fields
+            # bypass strict mode entirely. See OP#838.
+            #
+            # Each indicator can pick a subset of the curated Invalid
+            # Values library (spp.scoring.invalid.value) — only active
+            # rows count. Matching strings are treated as missing for
+            # both required and non-required indicators.
+            invalid_values = set(indicator.invalid_value_ids.filtered("active").mapped("name"))
+            if self._is_missing_value(field_value, indicator.is_required, invalid_values):
                 if indicator.is_required:
-                    result["error"] = _("Required field '%s' is missing.") % indicator.field_path
+                    result["error"] = _("Required field '%(path)s' has no valid value (got: %(value)s).") % {
+                        "path": indicator.field_path,
+                        "value": repr(field_value),
+                    }
                     return result
                 field_value = indicator._convert_default_value()
 
@@ -244,6 +274,37 @@ def _calculate_indicator(self, indicator, registrant):
 
         return result
 
+    @staticmethod
+    def _is_missing_value(field_value, is_required, invalid_patterns=None):
+        """Decide whether a field value should be treated as 'missing'.
+
+        Non-required indicators only treat ``None`` as missing — anything
+        else (including 0, "", False) flows through to the indicator's
+        own calculation, which decides what to do with it.
+
+        Required indicators additionally treat boolean ``False`` and
+        empty strings/containers as missing. Numeric 0 (int or float),
+        and the literal ``True``, stay valid.
+
+        ``invalid_patterns`` is the global set of curated invalid-value
+        strings (from ``spp.scoring.invalid.value``). Matching strings
+        are treated as missing for both required and non-required
+        indicators.
+        """
+        if field_value is None:
+            return True
+        # Configured invalid-value strings always count as missing.
+        if invalid_patterns and isinstance(field_value, str) and field_value.strip() in invalid_patterns:
+            return True
+        if not is_required:
+            return False
+        # Booleans subclass int in Python, so check this branch first.
+        if isinstance(field_value, bool):
+            return not field_value
+        if isinstance(field_value, (int, float)):
+            return False
+        return not field_value
+
     def _evaluate_indicator_formula(self, indicator, registrant):
         """Evaluate a CEL formula for an indicator."""
         if not indicator.cel_expression:
@@ -401,6 +462,21 @@ def _batch_score_sync(self, registrants, scoring_model, options):
                     scoring_model,
                     mode="batch",
                 )
+                # Strict-mode required-indicator failures persist an
+                # incomplete audit row instead of raising. Count those as
+                # batch failures so the wizard summary shows the real
+                # picture and fail_fast still works.
+                if not result.is_complete:
+                    errors.append(
+                        {
+                            "registrant_id": registrant.id,
+                            "registrant_name": registrant.name,
+                            "error": result.error_messages or _("Scoring incomplete"),
+                        }
+                    )
+                    if options.get("fail_fast"):
+                        raise UserError(result.error_messages or _("Scoring incomplete"))
+                    continue
                 results.append(result)
 
             except Exception as e:
@@ -590,7 +666,9 @@ def get_or_calculate_score(self, registrant, scoring_model, max_age_days=None):
         """
         Result = self.env["spp.scoring.result"]
 
-        if max_age_days:
+        # max_age_days > 0: reuse cached score if fresh enough
+        # max_age_days = 0 or None: always recalculate
+        if max_age_days and max_age_days > 0:
             existing = Result.get_latest_score(registrant, scoring_model, max_age_days)
             if existing:
                 return existing

spp_scoring/models/scoring_indicator.py

@@ -93,6 +93,16 @@ class SppScoringIndicator(models.Model):
         default=0.0,
         help="Score to use when value is missing or not mapped",
     )
+    invalid_value_ids = fields.Many2many(
+        comodel_name="spp.scoring.invalid.value",
+        string="Invalid Values",
+        help=(
+            "Pick from the curated library of strings that should be "
+            "treated as missing for this indicator (e.g. 'No Birthdate!'). "
+            "Manage the library under Scoring → Configuration → "
+            "Invalid Values."
+        ),
+    )
 
     # Value mappings
     value_mapping_ids = fields.One2many(