fix(psa): use schedule entries for ticket co-assignees (CW canonical pattern)

The previous implementation PATCHed the `resources` string directly, which CW
silently ignores because `resources` is a server-derived read-only field (it's
populated from schedule entries of type/id=4, not freely writable).

Per CW docs (openapi line 70949): "Please use the
/schedule/entries?conditions=type/id=4 AND objectId={id} endpoint".

Behavior per spec:
- No owner + assign user → set owner (existing behavior kept)
- Has owner + assign different user → POST /schedule/entries with type/id=4,
  member, objectId; owner untouched
- User already assigned (owner or schedule entry) → idempotent no-op
- Remove owner → clear owner (existing behavior kept)
- Remove co-assignee → DELETE /schedule/entries/{entry_id}
- list_resources now merges owner + schedule-entry members, deduped by id

Required CW security role permission on the API member:
- Service > Resource Scheduling > Add/Inquire/Delete

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

backend/app/services/psa/connectwise/provider.py

@@ -648,33 +648,46 @@ class ConnectWiseProvider(PSAProvider):
 
     # ── Resource management ───────────────────────────────────────────
 
-    async def _get_ticket_assignment(self, ticket_id: int) -> tuple[dict | None, str | None]:
-        """Fetch the ticket's current owner (MemberReference) and resources string.
+    # Schedule type id for "Service Ticket" resources — CW's canonical type for ticket co-assignees
+    _SCHEDULE_TYPE_SERVICE_TICKET = 4
 
-        Returns (owner_dict, resources_string). owner_dict has keys like
-        {id, identifier, name} when set; None when unassigned.
-        """
+    async def _get_ticket_owner(self, ticket_id: int) -> dict | None:
+        """Fetch the ticket's current owner (MemberReference) or None if unassigned."""
         data = await self.client.get(
             f"/service/tickets/{ticket_id}",
-            params={"fields": "id,owner,resources"},
+            params={"fields": "id,owner"},
         )
         if not isinstance(data, dict):
-            return None, None
+            return None
         owner_raw = data.get("owner")
-        owner = owner_raw if isinstance(owner_raw, dict) and owner_raw.get("id") else None
-        resources = data.get("resources")
-        return owner, (str(resources) if resources else None)
+        return owner_raw if isinstance(owner_raw, dict) and owner_raw.get("id") else None
+
+    async def _list_ticket_schedule_entries(self, ticket_id: int) -> list[dict]:
+        """List schedule entries for a ticket's co-assignees.
+
+        Returns raw CW schedule entry dicts with at least id and member info.
+        """
+        data = await self.client.get(
+            "/schedule/entries",
+            params={
+                "conditions": (
+                    f"type/id={self._SCHEDULE_TYPE_SERVICE_TICKET} AND objectId={ticket_id}"
+                ),
+                "fields": "id,member,name",
+                "pageSize": 100,
+            },
+        )
+        return data if isinstance(data, list) else []
 
     async def list_resources(self, ticket_id: int) -> list[PSAResource]:
         """List members assigned to a CW ticket.
 
-        CW exposes assignment in two places: the `owner` MemberReference (primary
-        assignee) and a derived `resources` string of identifiers. We merge both
-        so the UI shows everyone CW considers assigned.
+        Merges the `owner` MemberReference (primary assignee) with schedule entries
+        of type 4 (Service Ticket resources — co-assignees). Deduped by member id.
         """
-        owner, resources_str = await self._get_ticket_assignment(ticket_id)
+        owner = await self._get_ticket_owner(ticket_id)
+        entries = await self._list_ticket_schedule_entries(ticket_id)
         members = await self.list_members()
-        by_identifier = {m.identifier: m for m in members if m.identifier}
         by_id = {str(m.id): m for m in members}
 
         seen_ids: set[str] = set()
@@ -689,48 +702,55 @@ class ConnectWiseProvider(PSAProvider):
                     member_name=m.name,
                     member_identifier=m.identifier,
                 ))
-                seen_ids.add(owner_id)
             else:
                 results.append(PSAResource(
                     member_id=int(owner.get("id") or 0),
                     member_name=str(owner.get("name") or ""),
                     member_identifier=str(owner.get("identifier") or ""),
                 ))
-                seen_ids.add(owner_id)
+            seen_ids.add(owner_id)
 
