Implement cross-system issue synchronization

[hussam789]

User description

PR #7

---

PR Type

Bug fix, Enhancement

---

Description

• Implement AssignmentSource tracking to prevent sync cycles

• Add assignment source parameter to sync methods

• Prevent outbound sync when source matches target integration

• Update group assignee operations to pass assignment source

---

Diagram Walkthrough

flowchart LR
  A["Assignment Action"] --> B["AssignmentSource Created"]
  B --> C["Sync Methods Check Source"]
  C --> D{"Source == Target Integration?"}
  D -->|Yes| E["Skip Sync - Prevent Cycle"]
  D -->|No| F["Proceed with Sync"]

Loading

File Walkthrough

	Relevant files
Enhancement	assignment_source.py New assignment source tracking dataclass                                  src/sentry/integrations/services/assignment_source.py New dataclass AssignmentSource to track sync operation origin Includes source_name, integration_id, and queued timestamp Provides from_integration() factory method for creating instances Implements to_dict() and from_dict() for serialization +35/-0    sync_assignee_outbound.py Thread assignment source through outbound sync task            src/sentry/integrations/tasks/sync_assignee_outbound.py Add assignment_source_dict parameter to sync_assignee_outbound() task Parse assignment source from dictionary using AssignmentSource.from_dict() Pass parsed assignment source to should_sync() and sync_assignee_outbound() calls +16/-3    sync.py Add assignment source propagation to sync utilities            src/sentry/integrations/utils/sync.py Import AssignmentSource class Update sync_group_assignee_inbound() to pass assignment source when assigning/deassigning Update sync_group_assignee_outbound() to accept and forward assignment source parameter Serialize assignment source to dictionary when queuing async tasks +25/-4    groupassignee.py Add assignment source to group assignee operations              src/sentry/models/groupassignee.py Import AssignmentSource class Add assignment_source parameter to assign() method Add assignment_source parameter to deassign() method Pass assignment source to sync_group_assignee_outbound() calls +9/-2
Bug fix	issues.py Add sync cycle prevention to integration mixins                    src/sentry/integrations/mixins/issues.py Add AssignmentSource import Update should_sync() method signature to accept optional sync_source parameter Add cycle prevention logic in IssueSyncIntegration.should_sync() to check if source integration matches target Update sync_status_outbound() abstract method signature to include assignment_source parameter +18/-3
Tests	test_assignment_source.py New tests for assignment source serialization                        tests/sentry/integrations/services/test_assignment_source.py Test from_dict() with empty dictionary returns None Test from_dict() with invalid data returns None Test from_dict() with valid data creates proper instance Test to_dict() serialization includes all fields +38/-0    test_groupassignee.py Add tests for sync cycle prevention                                            tests/sentry/models/test_groupassignee.py Import AssignmentSource class Update existing test to verify assignment_source=None parameter Add new test test_assignee_sync_outbound_assign_with_matching_source_integration() to verify sync is skipped when source matches target Update deassign test to verify assignment source parameter handling +71/-3

---

[qodo-code-review]

PR Compliance Guide 🔍

Below is a summary of compliance checks for this PR:

Security Compliance
⚪	Mutable default timestamp Description: The dataclass field 'queued' uses timezone.now() directly as a default which is evaluated at import time, causing all instances without an explicit 'queued' to share the same timestamp; use default_factory=timezone.now instead. assignment_source.py [14-35] Referred Code @dataclass(frozen=True) class AssignmentSource: source_name: str integration_id: int queued: datetime = timezone.now() @classmethod def from_integration(cls, integration: Integration | RpcIntegration) -> AssignmentSource: return AssignmentSource( source_name=integration.name, integration_id=integration.id, ) def to_dict(self) -> dict[str, Any]: return asdict(self) @classmethod def from_dict(cls, input_dict: dict[str, Any]) -> AssignmentSource | None: try: return cls(**input_dict) except (ValueError, TypeError): ... (clipped 1 lines)
Ticket Compliance
⚪	🎫 No ticket provided Create ticket/issue
Codebase Duplication Compliance
⚪	Codebase context is not defined Follow the guide to enable codebase context checks.
Custom Compliance
🟢	Generic: Meaningful Naming and Self-Documenting Code Objective: Ensure all identifiers clearly express their purpose and intent, making code self-documenting Status: Passed
	Generic: Secure Error Handling Objective: To prevent the leakage of sensitive system information through error messages while providing sufficient detail for internal debugging. Status: Passed
	Generic: Secure Logging Practices Objective: To ensure logs are useful for debugging and auditing without exposing sensitive information like PII, PHI, or cardholder data. Status: Passed
