fix(ecosystem): Breaks issue sync cycles

[linxia0415]

The problem:

When a Jira issue's assignee field is quickly modified multiple times in a row, we end up assigning and resyncing the assignees to Jira, which causes a loop. This is because all inbound integration webhooks that modify an issue assignment generate an outbound update for all issue sync integrations as well.

Solution:

This PR adds a new AssignmentSource dataclass, which acts as a context carrier when an external webhook triggers an issue assignment change. This is then checked on the outbound sync logic, preventing the change from being propagated to the same integration that initially triggered the change.

[code-hawk-sit]

AI代码审查报告

变更概览

本次 PR 涉及 7 个文件，新增 +212 行，删除 -15 行。

功能变更摘要

本 PR 在集成系统中引入分配来源追踪机制，通过新增服务层管理分配来源，并在模型、同步任务及混入类中集成来源信息记录，以区分用户手动与系统自动分配。

文件变更摘要

文件	变更	行数	摘要	发现问题
src/sentry/integrations/services/assignment_source.py	新增	+35/-0	新增分配来源管理服务类	1 个
src/sentry/integrations/tasks/sync_assignee_outbound.py	修改	+16/-3	同步任务中携带并传递分配来源信息	1 个
tests/sentry/models/test_groupassignee.py	修改	+71/-3	补充模型层分配来源相关测试用例	—
tests/sentry/integrations/services/test_assignment_source.py	新增	+38/-0	新增分配来源服务的单元测试	—
src/sentry/integrations/utils/sync.py	修改	+25/-4	工具函数支持分配来源参数传递	—
src/sentry/integrations/mixins/issues.py	修改	+18/-3	集成分配来源校验逻辑以控制分配权限	—
src/sentry/models/groupassignee.py	修改	+9/-2	模型层添加分配来源字段存储	—

问题严重级别分布

级别	数量	占比
🟡 中危	2	100%

代表性问题（至多 10 条，按严重级别优先）

1. 🟡 中危 src/sentry/integrations/services/assignment_source.py L17: 可变默认参数陷阱

line 17 处 queued: datetime = timezone.now() 在类定义时即求值，导致所有未显式传入 queued 的实例共享同一时间戳（类定义时的…
2. 🟡 中危 src/sentry/integrations/tasks/sync_assignee_outbound.py L55: 向后兼容性风险：方法签名不匹配

line 48-50 处仅检查 should_sync 和 sync_assignee_outbound 方法是否存在，但 line 55 和 line 59…

---

由 CodeHawk 提供支持 · nuwa

[code-hawk-sit]

🟡 AI 代码审查发现问题

📋 问题概述

可变默认参数陷阱

line 17 处 queued: datetime = timezone.now() 在类定义时即求值，导致所有未显式传入 queued 的实例共享同一时间戳（类定义时的时间），而非实例创建时间。当通过 line 21-24 from_integration() 创建实例时，若调用方未提供 queued 参数，记录的时间将永远是模块首次加载时的时间，造成分配来源时间记录错误。

📍 问题详情

🟡 问题 1 | 严重程度: MEDIUM | 行号: 17

💬 详细说明：

• 可变默认参数陷阱
• line 17 处 queued: datetime = timezone.now() 在类定义时即求值，导致所有未显式传入 queued 的实例共享同一时间戳（类定义时的时间），而非实例创建时间。当通过 line 21-24 from_integration() 创建实例时，若调用方未提供 queued 参数，记录的时间将永远是模块首次加载时的时间，造成分配来源时间记录错误。

📝 问题代码：

    queued: datetime = timezone.now()

💡 修复建议：

使用 dataclasses.field(default_factory=...) 替代直接默认值，确保每次实例化时动态获取当前时间。将 timezone.now() 改为 field(default_factory=timezone.now)，使 queued 在实例创建时才求值。

✅ 修复示例：

    queued: datetime = field(default_factory=timezone.now)

---

🔗 参考链接

无

[code-hawk-sit]

🟡 AI 代码审查发现问题

📋 问题概述

向后兼容性风险：方法签名不匹配

line 48-50 处仅检查 should_sync 和 sync_assignee_outbound 方法是否存在，但 line 55 和 line 59 处调用这些方法时传入了新增的 assignment_source 参数。若某些 integration installation 的旧实现仅支持原有签名（如 should_sync(self, sync_type) 或 sync_assignee_outbound(self, external_issue, user, assign)），调用将抛出 TypeError 导致任务失败。

当调用方传入 assignment_source_dict 且 installation 为旧实现时，line 55 的 should_sync 调用会因参数过多而崩溃。

📍 问题详情

🟡 问题 1 | 严重程度: MEDIUM | 行号: 55

💬 详细说明：

• 向后兼容性风险：方法签名不匹配
• line 48-50 处仅检查 should_sync 和 sync_assignee_outbound 方法是否存在，但 line 55 和 line 59 处调用这些方法时传入了新增的 assignment_source 参数。若某些 integration installation 的旧实现仅支持原有签名（如 should_sync(self, sync_type) 或 sync_assignee_outbound(self, external_issue, user, assign)），调用将抛出 TypeError 导致任务失败。
• 当调用方传入 assignment_source_dict 且 installation 为旧实现时，line 55 的 should_sync 调用会因参数过多而崩溃。

📝 问题代码：

    if installation.should_sync("outbound_assignee", parsed_assignment_source):

💡 修复建议：

在调用前检查方法签名是否支持新参数，或采用安全降级策略：

1. 使用 inspect.signature 检查 should_sync 和 sync_assignee_outbound 是否接受 assignment_source 参数
2. 若不支持，则降级为旧版调用（不传 assignment_source 参数）
3. 或在 hasattr 检查后增加版本/能力检查，确保 installation 支持新接口

具体实现：在 line 52-54 后增加签名检查逻辑，对不支持新参数的 installation 使用兼容模式调用。

✅ 修复示例：

    # Check if installation supports assignment_source parameter
    import inspect
    
    should_sync_supports_source = False
    sync_assignee_supports_source = False
    
    try:
        should_sync_sig = inspect.signature(installation.should_sync)
        should_sync_supports_source = len(should_sync_sig.parameters) > 2  # self + sync_type + source
    except (ValueError, TypeError):
        pass
    
    try:
        sync_sig = inspect.signature(installation.sync_assignee_outbound)
        sync_assignee_supports_source = 'assignment_source' in sync_sig.parameters
    except (ValueError, TypeError):
        pass
    
    parsed_assignment_source = (
        AssignmentSource.from_dict(assignment_source_dict) if assignment_source_dict else None
    )
    
    if should_sync_supports_source:
        should_sync_result = installation.should_sync("outbound_assignee", parsed_assignment_source)
    else:
        should_sync_result = installation.should_sync("outbound_assignee")
    
    if should_sync_result:
        user = user_service.get_user(user_id) if user_id else None
        if sync_assignee_supports_source:
            installation.sync_assignee_outbound(
                external_issue, user, assign=assign, assignment_source=parsed_assignment_source
            )
        else:
            installation.sync_assignee_outbound(external_issue, user, assign=assign)

---

🔗 参考链接

无