-        if resources_str:
-            for ident in (p.strip() for p in resources_str.split(",") if p.strip()):
-                m = by_identifier.get(ident)
-                if m and str(m.id) not in seen_ids:
-                    results.append(PSAResource(
-                        member_id=int(m.id),
-                        member_name=m.name,
-                        member_identifier=m.identifier,
-                    ))
-                    seen_ids.add(str(m.id))
-                elif not m:
-                    results.append(PSAResource(
-                        member_id=0,
-                        member_name=ident,
-                        member_identifier=ident,
-                    ))
+        for entry in entries:
+            entry_member = entry.get("member") if isinstance(entry, dict) else None
+            if not isinstance(entry_member, dict):
+                continue
+            mid = str(entry_member.get("id") or "")
+            if not mid or mid in seen_ids:
+                continue
+            m = by_id.get(mid)
+            if m:
+                results.append(PSAResource(
+                    member_id=int(m.id),
+                    member_name=m.name,
+                    member_identifier=m.identifier,
+                ))
+            else:
+                results.append(PSAResource(
+                    member_id=int(entry_member.get("id") or 0),
+                    member_name=str(entry_member.get("name") or ""),
+                    member_identifier=str(entry_member.get("identifier") or ""),
+                ))
+            seen_ids.add(mid)
 
         return results
 
     async def add_resource(self, ticket_id: int, member_id: int) -> PSAResource:
         """Assign a member to a CW ticket.
 
-        CW's `resources` string is derived server-side and PATCHing it is unreliable.
-        The canonical writable field is `owner`. If the ticket is unassigned we set
-        the owner; if it already has a different owner, we append to `resources`
-        (best-effort — CW may not surface it in every view).
+        - If the ticket has no owner, set the target as `owner` (CW's canonical
+          primary assignee field). CW typically mirrors this into the derived
+          `resources` string automatically.
+        - If the ticket is already owned by someone else, add the target as a
+          co-assignee via a schedule entry of type 4 (Service Ticket). The
+          existing owner is not changed.
+        - Idempotent when target is already owner or already has a schedule entry.
         """
         members = await self.list_members()
         target = next((m for m in members if str(m.id) == str(member_id)), None)
         if target is None:
             raise PSAError(f"Member {member_id} not found")
 
-        current_owner, resources_str = await self._get_ticket_assignment(ticket_id)
+        current_owner = await self._get_ticket_owner(ticket_id)
 
         if current_owner is None:
             # Primary assign — set owner
@@ -739,13 +759,22 @@ class ConnectWiseProvider(PSAProvider):
                 json_body=[{"op": "replace", "path": "owner", "value": {"id": int(target.id)}}],
             )
         elif str(current_owner.get("id")) != str(target.id):
-            # Ticket already owned by someone else — add to resources string as co-assignee.
-            identifiers = [p.strip() for p in (resources_str or "").split(",") if p.strip()]
-            if target.identifier and target.identifier not in identifiers:
-                identifiers.append(target.identifier)
-                await self.client.patch(
-                    f"/service/tickets/{ticket_id}",
-                    json_body=[{"op": "replace", "path": "resources", "value": ",".join(identifiers)}],
+            # Ticket owned by someone else — add as co-assignee via schedule entry.
+            # Idempotent: skip if a schedule entry already exists for this member.
+            existing = await self._list_ticket_schedule_entries(ticket_id)
+            already_assigned = any(
+                str((e.get("member") or {}).get("id") or "") == str(target.id)
+                for e in existing
+            )
+            if not already_assigned:
+                await self.client.post(
+                    "/schedule/entries",
+                    json_body={
+                        "member": {"id": int(target.id)},
+                        "objectId": int(ticket_id),
+                        "type": {"id": self._SCHEDULE_TYPE_SERVICE_TICKET},
+                        "name": target.name or target.identifier or f"Member {target.id}",
+                    },
                 )
         # else: already the owner — idempotent no-op
 
@@ -758,15 +787,15 @@ class ConnectWiseProvider(PSAProvider):
     async def remove_resource(self, ticket_id: int, member_id: int) -> None:
         """Remove a member from a CW ticket (idempotent).
 
-        If the target is the current owner, clear the owner. Otherwise remove them
-        from the `resources` string.
+        - If the target is the current owner, clear the owner field.
+        - Otherwise, delete their schedule entry (Service Ticket type).
         """
         members = await self.list_members()
         target = next((m for m in members if str(m.id) == str(member_id)), None)
         if target is None:
             return
 
-        current_owner, resources_str = await self._get_ticket_assignment(ticket_id)
+        current_owner = await self._get_ticket_owner(ticket_id)
 
         if current_owner is not None and str(current_owner.get("id")) == str(target.id):
             # Unassign the owner. Try RFC 6902 "remove" first; fall back to
@@ -783,14 +812,15 @@ class ConnectWiseProvider(PSAProvider):
                 )
             return
 
-        if target.identifier and resources_str:
-            identifiers = [p.strip() for p in resources_str.split(",") if p.strip()]
-            if target.identifier in identifiers:
-                new_list = [i for i in identifiers if i != target.identifier]
-                await self.client.patch(
-                    f"/service/tickets/{ticket_id}",
-                    json_body=[{"op": "replace", "path": "resources", "value": ",".join(new_list)}],
-                )
+        # Not the owner — find and delete the schedule entry for this member.
+        entries = await self._list_ticket_schedule_entries(ticket_id)
+        for entry in entries:
+            entry_member = entry.get("member") if isinstance(entry, dict) else None
+            if isinstance(entry_member, dict) and str(entry_member.get("id") or "") == str(target.id):
+                entry_id = entry.get("id")
+                if entry_id:
+                    await self.client.delete(f"/schedule/entries/{entry_id}")
+                break
 
     # ── Ticket creation ───────────────────────────────────────────────