⚪	Generic: Comprehensive Audit Trails Objective: To create a detailed and reliable record of critical system actions for security analysis and compliance. Status: Missing auditing: New assignment propagation and cycle-prevention logic executes without adding or verifying audit log entries capturing user/integration source, timestamp, and outcome. Referred Code "issue_key": external_issue_key, } if not affected_groups: logger.info("no-affected-groups", extra=log_context) return [] if not assign: for group in affected_groups: GroupAssignee.objects.deassign( group, assignment_source=AssignmentSource.from_integration(integration), ) return affected_groups users = user_service.get_many_by_email(emails=[email], is_verified=True) users_by_id = {user.id: user for user in users} projects_by_user = Project.objects.get_by_users(users) groups_assigned = [] for group in affected_groups: ... (clipped 9 lines)
	Generic: Robust Error Handling and Edge Case Management Objective: Ensure comprehensive error handling that provides meaningful context and graceful degradation Status: Edge case handling: The dataclass sets a timezone-aware default at class definition time and from_dict silently returns None on invalid input without logging, which may obscure issues when assignment_source is malformed. Referred Code @dataclass(frozen=True) class AssignmentSource: source_name: str integration_id: int queued: datetime = timezone.now() @classmethod def from_integration(cls, integration: Integration | RpcIntegration) -> AssignmentSource: return AssignmentSource( source_name=integration.name, integration_id=integration.id, ) def to_dict(self) -> dict[str, Any]: return asdict(self) @classmethod def from_dict(cls, input_dict: dict[str, Any]) -> AssignmentSource | None: try: return cls(**input_dict) except (ValueError, TypeError): ... (clipped 1 lines)
	Generic: Security-First Input Validation and Data Handling Objective: Ensure all data inputs are validated, sanitized, and handled securely to prevent vulnerabilities Status: Input validation: The task accepts assignment_source_dict from external queues and uses it after minimal validation, relying on dataclass coercion without explicit schema checks or logging when input is invalid. Referred Code def sync_assignee_outbound( external_issue_id: int, user_id: int | None, assign: bool, assignment_source_dict: dict[str, Any] | None = None, ) -> None: # Sync Sentry assignee to an external issue. external_issue = ExternalIssue.objects.get(id=external_issue_id) organization = Organization.objects.get(id=external_issue.organization_id) has_issue_sync = features.has("organizations:integrations-issue-sync", organization) if not has_issue_sync: return integration = integration_service.get_integration(integration_id=external_issue.integration_id) if not integration: return installation = integration.get_installation(organization_id=external_issue.organization_id) if not ( hasattr(installation, "should_sync") and hasattr(installation, "sync_assignee_outbound") ): ... (clipped 11 lines)

Compliance status legend

🟢 - Fully Compliant
🟡 - Partial Compliant
🔴 - Not Compliant
⚪ - Requires Further Human Verification
🏷️ - Compliance label

[qodo-code-review]

PR Code Suggestions ✨

Explore these optional code suggestions:

Category	Suggestion	Impact
General	Explicitly add parameter to method signature Explicitly add the assignment_source: AssignmentSource | None = None parameter to the sync_assignee_outbound abstract method signature for improved clarity and consistency. src/sentry/integrations/mixins/issues.py [396-408] @abstractmethod def sync_assignee_outbound( self, external_issue: ExternalIssue, user: RpcUser | None, assign: bool = True, + assignment_source: AssignmentSource | None = None, **kwargs: Any, ) -> None: """ Propagate a sentry issue's assignee to a linked issue's assignee. """ raise NotImplementedError Apply / Chat Suggestion importance[1-10]: 7 __ Why: The suggestion correctly points out that assignment_source is passed to sync_assignee_outbound but is not an explicit parameter in its abstract definition. Adding it improves code clarity, consistency with other methods in the file, and type safety for implementers.	Medium
	Improve deserialization to handle extra keys Improve the from_dict method in AssignmentSource to be more robust by explicitly checking for required parameters and filtering out extraneous keys, rather than broadly catching TypeError. src/sentry/integrations/services/assignment_source.py [30-35] @classmethod def from_dict(cls, input_dict: dict[str, Any]) -> AssignmentSource | None: - try: - return cls(**input_dict) - except (ValueError, TypeError): + import inspect + + if not isinstance(input_dict, dict): return None + required_params = { + p.name + for p in inspect.signature(cls).parameters.values() + if p.default is inspect.Parameter.empty + } + if not required_params.issubset(input_dict.keys()): + return None + + # Filter out unexpected keys to avoid TypeError on `__init__` + known_params = {p.name for p in inspect.signature(cls).parameters.values()} + filtered_dict = {k: v for k, v in input_dict.items() if k in known_params} + + try: + return cls(**filtered_dict) + except TypeError: + # This could still fail if types are wrong, e.g. integration_id="abc" + return None + Apply / Chat Suggestion importance[1-10]: 6 __ Why: The suggestion correctly identifies that the broad exception handling in the new from_dict method can hide bugs by silently returning None. The proposed change improves robustness by explicitly handling required and extraneous parameters.	Low
More