fix(configurator/bridge_mrp): address all bugs from code review

Two critical, one important, four polish fixes found by the
pr-review-toolkit code-reviewer.

C1 (CRITICAL) Start-at-node filter dropped later siblings
  fusion_plating_bridge_mrp/models/mrp_production.py:448
  The allowed_ids set was {descendants} ∪ {ancestors}, which wrongly
  excluded nodes that should run AFTER the start node — including
  later siblings of the start node and all operations in subsequent
  sub-processes. Rewrote the upward walk to ALSO include each
  ancestor's later-sequence siblings and their descendants. Smoke on
  ENP-ALUM-BASIC: full=9 WOs, partial from mid-tree 'De-Masking'=5
  WOs (previously was 1).

C2 (CRITICAL) Duplicate MO on re-confirm of pre-PR SOs
  fusion_plating_bridge_mrp/models/sale_order.py:96
  Legacy untagged MOs (created before this PR had line-linkage m2m)
  were not recognized by the untagged idempotency check, so
  re-confirming an already-processed SO would create one additional
  MO per untagged plating line. Fix: pre-scan for a single legacy
  untagged MO and adopt it by linking ALL untagged plating lines
  onto it. Those lines are then treated as covered and no per-line
  MOs are created on top. Smoke: S00066 before=1 MO, after
  re-run=1 MO.

I5 (IMPORTANT) push_to_defaults wrote to pre-bump revision
  fusion_plating_configurator/wizard/fp_direct_order_wizard.py:236
  When create_new_revision=True, _get_or_bump_revision() returned a
  new part record that got written to the SO line, but the
  post-confirm push_to_defaults loop re-read line.part_catalog_id
  (still the OLD rev) and wrote defaults there, defeating the whole
  point of "save as default". Fix: cache resolved parts in a dict
  keyed by wizard-line ID during the build loop, and use that cache
  in the push_to_defaults pass.

I3/I4/I6 (PERF) Computes lacked @api.depends and did per-record
  search_count / search queries
  fusion_plating_configurator/models/sale_order.py
  _compute_nav_counts, _compute_workorder_count, _compute_wo_completion
  now:
  - declare @api.depends
  - batch via read_group across the whole self recordset
  - rebuild {origin: counts} dicts and assign per record

M7 (MEDIUM) No savepoint around per-group MO creation
  fusion_plating_bridge_mrp/models/sale_order.py:_fp_auto_create_mo
  A mid-loop exception left group 1's MO persisted and aborted
  groups 2..N. Wrapped each group's create in SAVEPOINT/RELEASE/
  ROLLBACK TO SAVEPOINT so one bad group no longer corrupts state.

M8 (MEDIUM) Email 'opened' status false-positived on internal CC
  fusion_plating_configurator/models/sale_order.py:_compute_email_status
  Switched from 'any notification is_read' to 'customer partner has
  a read email notification on this SO'.

M9 (LOW) start_at_node_id domain silently empty when coating unset
  fusion_plating_configurator/wizard/fp_direct_order_line.py:94
  Changed `('parent_id', 'child_of', ...)` to
  `('id', 'child_of', ..., or 0)` and clarified the help text.

Regression smoke passed all checks on odoo-entech.

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

fusion_plating/fusion_plating_configurator/models/sale_order.py

@@ -131,18 +131,44 @@ class SaleOrder(models.Model):
         currency_field='currency_id',
     )
 
+    @api.depends('name')
     def _compute_wo_completion(self):
+        """Batched: one grouped query across all records in self."""
+        for rec in self:
+            rec.x_fc_wo_completion = '0/0'
+        names = [so.name for so in self if so.name]
+        if not names:
+            return
         WO = self.env['mrp.workorder'].sudo()
+        rows = WO.read_group(
+            [('production_id.origin', 'in', names)],
+            ['production_id.origin', 'state'],
+            ['production_id', 'state'],
+            lazy=False,
+        )
+        # Build {origin: {'done': n, 'total': n}}
+        # read_group returns production_id as (id, name) tuples; we need
+        # to translate back to origin. Do a small lookup.
+        mos = self.env['mrp.production'].sudo().search(
+            [('origin', 'in', names)]
+        )
+        mo_to_origin = {m.id: m.origin for m in mos}
+        totals = {}  # {origin: [total, done]}
+        for r in rows:
+            mo_id = r['production_id'][0] if r['production_id'] else False
+            origin = mo_to_origin.get(mo_id)
+            if not origin:
+                continue
+            cnt = r['__count']
+            bucket = totals.setdefault(origin, [0, 0])
+            bucket[0] += cnt
+            if r['state'] == 'done':
+                bucket[1] += cnt
         for rec in self:
             if not rec.name:
