feat(quality_dashboard): banner with overdue + critical (Task 4)

_fetch_banner_candidates collects (overdue) OR (critical-customer +
open) records per type. _critical_customer_ids reuses partner.x_fc_rush
and partner.x_fc_vip flags when defined (gracefully no-ops when
absent). _critical_badge returns RUSH/VIP/AEROSPACE/AS9100 label
when the banner reason is critical-customer (no badge when overdue).
_build_banner ranks: overdue first by oldest, then critical-customer
by oldest, takes top 6, reports total_matching.

build() now collects banner candidates from every section in one
pass + invokes _build_banner once.

Tests cover overdue hold pickup, 6-cap with overflow count, and
all_clear when DB is empty.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

fusion_plating/fusion_plating_quality/controllers/fp_quality_dashboard.py

@@ -90,13 +90,23 @@ class FpQualityDashboardSnapshot:
 
     def build(self):
         sections = []
+        all_banner_candidates = []
         for type_code in SECTION_ORDER:
+            cfg = TYPE_CONFIG[type_code]
+            if cfg['model'] not in self.env:
+                continue
+            Model = self.env[cfg['model']].sudo()
             section = self._build_section(type_code)
             if section is None:
-                continue  # model not installed — omit
+                continue
             sections.append(section)
-        # Banner placeholder for Task 2 — fully implemented in Task 4.
+            # Collect banner candidates with type_code attached
-        banner = {'items': [], 'all_clear': True, 'total_matching': 0}
+            for rec, urgency, badge in self._fetch_banner_candidates(
+                    type_code, Model):
+                all_banner_candidates.append(
+                    (rec, urgency, badge, type_code),
+                )
+        banner = self._build_banner(all_banner_candidates)
         return {
             'banner': banner,
             'sections': sections,
@@ -217,6 +227,132 @@ class FpQualityDashboardSnapshot:
             return f'{age_str} overdue' if age_str else 'overdue'
         return age_str
 
+    # ===== Banner =====================================================
+
+    def _fetch_banner_candidates(self, type_code, Model):
+        """Per-type pull of records that qualify for the banner:
+        (overdue) OR (critical-customer AND state-is-open). Returns
+        list of (rec, urgency, critical_badge) tuples — deduped.
+        """
+        overdue_ids = set(self._overdue_ids(type_code, Model))
+        critical_ids = set(self._critical_customer_ids(type_code, Model))
+        # Open state filter (anything in the type's state_domain)
+        state_dom = list(OVERDUE_THRESHOLDS[type_code]['state_domain'])
+        open_recs = Model.search(state_dom)
+        results = []
+        for rec in open_recs:
+            is_overdue = rec.id in overdue_ids
+            is_critical = rec.id in critical_ids
+            if not (is_overdue or is_critical):
+                continue
+            urgency = 'overdue' if is_overdue else 'critical_customer'
+            badge = None if is_overdue else self._critical_badge(rec)
+            results.append((rec, urgency, badge))
+        return results
+
+    def _critical_customer_ids(self, type_code, Model):
+        """IDs of records belonging to critical customers (rush / vip /
+        aerospace). Defensive on missing fields per Rule 24."""
+        partner_field_map = {
+            'cert':  'partner_id',
+            'hold':  'partner_id',
+            'ncr':   'partner_id',
+            'rma':   'partner_id',
+            'capa':  'partner_id',  # may be False if linked via ncr_id
+            'check': 'job_id.partner_id',
+        }
+        # Find partners with critical flags
+        Partner = self.env['res.partner'].sudo()
+        critical_partner_clauses = []
+        if 'x_fc_rush' in Partner._fields:
+            critical_partner_clauses.append(('x_fc_rush', '=', True))
+        if 'x_fc_vip' in Partner._fields:
+            critical_partner_clauses.append(('x_fc_vip', '=', True))
+        if not critical_partner_clauses:
+            return []  # no flags defined in this codebase
+        # OR the conditions
+        critical_partner_domain = list(critical_partner_clauses)
+        if len(critical_partner_clauses) > 1:
+            critical_partner_domain = (
+                ['|'] * (len(critical_partner_clauses) - 1)
+                + critical_partner_clauses
+            )
+        critical_partner_ids = Partner.search(critical_partner_domain).ids
+        if not critical_partner_ids:
+            return []
+        partner_path = partner_field_map.get(type_code, 'partner_id')
+        # Compose the record-side filter — direct or via dotted path
+        dom = list(OVERDUE_THRESHOLDS[type_code]['state_domain'])
+        dom.append((partner_path, 'in', critical_partner_ids))
+        try:
+            return Model.search(dom).ids
+        except Exception as e:
+            # Field doesn't exist on the model (e.g. fusion.plating.capa
+            # may not have partner_id directly). Best-effort: skip.
+            _logger.debug(
+                "critical_customer probe failed for %s: %s", type_code, e,
+            )
+            return []
+
+    def _critical_badge(self, rec):
+        """Return 'RUSH' / 'VIP' / 'AEROSPACE' / 'AS9100' / None based
+        on which signal qualified the record. Defensive on missing fields."""
+        partner = self._resolve_partner(rec)
+        if partner:
+            if 'x_fc_rush' in partner._fields and getattr(partner, 'x_fc_rush', False):
+                return 'RUSH'
+            if 'x_fc_vip' in partner._fields and getattr(partner, 'x_fc_vip', False):
+                return 'VIP'
+        # Aerospace — check part name OR spec code if reachable
+        for path in ('part_catalog_id.name', 'customer_spec_id.code'):
+            head, _, attr = path.partition('.')
+            if head in rec._fields and rec[head] \
+                    and attr in rec[head]._fields:
+                val = (rec[head][attr] or '').upper()
+                if 'AEROSPACE' in val:
+                    return 'AEROSPACE'
+                if val.startswith('AS9100') or val.startswith('NADCAP'):
+                    return 'AS9100'
+        return None
+
+    def _build_banner(self, all_candidates):
+        """Rank: overdue first by oldest create_date, then critical-
+        customer by oldest create_date. Take top 6.
+        all_candidates: list of (rec, urgency, badge, type_code) tuples.
+        """
+        def rank_key(tup):
+            rec, urgency, _badge, _t = tup
+            # overdue=0 sorts before critical_customer=1; then oldest first.
+            return (0 if urgency == 'overdue' else 1,
+                    rec.create_date or self.now)
+        ranked = sorted(all_candidates, key=rank_key)
+        top = ranked[:self.BANNER_MAX]
+        return {
+            'items': [self._build_banner_item(t) for t in top],
+            'all_clear': len(all_candidates) == 0,
+            'total_matching': len(all_candidates),
+        }
+
+    def _build_banner_item(self, tup):
+        """Shape one banner item from (rec, urgency, badge, type_code)."""
+        rec, urgency, badge, type_code = tup
+        cfg = TYPE_CONFIG[type_code]
+        partner = self._resolve_partner(rec)
+        is_overdue = (urgency == 'overdue')
+        return {
+            'type': type_code,
+            'id': rec.id,
+            'name': rec.display_name or rec.name or f'#{rec.id}',
+            'customer': partner.name if partner else '',
+            'subtitle': self._build_subtitle(type_code, rec, is_overdue),
+            'urgency': urgency,
+            'critical_badge': badge,
+            'open_action': {
+                'res_model': cfg['model'],
+                'res_id': rec.id,
+            },
+        }
+
 
 class FpQualityDashboardController(http.Controller):
 

fusion_plating/fusion_plating_quality/tests/test_dashboard_snapshot.py

@@ -109,3 +109,64 @@ class TestDashboardSnapshotItems(TransactionCase):
         self.assertEqual(item['open_action']['res_model'],
                          'fusion.plating.quality.hold')
         self.assertEqual(item['open_action']['res_id'], hold.id)
+
+
+class TestDashboardSnapshotBanner(TransactionCase):
+    """Banner population + ranking."""
+
+    def _build(self):
+        from odoo.addons.fusion_plating_quality.controllers.fp_quality_dashboard \
+            import FpQualityDashboardSnapshot
+        return FpQualityDashboardSnapshot(self.env).build()
+
+    def test_banner_picks_up_overdue_hold(self):
+        from datetime import datetime, timedelta
+        if 'fusion.plating.quality.hold' not in self.env:
+            self.skipTest('fusion.plating.quality.hold not installed')
+        partner = self.env['res.partner'].create({'name': 'Cust'})
+        hold = self.env['fusion.plating.quality.hold'].create({
+            'partner_id': partner.id, 'state': 'on_hold', 'reason': 'old',
+        })
+        # Backdate create_date past the 3-day overdue threshold
+        old = datetime.now() - timedelta(days=5)
+        self.env.cr.execute(
+            "UPDATE fusion_plating_quality_hold SET create_date = %s WHERE id = %s",
+            (old, hold.id),
+        )
+        hold.invalidate_recordset(['create_date'])
+
+        snap = self._build()
+        self.assertFalse(snap['banner']['all_clear'])
+        self.assertEqual(snap['banner']['total_matching'], 1)
+        self.assertEqual(len(snap['banner']['items']), 1)
+        item = snap['banner']['items'][0]
+        self.assertEqual(item['type'], 'hold')
+        self.assertEqual(item['urgency'], 'overdue')
+
+    def test_banner_caps_at_6_with_overflow_count(self):
+        from datetime import datetime, timedelta
+        if 'fusion.plating.quality.hold' not in self.env:
+            self.skipTest('fusion.plating.quality.hold not installed')
+        partner = self.env['res.partner'].create({'name': 'Cust'})
+        Hold = self.env['fusion.plating.quality.hold']
+        # 8 overdue holds
+        old = datetime.now() - timedelta(days=5)
+        for i in range(8):
+            h = Hold.create({
+                'partner_id': partner.id, 'state': 'on_hold',
+                'reason': f'overdue {i}',
+            })
+            self.env.cr.execute(
+                "UPDATE fusion_plating_quality_hold SET create_date = %s WHERE id = %s",
+                (old, h.id),
+            )
+        snap = self._build()
+        self.assertEqual(len(snap['banner']['items']), 6)
+        self.assertEqual(snap['banner']['total_matching'], 8)
+        self.assertFalse(snap['banner']['all_clear'])
+
+    def test_banner_all_clear_when_zero(self):
+        snap = self._build()
+        # Empty DB — no overdue, no critical
+        self.assertTrue(snap['banner']['all_clear'])
+        self.assertEqual(snap['banner']['items'], [])