-                rec.x_fc_wo_completion = '0/0'
                 continue
-            total = WO.search_count([('production_id.origin', '=', rec.name)])
-            done = WO.search_count([
-                ('production_id.origin', '=', rec.name),
-                ('state', '=', 'done'),
-            ])
-            rec.x_fc_wo_completion = '%d/%d' % (done, total) if total else '0/0'
+            tot, done = totals.get(rec.name, [0, 0])
+            rec.x_fc_wo_completion = '%d/%d' % (done, tot) if tot else '0/0'
 
     # ---- Phase F: quotes list view polish ----
     x_fc_follow_up_date = fields.Date(
@@ -195,10 +221,14 @@ class SaleOrder(models.Model):
     def _compute_email_status(self):
         """Map state + mail tracking to a single visible pill.
 
-        - draft SO with no tracked email sent => draft
-        - sent (Odoo state) => sent
-        - sent + mail opened => opened (detected via mail.message)
-        - state=sale/done => won
+        - state draft => draft
+        - state sent => sent (or 'opened' if the customer partner has
+          a read notification for any email message on this SO)
+        - state sale / done => won
+
+        'Opened' is scoped to the CUSTOMER partner's notifications —
+        not internal CCs — to avoid false positives from sales-ops
+        viewing the thread.
         """
         for rec in self:
             if rec.state in ('sale', 'done'):
@@ -209,19 +239,17 @@ class SaleOrder(models.Model):
                 continue
             # state == 'sent'
             opened = False
-            if rec.id:
-                msgs = self.env['mail.message'].sudo().search([
-                    ('model', '=', 'sale.order'),
-                    ('res_id', '=', rec.id),
-                    ('message_type', '=', 'email'),
-                ], limit=10)
-                # mail.notification tracks read timestamps
-                for m in msgs:
-                    if m.notification_ids.filtered(
-                        lambda n: n.is_read
-                    ):
-                        opened = True
-                        break
+            if rec.id and rec.partner_id:
+                # Look for any read notification on any email message
+                # of this SO that targeted the customer.
+                notif_count = self.env['mail.notification'].sudo().search_count([
+                    ('mail_message_id.model', '=', 'sale.order'),
+                    ('mail_message_id.res_id', '=', rec.id),
+                    ('mail_message_id.message_type', '=', 'email'),
+                    ('res_partner_id', '=', rec.partner_id.id),
+                    ('is_read', '=', True),
+                ])
+                opened = notif_count > 0
             rec.x_fc_email_status = 'opened' if opened else 'sent'
 
     @api.depends('order_line.x_fc_part_catalog_id.part_number')
@@ -254,16 +282,33 @@ class SaleOrder(models.Model):
                 - sum(refunds.mapped('amount_total'))
             )
 
+    @api.depends('name')
     def _compute_workorder_count(self):
-        WO = self.env['mrp.workorder'].sudo()
         for rec in self:
-            if not rec.name:
-                rec.x_fc_workorder_count = 0
-                continue
-            rec.x_fc_workorder_count = WO.search_count([
-                ('production_id.origin', '=', rec.name),
-                ('state', 'not in', ('done', 'cancel')),
-            ])
+            rec.x_fc_workorder_count = 0
+        names = [so.name for so in self if so.name]
+        if not names:
+            return
+        WO = self.env['mrp.workorder'].sudo()
+        rows = WO.read_group(
+            [('production_id.origin', 'in', names),
+             ('state', 'not in', ('done', 'cancel'))],
+            ['production_id'],
+            ['production_id'],
+            lazy=False,
+        )
+        mos = self.env['mrp.production'].sudo().search(
+            [('origin', 'in', names)]
+        )
+        mo_to_origin = {m.id: m.origin for m in mos}
+        totals = {}
+        for r in rows:
+            mo_id = r['production_id'][0] if r['production_id'] else False
+            origin = mo_to_origin.get(mo_id)
+            if origin:
+                totals[origin] = totals.get(origin, 0) + r['__count']
+        for rec in self:
+            rec.x_fc_workorder_count = totals.get(rec.name, 0)
 
     def action_view_workorders(self):
         self.ensure_one()
@@ -290,21 +335,41 @@ class SaleOrder(models.Model):
         string='Files', compute='_compute_nav_counts',
     )
 
+    @api.depends('invoice_ids', 'picking_ids')
     def _compute_nav_counts(self):
-        NCR = self.env.get('fusion.plating.ncr')
+        # Invoice + picking counts are cheap (related collections).
         for rec in self:
             rec.x_fc_invoice_count = len(rec.invoice_ids)
             rec.x_fc_picking_count = len(rec.picking_ids)
-            rec.x_fc_attachment_count = self.env['ir.attachment'].sudo().search_count([
-                ('res_model', '=', 'sale.order'),
-                ('res_id', '=', rec.id),
-            ])
-            if NCR and 'sale_order_id' in NCR._fields:
-                rec.x_fc_ncr_count = NCR.sudo().search_count([
-                    ('sale_order_id', '=', rec.id),
-                ])
-            else:
-                rec.x_fc_ncr_count = 0
+
+        # Attachment counts — batched read_group.
+        ids = self.ids
+        att_counts = {}
+        if ids:
+            rows = self.env['ir.attachment'].sudo().read_group(
+                [('res_model', '=', 'sale.order'),
+                 ('res_id', 'in', ids)],
+                ['res_id'], ['res_id'], lazy=False,
+            )
+            att_counts = {r['res_id']: r['__count'] for r in rows}
+        for rec in self:
+            rec.x_fc_attachment_count = att_counts.get(rec.id, 0)
+
+        # NCR counts — only if the module is installed.
+        NCR = self.env.get('fusion.plating.ncr')
+        ncr_counts = {}
+        if ids and NCR is not None and 'sale_order_id' in NCR._fields:
+            rows = NCR.sudo().read_group(
+                [('sale_order_id', 'in', ids)],
+                ['sale_order_id'], ['sale_order_id'], lazy=False,
+            )
+            ncr_counts = {
+                (r['sale_order_id'][0] if r['sale_order_id'] else False):
+                    r['__count']
+                for r in rows
+            }
+        for rec in self:
+            rec.x_fc_ncr_count = ncr_counts.get(rec.id, 0)
 
     def action_view_invoices(self):
         self.ensure_one()

fusion_plating/fusion_plating_configurator/wizard/fp_direct_order_line.py

@@ -94,9 +94,11 @@ class FpDirectOrderLine(models.TransientModel):
     start_at_node_id = fields.Many2one(
         'fusion.plating.process.node',
         string='Start at Node',
-        domain="[('parent_id', 'child_of', coating_config_id and coating_config_id.recipe_id.id)]",
+        domain="[('id', 'child_of', coating_config_id and coating_config_id.recipe_id.id or 0)]",
         help='For re-work jobs: pick the recipe step where this job should '
-             'begin. Skips ancestor steps in the generated work order.',
+             'begin. Pick a coating first — nodes are scoped to its '
+             'recipe tree. Skips earlier steps in the generated WO but '
+             'keeps later siblings and sub-processes.',
     )
     is_one_off = fields.Boolean(
         string='One-off Part',

fusion_plating/fusion_plating_configurator/wizard/fp_direct_order_wizard.py

@@ -235,9 +235,13 @@ class FpDirectOrderWizard(models.TransientModel):
             'order_line': [],
         }
 
-        # 4. One SO line per wizard line
+        # 4. One SO line per wizard line. Cache resolved parts (post
+        # rev-bump) so the push-to-defaults pass writes to the right
+        # catalog entry.
+        resolved_parts = {}  # {wizard_line_id: resolved part record}
         for line in self.line_ids:
             part = line._get_or_bump_revision()
+            resolved_parts[line.id] = part
             header = '%s - %s Rev %s (x%d)' % (
                 line.coating_config_id.name,
                 part.name,
@@ -270,14 +274,14 @@ class FpDirectOrderWizard(models.TransientModel):
         so = self.env['sale.order'].create(so_vals)
         so.action_confirm()
 
-        # 6. Push-to-defaults (C4) — after the part has been resolved /
-        # revision-bumped, write coating + treatments back onto the part
-        # catalog entry so the next order inherits the same defaults.
+        # 6. Push-to-defaults (C4) — uses the resolved part cached
+        # during the build loop so rev-bumped lines write defaults to
+        # the NEW revision, not the pre-bump one.
         for line in self.line_ids:
-            if not line.push_to_defaults:
+            if not line.push_to_defaults or line.is_one_off:
                 continue
-            part = line.part_catalog_id
-            if not part or line.is_one_off:
+            part = resolved_parts.get(line.id) or line.part_catalog_id
+            if not part:
                 continue
             part.write({
                 'x_fc_default_coating_config_id': line.coating_config_id.id or False,