chore(api): remove generated API spec changes

api/changelog.d/jira-issue-dedup.added.md

@@ -0,0 +1 @@
+Finding-to-Jira issue tracking across scans, concurrent duplicate prevention, completed-issue replacement, and confirmed links through GET /api/v1/jira-issues

api/src/backend/api/filters.py

@@ -17,6 +17,7 @@ from api.models import (
     FindingGroupDailySummary,
     Integration,
     Invitation,
+    JiraIssue,
     LighthouseProviderConfiguration,
     LighthouseProviderModels,
     Membership,
@@ -1900,3 +1901,30 @@ class ComplianceWatchlistFilter(BaseProviderFilter):
 
     class Meta(BaseProviderFilter.Meta):
         model = ProviderComplianceScore
+
+
+class JiraIssueFilter(BaseProviderFilter):
+    finding_uid = CharFilter(field_name="finding_uid", lookup_expr="exact")
+    finding_uid__in = CharInFilter(field_name="finding_uid", lookup_expr="in")
+    finding_id = UUIDFilter(field_name="finding_id", lookup_expr="exact")
+    finding_id__in = UUIDInFilter(field_name="finding_id", lookup_expr="in")
+    integration = UUIDFilter(field_name="integration__id", lookup_expr="exact")
+    integration__in = UUIDInFilter(field_name="integration__id", lookup_expr="in")
+    issue_key = CharFilter(field_name="issue_key", lookup_expr="exact")
+    issue_key__in = CharInFilter(field_name="issue_key", lookup_expr="in")
+    issue_status_category = ChoiceFilter(
+        choices=JiraIssue.StatusCategoryChoices.choices
+    )
+    issue_status_category__in = ChoiceInFilter(
+        choices=JiraIssue.StatusCategoryChoices.choices,
+        field_name="issue_status_category",
+        lookup_expr="in",
+    )
+
+    class Meta(BaseProviderFilter.Meta):
+        model = JiraIssue
+        fields = {
+            "inserted_at": ["date", "gte", "lte"],
+            "updated_at": ["date", "gte", "lte"],
+            "project_key": ["exact", "in"],
+        }

api/src/backend/api/migrations/0098_jira_issues.py

@@ -0,0 +1,161 @@
+import uuid
+
+import api.rls
+import django.db.models.deletion
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("api", "0097_attack_paths_scan_db_defaults"),
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name="JiraIssue",
+            fields=[
+                (
+                    "id",
+                    models.UUIDField(
+                        default=uuid.uuid4,
+                        editable=False,
+                        primary_key=True,
+                        serialize=False,
+                    ),
+                ),
+                ("inserted_at", models.DateTimeField(auto_now_add=True)),
+                ("updated_at", models.DateTimeField(auto_now=True)),
+                ("finding_uid", models.CharField(max_length=300)),
+                ("finding_id", models.UUIDField()),
+                (
+                    "issue_id",
+                    models.CharField(blank=True, max_length=64, null=True),
+                ),
+                (
+                    "issue_key",
+                    models.CharField(blank=True, max_length=64, null=True),
+                ),
+                (
+                    "issue_url",
+                    models.URLField(blank=True, max_length=2048, null=True),
+                ),
+                (
+                    "project_key",
+                    models.CharField(blank=True, max_length=64, null=True),
+                ),
+                (
+                    "issue_status",
+                    models.CharField(blank=True, max_length=64, null=True),
+                ),
+                (
+                    "issue_status_category",
+                    models.CharField(
+                        blank=True,
+                        choices=[
+                            ("new", "New"),
+                            ("indeterminate", "In progress"),
+                            ("done", "Done"),
+                        ],
+                        max_length=16,
+                        null=True,
+                    ),
+                ),
+                ("status_synced_at", models.DateTimeField(blank=True, null=True)),
+                (
+                    "delivery_attempt_token",
+                    models.UUIDField(blank=True, null=True),
+                ),
+                (
+                    "delivery_started_at",
+                    models.DateTimeField(blank=True, null=True),
+                ),
+                (
+                    "integration",
+                    models.ForeignKey(
+                        on_delete=django.db.models.deletion.CASCADE,
+                        related_name="jira_issues",
+                        to="api.integration",
+                    ),
+                ),
+                (
+                    "provider",
+                    models.ForeignKey(
+                        on_delete=django.db.models.deletion.CASCADE,
+                        related_name="jira_issues",
+                        to="api.provider",
+                    ),
+                ),
+                (
+                    "tenant",
+                    models.ForeignKey(
+                        on_delete=django.db.models.deletion.CASCADE, to="api.tenant"
+                    ),
+                ),
+            ],
+            options={
+                "db_table": "jira_issues",
+                "abstract": False,
+            },
+        ),
+        migrations.AddConstraint(
+            model_name="jiraissue",
+            constraint=models.UniqueConstraint(
+                fields=("tenant_id", "integration_id", "provider_id", "finding_uid"),
+                name="unique_jira_issue_per_finding",
+            ),
+        ),
+        migrations.AddConstraint(
+            model_name="jiraissue",
+            constraint=models.UniqueConstraint(
+                condition=models.Q(("delivery_attempt_token__isnull", False)),
+                fields=("delivery_attempt_token",),
+                name="unique_jira_delivery_attempt",
+            ),
+        ),
+        migrations.AddConstraint(
+            model_name="jiraissue",
+            constraint=models.CheckConstraint(
+                condition=models.Q(
+                    ("delivery_started_at__isnull", True),
+                    ("delivery_attempt_token__isnull", False),
+                    _connector="OR",
+                ),
+                name="jira_delivery_started_requires_token",
+            ),
+        ),
+        migrations.AddConstraint(
+            model_name="jiraissue",
+            constraint=models.CheckConstraint(
+                condition=models.Q(
+                    models.Q(
+                        ("issue_id__isnull", True),
+                        ("issue_key__isnull", True),
+                        ("issue_url__isnull", True),
+                        ("project_key__isnull", True),
+                    ),
+                    models.Q(
+                        models.Q(
+                            ("issue_id__isnull", False),
+                            ("issue_key__isnull", False),
+                            ("issue_url__isnull", False),
+                            ("project_key__isnull", False),
+                        ),
+                        models.Q(("issue_id", ""), _negated=True),
+                        models.Q(("issue_key", ""), _negated=True),
+                        models.Q(("issue_url", ""), _negated=True),
+                        models.Q(("project_key", ""), _negated=True),
+                    ),
+                    _connector="OR",
+                ),
+                name="jira_issue_link_all_or_none",
+            ),
+        ),
+        migrations.AddConstraint(
+            model_name="jiraissue",
+            constraint=api.rls.RowLevelSecurityConstraint(
+                "tenant_id",
+                name="rls_on_jiraissue",
+                statements=["SELECT", "INSERT", "UPDATE", "DELETE"],
+            ),
+        ),
+    ]

api/src/backend/api/migrations/0099_jira_issues_indexes.py

@@ -0,0 +1,17 @@
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("api", "0098_jira_issues"),
+    ]
+
+    operations = [
+        migrations.AddIndex(
+            model_name="jiraissue",
+            index=models.Index(
+                fields=["tenant_id", "provider_id", "finding_uid", "integration_id"],
+                name="ji_tenant_prov_uid_int_idx",
+            ),
+        ),
+    ]

api/src/backend/api/models.py

@@ -3113,3 +3113,112 @@ class TenantComplianceSummary(RowLevelSecurityProtectedModel):
                 statements=["SELECT", "INSERT", "UPDATE", "DELETE"],
             ),
         ]
+
+
+class JiraIssue(RowLevelSecurityProtectedModel):
+    """Current Jira issue linked to one finding and Jira integration.
+
+    One row per (integration, provider, finding uid). Keyed on the finding ``uid``
+    rather than the per-scan finding id so the link survives rescans. The current
+    link stays populated while a replacement attempt is in progress.
+    """
+
+    class StatusCategoryChoices(models.TextChoices):
+        NEW = "new", _("New")
+        INDETERMINATE = "indeterminate", _("In progress")
+        DONE = "done", _("Done")
+
+    id = models.UUIDField(primary_key=True, default=uuid4, editable=False)
+    inserted_at = models.DateTimeField(auto_now_add=True, editable=False)
+    updated_at = models.DateTimeField(auto_now=True, editable=False)
+    integration = models.ForeignKey(
+        Integration, on_delete=models.CASCADE, related_name="jira_issues"
+    )
+    provider = models.ForeignKey(
+        Provider, on_delete=models.CASCADE, related_name="jira_issues"
+    )
+    finding_uid = models.CharField(max_length=300)
+    # Findings are partitioned and rotate per scan, so this is not a foreign key.
+    finding_id = models.UUIDField()
+    issue_id = models.CharField(max_length=64, null=True, blank=True)
+    issue_key = models.CharField(max_length=64, null=True, blank=True)
+    issue_url = models.URLField(max_length=2048, null=True, blank=True)
+    project_key = models.CharField(max_length=64, null=True, blank=True)
+    issue_status = models.CharField(max_length=64, null=True, blank=True)
+    issue_status_category = models.CharField(
+        max_length=16,
+        choices=StatusCategoryChoices.choices,
+        null=True,
+        blank=True,
+    )
+    status_synced_at = models.DateTimeField(null=True, blank=True)
+    delivery_attempt_token = models.UUIDField(null=True, blank=True)
+    delivery_started_at = models.DateTimeField(null=True, blank=True)
+
+    class Meta(RowLevelSecurityProtectedModel.Meta):
+        db_table = "jira_issues"
+
+        constraints = [
+            models.UniqueConstraint(
+                fields=("tenant_id", "integration_id", "provider_id", "finding_uid"),
+                name="unique_jira_issue_per_finding",
+            ),
+            models.UniqueConstraint(
+                fields=("delivery_attempt_token",),
+                condition=Q(delivery_attempt_token__isnull=False),
+                name="unique_jira_delivery_attempt",
+            ),
+            models.CheckConstraint(
+                condition=(
+                    Q(delivery_started_at__isnull=True)
+                    | Q(delivery_attempt_token__isnull=False)
+                ),
+                name="jira_delivery_started_requires_token",
+            ),
+            models.CheckConstraint(
+                condition=(
+                    Q(
+                        issue_id__isnull=True,
+                        issue_key__isnull=True,
+                        issue_url__isnull=True,
+                        project_key__isnull=True,
+                    )
+                    | (
+                        Q(
+                            issue_id__isnull=False,
+                            issue_key__isnull=False,
+                            issue_url__isnull=False,
+                            project_key__isnull=False,
+                        )
+                        & ~Q(issue_id="")
+                        & ~Q(issue_key="")
+                        & ~Q(issue_url="")
+                        & ~Q(project_key="")
+                    )
+                ),
+                name="jira_issue_link_all_or_none",
+            ),
+            RowLevelSecurityConstraint(
+                field="tenant_id",
+                name="rls_on_%(class)s",
+                statements=["SELECT", "INSERT", "UPDATE", "DELETE"],
+            ),
+        ]
+        indexes = [
+            models.Index(
+                fields=["tenant_id", "provider_id", "finding_uid", "integration_id"],
+                name="ji_tenant_prov_uid_int_idx",
+            ),
+        ]
+
+    class JSONAPIMeta:
+        resource_name = "jira-issues"
+
+    @property
+    def is_linked(self) -> bool:
+        """Whether the row points at a confirmed Jira issue (not a reservation)."""
+        return self.issue_id is not None
+
+    @property
+    def is_done(self) -> bool:
+        return self.issue_status_category == self.StatusCategoryChoices.DONE

api/src/backend/api/tests/test_models.py

@@ -1,9 +1,12 @@
 from datetime import UTC, datetime
+from uuid import uuid4
 
 import pytest
 from allauth.socialaccount.models import SocialApp
 from api.db_router import MainRouter
+from api.db_utils import rls_transaction
 from api.models import (
+    JiraIssue,
     ProviderComplianceScore,
     Resource,
     ResourceTag,
@@ -14,7 +17,7 @@ from api.models import (
     TenantComplianceSummary,
 )
 from django.core.exceptions import ValidationError
-from django.db import IntegrityError
+from django.db import IntegrityError, transaction
 
 
 @pytest.mark.django_db
@@ -524,3 +527,133 @@ class TestTenantComplianceSummaryModel:
 
         assert summary1.id != summary2.id
         assert summary1.requirements_passed != summary2.requirements_passed
+
+
+@pytest.mark.django_db
+class TestJiraIssueModel:
+    def test_create_jira_issue(
+        self, jira_integration_fixture, aws_provider, findings_fixture
+    ):
+        finding = findings_fixture[0]
+        with rls_transaction(str(jira_integration_fixture.tenant_id)):
+            issue = JiraIssue.objects.create(
+                tenant_id=jira_integration_fixture.tenant_id,
+                integration=jira_integration_fixture,
+                provider=aws_provider,
+                finding_uid=finding.uid,
+                finding_id=finding.id,
+                issue_id="10001",
+                issue_key="TEST-1",
+                issue_url="https://test.atlassian.net/browse/TEST-1",
+                project_key="TEST",
+            )
+        assert issue.is_linked
+        assert not issue.is_done
+        assert issue.issue_status_category is None
+
+    def test_reservation_is_not_linked(
+        self, jira_integration_fixture, aws_provider, findings_fixture
+    ):
+        finding = findings_fixture[0]
+        with rls_transaction(str(jira_integration_fixture.tenant_id)):
+            issue = JiraIssue.objects.create(
+                tenant_id=jira_integration_fixture.tenant_id,
+                integration=jira_integration_fixture,
+                provider=aws_provider,
+                finding_uid=finding.uid,
+                finding_id=finding.id,
+                delivery_attempt_token=uuid4(),
+            )
+        assert not issue.is_linked
+
+    def test_delivery_started_at_requires_attempt_token(
+        self, jira_integration_fixture, aws_provider, findings_fixture
+    ):
+        finding = findings_fixture[0]
+        with rls_transaction(str(jira_integration_fixture.tenant_id)):
+            with pytest.raises(IntegrityError), transaction.atomic():
+                JiraIssue.objects.create(
+                    tenant_id=jira_integration_fixture.tenant_id,
+                    integration=jira_integration_fixture,
+                    provider=aws_provider,
+                    finding_uid=finding.uid,
+                    finding_id=finding.id,
+                    delivery_started_at=datetime.now(UTC),
+                )
+
+    def test_unique_per_integration_provider_and_finding_uid(
+        self, jira_integration_fixture, aws_provider_pair, findings_fixture
+    ):
+        provider, provider2 = aws_provider_pair
+        finding = findings_fixture[0]
+        common = {
+            "tenant_id": jira_integration_fixture.tenant_id,
+            "integration": jira_integration_fixture,
+            "finding_uid": finding.uid,
+            "finding_id": finding.id,
+            "project_key": "TEST",
+        }
+        with rls_transaction(str(jira_integration_fixture.tenant_id)):
+            JiraIssue.objects.create(
+                provider=provider,
+                issue_id="10001",
+                issue_key="TEST-1",
+                issue_url="https://test.atlassian.net/browse/TEST-1",
+                **common,
+            )
+            # Same finding uid on another provider is a different finding
+            JiraIssue.objects.create(
+                provider=provider2,
+                issue_id="10002",
+                issue_key="TEST-2",
+                issue_url="https://test.atlassian.net/browse/TEST-2",
+                **common,
+            )
+            with pytest.raises(IntegrityError), transaction.atomic():
+                JiraIssue.objects.create(
+                    provider=provider,
+                    issue_id="10003",
+                    issue_key="TEST-3",
+                    issue_url="https://test.atlassian.net/browse/TEST-3",
+                    **common,
+                )
+
+    def test_delivery_attempt_token_is_unique_when_present(
+        self, jira_integration_fixture, aws_provider_pair, findings_fixture
+    ):
+        provider, provider2 = aws_provider_pair
+        attempt_token = uuid4()
+        common = {
+            "tenant_id": jira_integration_fixture.tenant_id,
+            "integration": jira_integration_fixture,
+            "finding_id": findings_fixture[0].id,
+            "delivery_attempt_token": attempt_token,
+        }
+        with rls_transaction(str(jira_integration_fixture.tenant_id)):
+            JiraIssue.objects.create(
+                provider=provider,
+                finding_uid="finding-one",
+                **common,
+            )
+
+            with pytest.raises(IntegrityError), transaction.atomic():
+                JiraIssue.objects.create(
+                    provider=provider2,
+                    finding_uid="finding-two",
+                    **common,
+                )
+
+    def test_link_fields_are_all_or_none(
+        self, jira_integration_fixture, aws_provider, findings_fixture
+    ):
+        finding = findings_fixture[0]
+        with rls_transaction(str(jira_integration_fixture.tenant_id)):
+            with pytest.raises(IntegrityError), transaction.atomic():
+                JiraIssue.objects.create(
+                    tenant_id=jira_integration_fixture.tenant_id,
+                    integration=jira_integration_fixture,
+                    provider=aws_provider,
+                    finding_uid=finding.uid,
+                    finding_id=finding.id,
+                    issue_id="10001",
+                )

api/src/backend/api/tests/test_rbac.py

@@ -1591,6 +1591,52 @@ class TestLimitedVisibility:
 
         assert response.status_code == status.HTTP_204_NO_CONTENT
 
+    def test_jira_issues_limited_to_visible_providers(
+        self, authenticated_client_rbac_limited, jira_issues_fixture
+    ):
+        linked, other_provider_issue, _ = jira_issues_fixture
+        response = authenticated_client_rbac_limited.get(reverse("jiraissue-list"))
+        assert response.status_code == status.HTTP_200_OK
+        assert [item["id"] for item in response.json()["data"]] == [str(linked.id)]
+
+        response = authenticated_client_rbac_limited.get(
+            reverse("jiraissue-detail", kwargs={"pk": other_provider_issue.id})
+        )
+        assert response.status_code == status.HTTP_404_NOT_FOUND
+
+    def test_jira_task_result_hides_issue_metadata_for_limited_role(
+        self,
+        authenticated_client_rbac_limited,
+        jira_issues_fixture,
+        tasks_fixture,
+    ):
+        _, hidden_issue, _ = jira_issues_fixture
+        task, *_ = tasks_fixture
+        task.task_runner_task.task_name = "integration-jira"
+        task.task_runner_task.result = json.dumps(
+            {
+                "skipped_count": 1,
+                "skipped": [
+                    {
+                        "finding_id": str(hidden_issue.finding_id),
+                        "issue_key": hidden_issue.issue_key,
+                        "issue_url": hidden_issue.issue_url,
+                        "issue_status": hidden_issue.issue_status,
+                    }
+                ],
+            }
+        )
+        task.task_runner_task.save(update_fields=["task_name", "result"])
+
+        response = authenticated_client_rbac_limited.get(
+            reverse("task-detail", kwargs={"pk": task.id})
+        )
+
+        assert response.status_code == status.HTTP_200_OK
+        assert response.json()["data"]["attributes"]["result"]["skipped"] == [
+            {"finding_id": str(hidden_issue.finding_id)}
+        ]
+
     def test_jira_issue_types_allowed_without_unlimited_visibility(
         self, authenticated_client_rbac_limited, jira_integration_fixture
     ):

api/src/backend/api/tests/test_views.py

@@ -34,6 +34,7 @@ from api.models import (
     Integration,
     Invitation,
     InvitationRoleRelationship,
+    JiraIssue,
     LighthouseProviderConfiguration,
     LighthouseProviderModels,
     LighthouseTenantConfiguration,
@@ -5128,6 +5129,40 @@ class TestTaskViewSet:
             "label": "True North",
         }
 
+    def test_tasks_retrieve_sanitizes_jira_result(
+        self, authenticated_client, tasks_fixture
+    ):
+        task, *_ = tasks_fixture
+        task.task_runner_task.task_name = "integration-jira"
+        task.task_runner_task.result = json.dumps(
+            {
+                "created_count": 0,
+                "failed_count": 0,
+                "skipped_count": 1,
+                "skipped": [
+                    {
+                        "finding_id": "00000000-0000-0000-0000-000000000001",
+                        "issue_key": "PRIVATE-1",
+                        "issue_url": "https://private.example/browse/PRIVATE-1",
+                        "issue_status": "In Progress",
+                    }
+                ],
+            }
+        )
+        task.task_runner_task.save(update_fields=["task_name", "result"])
+
+        response = authenticated_client.get(
+            reverse("task-detail", kwargs={"pk": task.id}),
+        )
+
+        assert response.status_code == status.HTTP_200_OK
+        assert response.json()["data"]["attributes"]["result"] == {
+            "created_count": 0,
+            "failed_count": 0,
+            "skipped_count": 1,
+            "skipped": [{"finding_id": "00000000-0000-0000-0000-000000000001"}],
+        }
+
     def test_tasks_retrieve_with_truncated_kwargs_returns_empty_task_args(
         self, authenticated_client, tasks_fixture
     ):
@@ -13557,8 +13592,234 @@ class TestScheduleViewSet:
         assert response.status_code == status.HTTP_409_CONFLICT
 
 
+@pytest.mark.django_db
+class TestJiraIssueViewSet:
+    def test_list_hides_reservations(self, authenticated_client, jira_issues_fixture):
+        linked, other_provider_issue, reservation = jira_issues_fixture
+        response = authenticated_client.get(reverse("jiraissue-list"))
+        assert response.status_code == status.HTTP_200_OK
+        ids = {item["id"] for item in response.json()["data"]}
+        assert ids == {str(linked.id), str(other_provider_issue.id)}
+        assert str(reservation.id) not in ids
+
+    def test_retrieve(self, authenticated_client, jira_issues_fixture):
+        linked, *_ = jira_issues_fixture
+        response = authenticated_client.get(
+            reverse("jiraissue-detail", kwargs={"pk": linked.id})
+        )
+        assert response.status_code == status.HTTP_200_OK
+        data = response.json()["data"]
+        assert data["type"] == "jira-issues"
+        attributes = data["attributes"]
+        assert attributes["finding_uid"] == linked.finding_uid
+        assert attributes["finding_id"] == str(linked.finding_id)
+        assert attributes["issue_key"] == "TEST-1"
+        assert attributes["issue_url"] == "https://test.atlassian.net/browse/TEST-1"
+        assert attributes["project_key"] == "TEST"
+        assert attributes["issue_status"] == "To Do"
+        assert attributes["issue_status_category"] == "new"
+        assert attributes["status_synced_at"] is not None
+        assert "delivery_attempt_token" not in attributes
+        assert "delivery_started_at" not in attributes
+        relationships = data["relationships"]
+        assert relationships["provider"]["data"]["id"] == str(linked.provider_id)
+        assert relationships["integration"]["data"]["id"] == str(linked.integration_id)
+
+    def test_retrieve_reservation_returns_404(
+        self, authenticated_client, jira_issues_fixture
+    ):
+        *_, reservation = jira_issues_fixture
+        response = authenticated_client.get(
+            reverse("jiraissue-detail", kwargs={"pk": reservation.id})
+        )
+        assert response.status_code == status.HTTP_404_NOT_FOUND
+
+    def test_other_tenant_issue_is_hidden(
+        self, authenticated_client, jira_issues_fixture, tenants_fixture
+    ):
+        linked, *_ = jira_issues_fixture
+        other_tenant = tenants_fixture[2]
+        with rls_transaction(str(other_tenant.id)):
+            other_provider = Provider.objects.create(
+                tenant_id=other_tenant.id,
+                provider=Provider.ProviderChoices.AWS,
+                uid="999999999999",
+                alias="other-tenant-provider",
+            )
+            other_integration = Integration.objects.create(
+                tenant_id=other_tenant.id,
+                enabled=True,
+                connected=True,
+                integration_type=Integration.IntegrationChoices.JIRA,
+                configuration={"projects": {"OTHER": "Other project"}},
+                credentials={
+                    "domain": "other-tenant",
+                    "user_mail": "other-tenant@example.com",
+                    "api_token": "fake-token",
+                },
+            )
+            other_issue = JiraIssue.objects.create(
+                tenant_id=other_tenant.id,
+                integration=other_integration,
+                provider=other_provider,
+                finding_uid="other-tenant-finding",
+                finding_id=datetime_to_uuid7(datetime.now(UTC)),
+                issue_id="20001",
+                issue_key="OTHER-1",
+                issue_url="https://other-tenant.atlassian.net/browse/OTHER-1",
+                project_key="OTHER",
+            )
+
+        response = authenticated_client.get(reverse("jiraissue-list"))
+        assert response.status_code == status.HTTP_200_OK
+        ids = {item["id"] for item in response.json()["data"]}
+        assert str(linked.id) in ids
+        assert str(other_issue.id) not in ids
+
+        response = authenticated_client.get(
+            reverse("jiraissue-detail", kwargs={"pk": other_issue.id})
+        )
+        assert response.status_code == status.HTTP_404_NOT_FOUND
+
+    @pytest.mark.parametrize(
+        "filter_name, filter_value, expected_keys",
+        [
+            ("finding_uid", "test_finding_uid_1", {"TEST-1"}),
+            (
+                "finding_uid__in",
+                "test_finding_uid_1,test_finding_uid_other_provider",
+                {"TEST-1", "TEST-2"},
+            ),
+            ("finding_uid__in", "does-not-exist", set()),
+            ("issue_key", "TEST-2", {"TEST-2"}),
+            ("issue_status_category", "done", {"TEST-2"}),
+            ("issue_status_category__in", "new,indeterminate", {"TEST-1"}),
+            ("project_key", "TEST", {"TEST-1", "TEST-2"}),
+            ("search", "TEST-1", {"TEST-1"}),
+        ],
+    )
+    def test_filters(
+        self,
+        authenticated_client,
+        jira_issues_fixture,
+        filter_name,
+        filter_value,
+        expected_keys,
+    ):
+        response = authenticated_client.get(
+            reverse("jiraissue-list"), {f"filter[{filter_name}]": filter_value}
+        )
+        assert response.status_code == status.HTTP_200_OK
+        keys = {item["attributes"]["issue_key"] for item in response.json()["data"]}
+        assert keys == expected_keys
+
+    def test_filter_by_provider(self, authenticated_client, jira_issues_fixture):
+        linked, other_provider_issue, _ = jira_issues_fixture
+        response = authenticated_client.get(
+            reverse("jiraissue-list"),
+            {"filter[provider_id]": str(other_provider_issue.provider_id)},
+        )
+        assert response.status_code == status.HTTP_200_OK
+        assert [item["id"] for item in response.json()["data"]] == [
+            str(other_provider_issue.id)
+        ]
+
+    def test_filter_by_integration_and_finding_id(
+        self, authenticated_client, jira_issues_fixture
+    ):
+        linked, *_ = jira_issues_fixture
+        response = authenticated_client.get(
+            reverse("jiraissue-list"),
+            {
+                "filter[integration]": str(linked.integration_id),
+                "filter[finding_id]": str(linked.finding_id),
+            },
+        )
+        assert response.status_code == status.HTTP_200_OK
+        assert [item["id"] for item in response.json()["data"]] == [str(linked.id)]
+
+    def test_invalid_filter(self, authenticated_client, jira_issues_fixture):
+        response = authenticated_client.get(
+            reverse("jiraissue-list"), {"filter[invalid]": "x"}
+        )
+        assert response.status_code == status.HTTP_400_BAD_REQUEST
+
+    def test_include_provider(self, authenticated_client, jira_issues_fixture):
+        response = authenticated_client.get(
+            reverse("jiraissue-list"), {"include": "provider"}
+        )
+        assert response.status_code == status.HTTP_200_OK
+        included_types = {item["type"] for item in response.json()["included"]}
+        assert included_types == {"providers"}
+
+    def test_read_only(self, authenticated_client, jira_issues_fixture):
+        linked, *_ = jira_issues_fixture
+        response = authenticated_client.post(
+            reverse("jiraissue-list"),
+            data=json.dumps({"data": {"type": "jira-issues", "attributes": {}}}),
+            content_type="application/vnd.api+json",
+        )
+        assert response.status_code == status.HTTP_405_METHOD_NOT_ALLOWED
+        response = authenticated_client.delete(
+            reverse("jiraissue-detail", kwargs={"pk": linked.id})
+        )
+        assert response.status_code == status.HTTP_405_METHOD_NOT_ALLOWED
+
+
 @pytest.mark.django_db
 class TestIntegrationViewSet:
+    @pytest.mark.parametrize(
+        ("force_retry_attribute", "expected_force_retry"),
+        (({}, False), ({"force_retry": True}, True)),
+    )
+    @patch("api.v1.views.Task.objects.get")
+    @patch("api.v1.views.jira_integration_task.delay")
+    def test_jira_dispatch_passes_force_retry_to_task(
+        self,
+        mock_jira_task,
+        mock_task_get,
+        force_retry_attribute,
+        expected_force_retry,
+        authenticated_client,
+        jira_integration_fixture,
+        findings_fixture,
+        tasks_fixture,
+    ):
+        finding, _ = findings_fixture
+        prowler_task = tasks_fixture[0]
+        mock_jira_task.return_value.id = prowler_task.id
+        mock_task_get.return_value = prowler_task
+        data = {
+            "data": {
+                "type": "integrations-jira-dispatches",
+                "attributes": {
+                    "project_key": "TEST",
+                    "issue_type": "Task",
+                    **force_retry_attribute,
+                },
+            }
+        }
+
+        response = authenticated_client.post(
+            reverse(
+                "integration-jira-dispatches",
+                kwargs={"integration_pk": jira_integration_fixture.id},
+            )
+            + f"?filter[finding_id]={finding.id}",
+            data=json.dumps(data),
+            content_type=API_JSON_CONTENT_TYPE,
+        )
+
+        assert response.status_code == status.HTTP_202_ACCEPTED
+        mock_jira_task.assert_called_once_with(
+            tenant_id=str(jira_integration_fixture.tenant_id),
+            integration_id=str(jira_integration_fixture.id),
+            project_key="TEST",
+            issue_type="Task",
+            finding_ids=[str(finding.id)],
+            force_retry=expected_force_retry,
+        )
+
     def test_integrations_list(self, authenticated_client, integrations_fixture):
         response = authenticated_client.get(reverse("integration-list"))
         assert response.status_code == status.HTTP_200_OK

api/src/backend/api/v1/serializer_utils/integrations.py

@@ -1,5 +1,6 @@
 import os
 import re
+from typing import Any
 
 from api.models import Integration, IntegrationProviderRelationship, Provider
 from api.v1.serializer_utils.base import BaseValidateSerializer
@@ -12,6 +13,24 @@ ATLASSIAN_SITE_NAME_REGEX = re.compile(
 )
 
 
+def sanitize_integration_task_result(task_name: str | None, result: Any) -> Any:
+    """Remove private integration metadata from task results."""
+    if task_name != "integration-jira" or not isinstance(result, dict):
+        return result
+
+    skipped = result.get("skipped")
+    if not isinstance(skipped, list):
+        return result
+
+    sanitized_result = result.copy()
+    sanitized_result["skipped"] = [
+        {"finding_id": entry["finding_id"]}
+        for entry in skipped
+        if isinstance(entry, dict) and "finding_id" in entry
+    ]
+    return sanitized_result
+
+
 def replace_integration_providers(
     integration: Integration, providers: list[Provider], tenant_id: str
 ) -> None:

api/src/backend/api/v1/serializers.py

@@ -14,6 +14,7 @@ from api.models import (
     IntegrationProviderRelationship,
     Invitation,
     InvitationRoleRelationship,
+    JiraIssue,
     LighthouseConfiguration,
     LighthouseProviderConfiguration,
     LighthouseProviderModels,
@@ -50,6 +51,7 @@ from api.v1.serializer_utils.integrations import (
     S3ConfigSerializer,
     SecurityHubConfigSerializer,
     replace_integration_providers,
+    sanitize_integration_task_result,
 )
 from api.v1.serializer_utils.lighthouse import (
     BedrockCredentialsSerializer,
@@ -637,7 +639,9 @@ class TaskSerializer(RLSSerializer, TaskBase):
 
     @extend_schema_field(serializers.JSONField())
     def get_result(self, obj):
-        return self.get_json_field(obj, "result")
+        result = self.get_json_field(obj, "result")
+        task_name = obj.task_runner_task.task_name if obj.task_runner_task else None
+        return sanitize_integration_task_result(task_name, result)
 
     @extend_schema_field(serializers.JSONField())
     def get_task_args(self, obj):
@@ -3217,6 +3221,11 @@ class IntegrationJiraDispatchSerializer(BaseSerializerV1):
 
     project_key = serializers.CharField(required=True)
     issue_type = serializers.CharField(required=True)
+    force_retry = serializers.BooleanField(
+        required=False,
+        default=False,
+        help_text="Retry a stale unresolved delivery after Jira returns zero marker matches. This can create a duplicate issue.",
+    )
 
     class JSONAPIMeta:
         resource_name = "integrations-jira-dispatches"
@@ -4149,6 +4158,41 @@ class LighthouseProviderModelsUpdateSerializer(BaseWriteSerializer):
 # Mute Rules
 
 
+class JiraIssueSerializer(RLSSerializer):
+    """
+    Read-only view of a Jira issue linked to a finding by a Jira integration.
+
+    Rows are keyed on the finding ``uid`` so the same finding maps to the same
+    issue across scans. ``issue_status`` is the last status Prowler observed in
+    Jira (refreshed whenever a dispatch touches the finding), not a live value.
+    """
+
+    class Meta:
+        model = JiraIssue
+        fields = [
+            "id",
+            "inserted_at",
+            "updated_at",
+            "finding_uid",
+            "finding_id",
+            "issue_key",
+            "issue_id",
+            "issue_url",
+            "project_key",
+            "issue_status",
+            "issue_status_category",
+            "status_synced_at",
+            "integration",
+            "provider",
+            "url",
+        ]
+        read_only_fields = fields
+
+    included_serializers = {
+        "provider": "api.v1.serializers.ProviderIncludeSerializer",
+    }
+
+
 class MuteRuleSerializer(RLSSerializer):
     """
     Serializer for reading MuteRule instances.

api/src/backend/api/v1/urls.py

@@ -14,6 +14,7 @@ from api.v1.views import (
     IntegrationViewSet,
     InvitationAcceptViewSet,
     InvitationViewSet,
+    JiraIssueViewSet,
     LighthouseConfigViewSet,
     LighthouseProviderConfigViewSet,
     LighthouseProviderModelsViewSet,
@@ -107,6 +108,7 @@ router.register(
     basename="lighthouse-models",
 )
 router.register(r"mute-rules", MuteRuleViewSet, basename="mute-rule")
+router.register(r"jira-issues", JiraIssueViewSet, basename="jiraissue")
 
 tenants_router = routers.NestedSimpleRouter(router, r"tenants", lookup="tenant")
 tenants_router.register(

api/src/backend/api/v1/views.py

@@ -54,6 +54,7 @@ from api.filters import (
     IntegrationFilter,
     IntegrationJiraFindingsFilter,
     InvitationFilter,
+    JiraIssueFilter,
     LatestFindingFilter,
     LatestFindingGroupFilter,
     LatestFindingGroupSummaryFilter,
@@ -89,6 +90,7 @@ from api.models import (
     Integration,
     Invitation,
     InvitationRoleRelationship,
+    JiraIssue,
     LighthouseConfiguration,
     LighthouseProviderConfiguration,
     LighthouseProviderModels,
@@ -179,6 +181,7 @@ from api.v1.serializers import (
     InvitationCreateSerializer,
     InvitationSerializer,
     InvitationUpdateSerializer,
+    JiraIssueSerializer,
     LighthouseConfigCreateSerializer,
     LighthouseConfigSerializer,
     LighthouseConfigUpdateSerializer,
@@ -7001,6 +7004,7 @@ class IntegrationJiraViewSet(BaseRLSViewSet):
 
         project_key = serializer.validated_data["project_key"]
         issue_type = serializer.validated_data["issue_type"]
+        force_retry = serializer.validated_data["force_retry"]
 
         with transaction.atomic():
             task = jira_integration_task.delay(
@@ -7009,6 +7013,7 @@ class IntegrationJiraViewSet(BaseRLSViewSet):
                 project_key=project_key,
                 issue_type=issue_type,
                 finding_ids=finding_ids,
+                force_retry=force_retry,
             )
         prowler_task = Task.objects.get(id=task.id)
         serializer = TaskSerializer(prowler_task)
@@ -7486,6 +7491,56 @@ class TenantApiKeyViewSet(BaseRLSViewSet):
         return Response(data=serializer.data, status=status.HTTP_200_OK)
 
 
+# Jira issues
+@extend_schema_view(
+    list=extend_schema(
+        tags=["Integration"],
+        summary="List Jira issues linked to findings",
+        description=(
+            "Retrieve the Jira issues created from findings through Jira integrations. "
+            "Each entry links a finding UID to the latest Jira issue created for it, "
+            "with the last status observed in Jira. Use `filter[finding_uid__in]` "
+            "and `filter[provider_id]` to check whether specific findings already "
+            "have a ticket."
+        ),
+    ),
+    retrieve=extend_schema(
+        tags=["Integration"],
+        summary="Retrieve a Jira issue link",
+        description="Fetch the Jira issue linked to a finding by the link ID.",
+    ),
+)
+class JiraIssueViewSet(BaseRLSViewSet):
+    queryset = JiraIssue.objects.all()
+    serializer_class = JiraIssueSerializer
+    filterset_class = JiraIssueFilter
+    http_method_names = ["get"]
+    search_fields = ["finding_uid", "issue_key"]
+    ordering = ["-inserted_at"]
+    ordering_fields = [
+        "inserted_at",
+        "updated_at",
+        "issue_key",
+        "project_key",
+        "issue_status",
+        "status_synced_at",
+    ]
+    # RBAC required permissions (implicit -> MANAGE_PROVIDERS enables unlimited
+    # visibility or check visibility via provider group, like findings)
+    required_permissions = []
+
+    def get_queryset(self):
+        if getattr(self, "swagger_fake_view", False):
+            return JiraIssue.objects.none()
+        # Pending reservations have no confirmed Jira issue and are private.
+        queryset = JiraIssue.objects.filter(
+            tenant_id=self.request.tenant_id, issue_id__isnull=False
+        )
+        if not self.user_role.unlimited_visibility:
+            queryset = queryset.filter(provider__in=get_providers(self.user_role))
+        return queryset.select_related("provider", "integration")
+
+
 # MuteRules
 @extend_schema_view(
     list=extend_schema(

api/src/backend/conftest.py

@@ -20,6 +20,7 @@ from api.models import (
     Integration,
     IntegrationProviderRelationship,
     Invitation,
+    JiraIssue,
     LighthouseConfiguration,
     Membership,
     MuteRule,
@@ -1470,6 +1471,52 @@ def jira_integration_fixture(tenants_fixture):
         )
 
 
+@pytest.fixture
+def jira_issues_fixture(jira_integration_fixture, aws_provider_pair, findings_fixture):
+    """Two linked issues (one per provider) and one in-flight reservation."""
+    provider, provider2 = aws_provider_pair
+    finding1, finding2 = findings_fixture
+    tenant_id = jira_integration_fixture.tenant_id
+    with rls_transaction(str(tenant_id)):
+        linked = JiraIssue.objects.create(
+            tenant_id=tenant_id,
+            integration=jira_integration_fixture,
+            provider=provider,
+            finding_uid=finding1.uid,
+            finding_id=finding1.id,
+            issue_key="TEST-1",
+            issue_id="10001",
+            issue_url="https://test.atlassian.net/browse/TEST-1",
+            project_key="TEST",
+            issue_status="To Do",
+            issue_status_category=JiraIssue.StatusCategoryChoices.NEW,
+            status_synced_at=datetime.now(UTC),
+        )
+        hidden_provider_issue = JiraIssue.objects.create(
+            tenant_id=tenant_id,
+            integration=jira_integration_fixture,
+            provider=provider2,
+            finding_uid="test_finding_uid_other_provider",
+            finding_id=finding2.id,
+            issue_key="TEST-2",
+            issue_id="10002",
+            issue_url="https://test.atlassian.net/browse/TEST-2",
+            project_key="TEST",
+            issue_status="Done",
+            issue_status_category=JiraIssue.StatusCategoryChoices.DONE,
+            status_synced_at=datetime.now(UTC),
+        )
+        reservation = JiraIssue.objects.create(
+            tenant_id=tenant_id,
+            integration=jira_integration_fixture,
+            provider=provider,
+            finding_uid=finding2.uid,
+            finding_id=finding2.id,
+            delivery_attempt_token=uuid4(),
+        )
+    return linked, hidden_provider_issue, reservation
+
+
 @pytest.fixture
 def backfill_scan_metadata_fixture(scans_fixture, findings_fixture):
     for scan_instance in scans_fixture:

api/src/backend/tasks/jobs/deletion.py

@@ -5,6 +5,7 @@ from api.db_utils import batch_delete, rls_transaction
 from api.models import (
     AttackPathsScan,
     Finding,
+    JiraIssue,
     Provider,
     ProviderComplianceScore,
     Resource,
@@ -86,6 +87,7 @@ def delete_provider(tenant_id: str, pk: str):
 
         deletion_steps = [
             ("Scan Summaries", ScanSummary.all_objects.filter(scan__provider=instance)),
+            ("Jira Issues", JiraIssue.objects.filter(provider=instance)),
             ("Findings", Finding.all_objects.filter(scan__provider=instance)),
             ("Resources", Resource.all_objects.filter(provider=instance)),
             ("Scans", Scan.all_objects.filter(provider=instance)),

api/src/backend/tasks/jobs/integrations.py

@@ -1,25 +1,37 @@
 import os
 import time
-from datetime import UTC, datetime
+from datetime import UTC, datetime, timedelta
+from enum import Enum
 from glob import glob
 from urllib.parse import quote
+from uuid import uuid4
 
 from api.db_router import READ_REPLICA_ALIAS, MainRouter
 from api.db_utils import REPLICA_MAX_ATTEMPTS, REPLICA_RETRY_BASE_DELAY, rls_transaction
-from api.models import Finding, Integration, Provider
+from api.models import Finding, Integration, JiraIssue, Provider
 from api.rls import Tenant
 from api.utils import initialize_prowler_integration, initialize_prowler_provider
 from celery.utils.log import get_task_logger
 from config.django.base import DJANGO_FINDINGS_BATCH_SIZE
 from django.conf import settings
-from django.db import OperationalError
+from django.db import IntegrityError, OperationalError
+from django.utils import timezone
 from prowler.lib.outputs.asff.asff import ASFF
 from prowler.lib.outputs.compliance.generic.generic import GenericCompliance
 from prowler.lib.outputs.csv.csv import CSV
 from prowler.lib.outputs.finding import Finding as FindingOutput
 from prowler.lib.outputs.html.html import HTML
-from prowler.lib.outputs.jira.exceptions.exceptions import JiraBaseException
 from prowler.lib.outputs.jira.jira import Jira
+from prowler.lib.outputs.jira.models import (
+    JiraCreationOutcome,
+    JiraCreationResult,
+    JiraIssueLookupOutcome,
+    JiraIssueReference,
+    JiraIssueSearchMatch,
+    JiraIssueSearchOutcome,
+    JiraIssueSearchResult,
+    JiraIssueStatusResult,
+)
 from prowler.lib.outputs.ocsf.ocsf import OCSF
 from prowler.providers.aws.aws_provider import AwsProvider
 from prowler.providers.aws.lib.s3.s3 import S3
@@ -530,114 +542,734 @@ def get_tenant_name(tenant_id: str) -> str:
         return ""
 
 
+# Findings are pre-checked in bounded index lookups however many a dispatch carries.
+JIRA_DEDUP_CHUNK_SIZE = 500
+JIRA_SKIPPED_REPORT_LIMIT = 100
+JIRA_ERROR_REPORT_MAX_LENGTH = 8192
+JIRA_FORCE_RETRY_MIN_AGE = timedelta(minutes=15)
+
+
+class _JiraPendingRecoveryOutcome(Enum):
+    LINKED = "linked"
+    NO_MATCH = "no_match"
+    UNRESOLVED = "unresolved"
+
+
+def _load_finding_refs(finding_ids: list[str]) -> dict[str, tuple[str, str]]:
+    """Map finding id -> (provider id, finding uid) for the batch, in one query."""
+    refs = {}
+    for finding_id, provider_id, uid in Finding.all_objects.filter(
+        id__in=finding_ids
+    ).values_list("id", "scan__provider_id", "uid"):
+        refs[str(finding_id)] = (str(provider_id), uid)
+    return refs
+
+
+def _load_existing_jira_issues(
+    tenant_id: str, integration_id: str, refs: dict[str, tuple[str, str]]
+) -> dict[tuple[str, str], JiraIssue]:
+    """Load the Jira issue rows already linked to the batch's findings.
+
+    Grouped by provider and chunked so each query is a bounded index lookup on
+    (tenant, integration, provider, finding_uid).
+    """
+    uids_by_provider: dict[str, list[str]] = {}
+    for provider_id, uid in refs.values():
+        uids_by_provider.setdefault(provider_id, []).append(uid)
+
+    existing: dict[tuple[str, str], JiraIssue] = {}
+    for provider_id, uids in uids_by_provider.items():
+        for start in range(0, len(uids), JIRA_DEDUP_CHUNK_SIZE):
+            chunk = uids[start : start + JIRA_DEDUP_CHUNK_SIZE]
+            for row in JiraIssue.objects.filter(
+                tenant_id=tenant_id,
+                integration_id=integration_id,
+                provider_id=provider_id,
+                finding_uid__in=chunk,
+            ):
+                existing[(str(row.provider_id), row.finding_uid)] = row
+    return existing
+
+
+def _load_jira_issue(
+    tenant_id: str,
+    integration_id: str,
+    provider_id: str,
+    finding_uid: str,
+) -> JiraIssue | None:
+    """Reload one ledger identity after a conditional write loses a race."""
+    with rls_transaction(tenant_id, using=MainRouter.default_db):
+        return JiraIssue.objects.filter(
+            tenant_id=tenant_id,
+            integration_id=integration_id,
+            provider_id=provider_id,
+            finding_uid=finding_uid,
+        ).first()
+
+
+def _refresh_jira_issue_statuses(
+    jira_integration: Jira, rows: list[JiraIssue]
+) -> dict[str, JiraIssueStatusResult]:
+    """Fetch linked issue statuses without holding a database transaction."""
+    if not rows:
+        return {}
+    references = [
+        JiraIssueReference(issue_id=row.issue_id, issue_key=row.issue_key)
+        for row in rows
+    ]
+    try:
+        results = jira_integration.get_issues_status(references)
+    except Exception:
+        logger.exception("Could not refresh Jira issue statuses")
+        results = []
+
+    if not isinstance(results, list) or len(results) != len(references):
+        results = [None] * len(references)
+
+    statuses = {}
+    for row, reference, status_result in zip(rows, references, results):
+        if (
+            not isinstance(status_result, JiraIssueStatusResult)
+            or status_result.reference != reference
+        ):
+            status_result = JiraIssueStatusResult(
+                reference=reference,
+                outcome=JiraIssueLookupOutcome.UNKNOWN,
+                error_code="malformed_status_result",
+                error_message="Jira returned an invalid issue status result.",
+            )
+        statuses[str(row.id)] = status_result
+    return statuses
+
+
+def _apply_jira_issue_status(
+    tenant_id: str, row: JiraIssue, status_result: JiraIssueStatusResult
+) -> bool:
+    """Cache a conclusive status if it still describes this linked issue."""
+    if status_result.outcome not in {
+        JiraIssueLookupOutcome.OPEN,
+        JiraIssueLookupOutcome.DONE,
+        JiraIssueLookupOutcome.MOVED,
+    }:
+        return False
+
+    current_values = (
+        status_result.current_issue_id,
+        status_result.current_issue_key,
+        status_result.current_issue_url,
+        status_result.status,
+        status_result.status_category,
+    )
+    if not all(isinstance(value, str) and value.strip() for value in current_values):
+        return False
+    if status_result.current_issue_id != row.issue_id:
+        return False
+    moved = status_result.outcome == JiraIssueLookupOutcome.MOVED
+    key_changed = status_result.current_issue_key != row.issue_key
+    if moved != key_changed:
+        return False
+    if status_result.status_category not in {
+        JiraIssue.StatusCategoryChoices.NEW,
+        JiraIssue.StatusCategoryChoices.INDETERMINATE,
+        JiraIssue.StatusCategoryChoices.DONE,
+    }:
+        return False
+    if (
+        status_result.outcome == JiraIssueLookupOutcome.OPEN
+        and status_result.status_category == JiraIssue.StatusCategoryChoices.DONE
+    ) or (
+        status_result.outcome == JiraIssueLookupOutcome.DONE
+        and status_result.status_category != JiraIssue.StatusCategoryChoices.DONE
+    ):
+        return False
+
+    now = timezone.now()
+    updates = {
+        "issue_status": status_result.status[:64],
+        "issue_status_category": status_result.status_category[:16],
+        "status_synced_at": now,
+        "updated_at": now,
+    }
+    if moved:
+        updates.update(
+            issue_key=status_result.current_issue_key[:64],
+            issue_url=status_result.current_issue_url[:2048],
+        )
+    with rls_transaction(tenant_id, using=MainRouter.default_db):
+        updated = JiraIssue.objects.filter(
+            id=row.id,
+            issue_id=row.issue_id,
+            issue_key=row.issue_key,
+            delivery_attempt_token__isnull=True,
+        ).update(**updates)
+    if not updated:
+        return False
+    for field, value in updates.items():
+        if field != "updated_at":
+            setattr(row, field, value)
+    return True
+
+
+def _update_latest_jira_finding_id(
+    tenant_id: str, row: JiraIssue, finding_id: str
+) -> None:
+    # Finding IDs are monotonic UUIDv7 values, so ordering reflects scan recency
+    # and prevents stale dispatches from moving the ledger pointer backward.
+    with rls_transaction(tenant_id, using=MainRouter.default_db):
+        JiraIssue.objects.filter(id=row.id, finding_id__lt=finding_id).update(
+            finding_id=finding_id,
+            updated_at=timezone.now(),
+        )
+
+
+def _reserve_initial_jira_issue(
+    tenant_id: str,
+    integration_id: str,
+    provider_id: str,
+    finding_uid: str,
+    finding_id: str,
+) -> JiraIssue | None:
+    """Reserve a new finding identity; the unique constraint chooses the sender."""
+    try:
+        with rls_transaction(tenant_id, using=MainRouter.default_db):
+            return JiraIssue.objects.create(
+                tenant_id=tenant_id,
+                integration_id=integration_id,
+                provider_id=provider_id,
+                finding_uid=finding_uid,
+                finding_id=finding_id,
+                delivery_attempt_token=uuid4(),
+            )
+    except IntegrityError:
+        return None
+
+
+def _reserve_jira_issue_replacement(
+    tenant_id: str, row: JiraIssue, finding_id: str
+) -> JiraIssue | None:
+    """Reserve replacement of a Done issue while preserving the current link."""
+    delivery_attempt_token = uuid4()
+    try:
+        with rls_transaction(tenant_id, using=MainRouter.default_db):
+            updated = JiraIssue.objects.filter(
+                id=row.id,
+                issue_id=row.issue_id,
+                issue_key=row.issue_key,
+                issue_status_category=JiraIssue.StatusCategoryChoices.DONE,
+                delivery_attempt_token__isnull=True,
+            ).update(
+                finding_id=finding_id,
+                delivery_attempt_token=delivery_attempt_token,
+                updated_at=timezone.now(),
+            )
+    except IntegrityError:
+        return None
+    if not updated:
+        return None
+    row.finding_id = finding_id
+    row.delivery_attempt_token = delivery_attempt_token
+    row.delivery_started_at = None
+    return row
+
+
+def _start_jira_delivery_attempt(
+    tenant_id: str, row: JiraIssue, delivery_attempt_token
+) -> bool:
+    """Mark a reserved delivery as possibly sent; only one worker may do so."""
+    started_at = timezone.now()
+    with rls_transaction(tenant_id, using=MainRouter.default_db):
+        updated = JiraIssue.objects.filter(
+            id=row.id,
+            delivery_attempt_token=delivery_attempt_token,
+            delivery_started_at__isnull=True,
+        ).update(
+            delivery_started_at=started_at,
+            updated_at=started_at,
+        )
+    if updated:
+        row.delivery_started_at = started_at
+    return bool(updated)
+
+
+def _link_jira_issue(
+    tenant_id: str,
+    row: JiraIssue,
+    delivery_attempt_token,
+    *,
+    issue_id: str,
+    issue_key: str,
+    issue_url: str,
+    project_key: str,
+    finding_id: str,
+) -> bool:
+    """Link a confirmed issue only if this worker still owns the marker."""
+    values = (issue_id, issue_key, issue_url, project_key)
+    if not all(isinstance(value, str) and value.strip() for value in values):
+        return False
+    updates = {
+        "issue_id": issue_id[:64],
+        "issue_key": issue_key[:64],
+        "issue_url": issue_url[:2048],
+        "project_key": project_key[:64],
+        "finding_id": finding_id,
+        "issue_status": None,
+        "issue_status_category": None,
+        "status_synced_at": None,
+        "delivery_attempt_token": None,
+        "delivery_started_at": None,
+        "updated_at": timezone.now(),
+    }
+    filters = {"id": row.id, "delivery_attempt_token": delivery_attempt_token}
+    if row.issue_id is None:
+        filters["issue_id__isnull"] = True
+    else:
+        filters["issue_id"] = row.issue_id
+    try:
+        with rls_transaction(tenant_id, using=MainRouter.default_db):
+            updated = JiraIssue.objects.filter(**filters).update(**updates)
+    except IntegrityError:
+        return False
+    if not updated:
+        return False
+    for field, value in updates.items():
+        if field != "updated_at":
+            setattr(row, field, value)
+    return True
+
+
+def _release_jira_delivery_attempt(
+    tenant_id: str, row: JiraIssue, delivery_attempt_token
+) -> bool:
+    """Release a confirmed failure without removing a previous issue link."""
+    with rls_transaction(tenant_id, using=MainRouter.default_db):
+        queryset = JiraIssue.objects.filter(
+            id=row.id,
+            delivery_attempt_token=delivery_attempt_token,
+        )
+        if row.issue_id is None:
+            deleted, _ = queryset.filter(issue_id__isnull=True).delete()
+            released = bool(deleted)
+        else:
+            released = bool(
+                queryset.filter(issue_id=row.issue_id).update(
+                    delivery_attempt_token=None,
+                    delivery_started_at=None,
+                    updated_at=timezone.now(),
+                )
+            )
+    if released:
+        row.delivery_attempt_token = None
+        row.delivery_started_at = None
+    return released
+
+
+def _skipped_entry(finding_id: str) -> dict:
+    return {"finding_id": str(finding_id)}
+
+
+def _recover_pending_jira_issue(
+    tenant_id: str,
+    jira_integration: Jira,
+    row: JiraIssue,
+    finding_id: str,
+) -> _JiraPendingRecoveryOutcome:
+    """Link exactly one marker match; every other result remains reserved."""
+    delivery_attempt_token = row.delivery_attempt_token
+    if delivery_attempt_token is None or row.delivery_started_at is None:
+        return _JiraPendingRecoveryOutcome.UNRESOLVED
+    try:
+        search_result = jira_integration.search_issues_by_delivery_attempt(
+            str(delivery_attempt_token)
+        )
+    except Exception:
+        logger.exception("Could not search Jira delivery marker for row %s", row.id)
+        return _JiraPendingRecoveryOutcome.UNRESOLVED
+    if not isinstance(search_result, JiraIssueSearchResult):
+        return _JiraPendingRecoveryOutcome.UNRESOLVED
+    if search_result.outcome != JiraIssueSearchOutcome.SUCCESS:
+        return _JiraPendingRecoveryOutcome.UNRESOLVED
+    if not search_result.matches:
+        return _JiraPendingRecoveryOutcome.NO_MATCH
+    if len(search_result.matches) != 1:
+        return _JiraPendingRecoveryOutcome.UNRESOLVED
+    match = search_result.matches[0]
+    if not isinstance(match, JiraIssueSearchMatch) or not all(
+        isinstance(value, str) and value.strip()
+        for value in (match.issue_id, match.issue_key, match.issue_url)
+    ):
+        return _JiraPendingRecoveryOutcome.UNRESOLVED
+    project_key, separator, _ = match.issue_key.rpartition("-")
+    if not separator or not project_key:
+        return _JiraPendingRecoveryOutcome.UNRESOLVED
+    linked = _link_jira_issue(
+        tenant_id,
+        row,
+        delivery_attempt_token,
+        issue_id=match.issue_id,
+        issue_key=match.issue_key,
+        issue_url=match.issue_url,
+        project_key=project_key,
+        finding_id=finding_id,
+    )
+    if linked:
+        return _JiraPendingRecoveryOutcome.LINKED
+    return _JiraPendingRecoveryOutcome.UNRESOLVED
+
+
+def _reset_stale_jira_delivery_attempt(
+    tenant_id: str,
+    row: JiraIssue,
+    delivery_attempt_token,
+) -> bool:
+    """Allow an explicit retry only while the same stale attempt is still owned."""
+    delivery_started_at = row.delivery_started_at
+    if delivery_started_at is None:
+        return False
+    retry_cutoff = timezone.now() - JIRA_FORCE_RETRY_MIN_AGE
+    if delivery_started_at > retry_cutoff:
+        return False
+
+    now = timezone.now()
+    with rls_transaction(tenant_id, using=MainRouter.default_db):
+        updated = JiraIssue.objects.filter(
+            id=row.id,
+            delivery_attempt_token=delivery_attempt_token,
+            delivery_started_at=delivery_started_at,
+        ).update(
+            delivery_started_at=None,
+            updated_at=now,
+        )
+    if updated:
+        row.delivery_started_at = None
+    return bool(updated)
+
+
+def _get_jira_send_payload(
+    tenant_id: str,
+    finding_id: str,
+    project_key: str,
+    issue_type: str,
+    tenant_info: str,
+) -> dict:
+    """Build a finding payload inside RLS, ready for an external Jira call."""
+    with rls_transaction(tenant_id, using=MainRouter.default_db):
+        finding = (
+            Finding.all_objects.select_related("scan__provider")
+            .prefetch_related("resources")
+            .get(id=finding_id)
+        )
+        resource = finding.resources.first() if finding.resources.exists() else None
+        resource_tags = (
+            resource.get_tags(tenant_id)
+            if resource and hasattr(resource, "tags")
+            else {}
+        )
+        check_metadata = finding.check_metadata or {}
+        remediation = check_metadata.get("remediation", {}) or {}
+        recommendation = remediation.get("recommendation", {}) or {}
+        remediation_code = remediation.get("code", {}) or {}
+        provider_type = finding.scan.provider.provider
+        return {
+            "check_id": finding.check_id,
+            "check_title": check_metadata.get("checktitle", ""),
+            "severity": finding.severity,
+            "status": finding.status,
+            "status_extended": finding.status_extended or "",
+            "provider": provider_type,
+            "region": resource.region if resource and resource.region else "",
+            "resource_uid": resource.uid if resource else "",
+            "resource_name": resource.name if resource else "",
+            "risk": check_metadata.get("risk", ""),
+            "recommendation_text": recommendation.get("text", ""),
+            "recommendation_url": recommendation.get("url", ""),
+            "remediation_code_native_iac": remediation_code.get("nativeiac", ""),
+            "remediation_code_terraform": remediation_code.get("terraform", ""),
+            "remediation_code_cli": remediation_code.get("cli", ""),
+            "remediation_code_other": remediation_code.get("other", ""),
+            "resource_tags": resource_tags,
+            "compliance": finding.compliance or {},
+            "project_key": project_key,
+            "issue_type": issue_type,
+            "issue_labels": build_jira_issue_labels(
+                finding_uid=finding.uid,
+                provider=provider_type,
+                severity=finding.severity,
+                check_id=finding.check_id,
+            ),
+            "finding_url": build_jira_finding_url(finding.uid),
+            "tenant_info": tenant_info,
+        }
+
+
+def _send_reserved_jira_finding(
+    jira_integration: Jira, payload: dict, delivery_attempt_token
+) -> JiraCreationResult:
+    try:
+        creation_result = jira_integration.send_finding(
+            **payload,
+            delivery_attempt_marker=str(delivery_attempt_token),
+        )
+    except Exception:
+        logger.exception("Jira raised while sending a reserved finding")
+        return JiraCreationResult(
+            outcome=JiraCreationOutcome.UNCERTAIN,
+            delivery_marker=str(delivery_attempt_token),
+            error_code="unexpected_send_exception",
+            error_message=JIRA_GENERIC_SEND_ERROR,
+        )
+    if not isinstance(creation_result, JiraCreationResult):
+        return JiraCreationResult(
+            outcome=JiraCreationOutcome.UNCERTAIN,
+            delivery_marker=str(delivery_attempt_token),
+            error_code="invalid_creation_result",
+            error_message=JIRA_GENERIC_SEND_ERROR,
+        )
+    return creation_result
+
+
+def _jira_creation_error(creation_result: JiraCreationResult) -> str:
+    message = str(creation_result.error_message or JIRA_GENERIC_SEND_ERROR).strip()
+    return message[:2048] or JIRA_GENERIC_SEND_ERROR
+
+
 def send_findings_to_jira(
     tenant_id: str,
     integration_id: str,
     project_key: str,
     issue_type: str,
     finding_ids: list[str],
+    force_retry: bool = False,
 ):
-    with rls_transaction(tenant_id):
+    """Deliver findings through the finding-to-Jira ledger."""
+    with rls_transaction(tenant_id, using=MainRouter.default_db):
         integration = Integration.objects.get(id=integration_id)
-        jira_integration = initialize_prowler_integration(integration)
         tenant_info = get_tenant_name(tenant_id)
+        finding_refs = _load_finding_refs(finding_ids)
+        existing = _load_existing_jira_issues(tenant_id, integration_id, finding_refs)
+    jira_integration = initialize_prowler_integration(integration)
 
-    num_tickets_created = 0
+    status_rows = [
+        row
+        for row in existing.values()
+        if row.is_linked and row.delivery_attempt_token is None
+    ]
+    statuses = _refresh_jira_issue_statuses(jira_integration, status_rows)
+
+    created_count = 0
+    deferred_count = 0
+    failed_count = 0
+    skipped_count = 0
+    skipped = []
     error_messages = []
+    processed_identities = set()
+
+    def record_skip(finding_id: str) -> None:
+        nonlocal skipped_count
+        skipped_count += 1
+        if len(skipped) < JIRA_SKIPPED_REPORT_LIMIT:
+            skipped.append(_skipped_entry(finding_id))
+
+    def record_lost_attempt(
+        finding_id: str,
+        identity: tuple[str, str],
+        previous_issue_id: str | None,
+    ) -> None:
+        nonlocal deferred_count, failed_count
+        current = _load_jira_issue(
+            tenant_id,
+            integration_id,
+            identity[0],
+            identity[1],
+        )
+        if current is None:
+            existing.pop(identity, None)
+        else:
+            existing[identity] = current
+
+        delivery_is_still_owned = (
+            current is not None and current.delivery_attempt_token is not None
+        )
+        another_issue_was_linked = (
+            current is not None
+            and current.issue_id is not None
+            and (previous_issue_id is None or current.issue_id != previous_issue_id)
+        )
+        if another_issue_was_linked:
+            record_skip(finding_id)
+            return
+        if delivery_is_still_owned:
+            deferred_count += 1
+            return
+
+        failed_count += 1
+        error_messages.append(JIRA_GENERIC_SEND_ERROR)
+
     for finding_id in finding_ids:
-        with rls_transaction(tenant_id):
-            finding_instance = (
-                Finding.all_objects.select_related("scan__provider")
-                .prefetch_related("resources")
-                .get(id=finding_id)
+        finding_id = str(finding_id)
+        finding_ref = finding_refs.get(finding_id)
+        if finding_ref is None:
+            logger.warning("Finding %s could not be loaded for Jira", finding_id)
+            failed_count += 1
+            error_messages.append(JIRA_GENERIC_SEND_ERROR)
+            continue
+
+        provider_id, finding_uid = finding_ref
+        identity = (provider_id, finding_uid)
+        row = existing.get(identity)
+        if row is not None:
+            _update_latest_jira_finding_id(tenant_id, row, finding_id)
+        if identity in processed_identities:
+            record_skip(finding_id)
+            continue
+        processed_identities.add(identity)
+
+        resume_reserved_attempt = (
+            row is not None
+            and row.delivery_attempt_token is not None
+            and row.delivery_started_at is None
+        )
+        if (
+            row is not None
+            and row.delivery_attempt_token is not None
+            and not resume_reserved_attempt
+        ):
+            recovery_outcome = _recover_pending_jira_issue(
+                tenant_id, jira_integration, row, finding_id
             )
-
-            # Extract resource information
-            resource = (
-                finding_instance.resources.first()
-                if finding_instance.resources.exists()
-                else None
-            )
-            resource_uid = resource.uid if resource else ""
-            resource_name = resource.name if resource else ""
-            resource_tags = {}
-            if resource and hasattr(resource, "tags"):
-                resource_tags = resource.get_tags(tenant_id)
-
-            # Get region
-            region = resource.region if resource and resource.region else ""
-
-            # Extract remediation information from check_metadata
-            check_metadata = finding_instance.check_metadata
-            remediation = check_metadata.get("remediation", {})
-            recommendation = remediation.get("recommendation", {})
-            remediation_code = remediation.get("code", {})
-
-            provider_type = finding_instance.scan.provider.provider
-            issue_labels = build_jira_issue_labels(
-                finding_uid=finding_instance.uid,
-                provider=provider_type,
-                severity=finding_instance.severity,
-                check_id=finding_instance.check_id,
-            )
-            finding_url = build_jira_finding_url(finding_instance.uid)
-
-            try:
-                # Send the individual finding to Jira
-                result = jira_integration.send_finding(
-                    check_id=finding_instance.check_id,
-                    check_title=check_metadata.get("checktitle", ""),
-                    severity=finding_instance.severity,
-                    status=finding_instance.status,
-                    status_extended=finding_instance.status_extended or "",
-                    provider=provider_type,
-                    region=region,
-                    resource_uid=resource_uid,
-                    resource_name=resource_name,
-                    risk=check_metadata.get("risk", ""),
-                    recommendation_text=recommendation.get("text", ""),
-                    recommendation_url=recommendation.get("url", ""),
-                    remediation_code_native_iac=remediation_code.get("nativeiac", ""),
-                    remediation_code_terraform=remediation_code.get("terraform", ""),
-                    remediation_code_cli=remediation_code.get("cli", ""),
-                    remediation_code_other=remediation_code.get("other", ""),
-                    resource_tags=resource_tags,
-                    compliance=finding_instance.compliance or {},
-                    project_key=project_key,
-                    issue_type=issue_type,
-                    issue_labels=issue_labels,
-                    finding_url=finding_url,
-                    tenant_info=tenant_info,
-                )
-            except JiraBaseException as error:
-                error_message = error.message or JIRA_GENERIC_SEND_ERROR
-                logger.exception(
-                    "Failed to send finding %s to Jira: %s", finding_id, error_message
-                )
-                error_messages.append(error_message)
+            if recovery_outcome == _JiraPendingRecoveryOutcome.LINKED:
+                created_count += 1
                 continue
-            except Exception:
-                logger.exception("Failed to send finding %s to Jira", finding_id)
-                error_messages.append(JIRA_GENERIC_SEND_ERROR)
+            if (
+                force_retry
+                and recovery_outcome == _JiraPendingRecoveryOutcome.NO_MATCH
+                and _reset_stale_jira_delivery_attempt(
+                    tenant_id, row, row.delivery_attempt_token
+                )
+            ):
+                logger.warning(
+                    "Force retrying stale Jira delivery for tenant %s, integration %s, provider %s, finding %s",
+                    tenant_id,
+                    integration_id,
+                    provider_id,
+                    finding_uid,
+                )
+                resume_reserved_attempt = True
+            else:
+                if (
+                    force_retry
+                    and recovery_outcome == _JiraPendingRecoveryOutcome.NO_MATCH
+                ):
+                    deferred_count += 1
+                else:
+                    record_skip(finding_id)
                 continue
 
-            if result:
-                num_tickets_created += 1
+        needs_replacement = bool(resume_reserved_attempt and row.is_linked)
+        if not resume_reserved_attempt and row is not None and row.is_linked:
+            status_result = statuses.get(str(row.id))
+            if status_result is None or not _apply_jira_issue_status(
+                tenant_id, row, status_result
+            ):
+                record_skip(finding_id)
+                continue
+            needs_replacement = (
+                status_result.outcome == JiraIssueLookupOutcome.DONE
+                or (
+                    status_result.outcome == JiraIssueLookupOutcome.MOVED
+                    and status_result.status_category
+                    == JiraIssue.StatusCategoryChoices.DONE
+                )
+            )
+            if not needs_replacement:
+                record_skip(finding_id)
+                continue
+        elif not resume_reserved_attempt and row is not None:
+            record_skip(finding_id)
+            continue
+
+        try:
+            payload = _get_jira_send_payload(
+                tenant_id,
+                finding_id,
+                project_key,
+                issue_type,
+                tenant_info,
+            )
+        except Exception:
+            logger.exception("Failed to build finding %s for Jira", finding_id)
+            failed_count += 1
+            error_messages.append(JIRA_GENERIC_SEND_ERROR)
+            continue
+
+        previous_issue_id = row.issue_id if needs_replacement else None
+        if not resume_reserved_attempt:
+            if needs_replacement:
+                row = _reserve_jira_issue_replacement(tenant_id, row, finding_id)
+            else:
+                row = _reserve_initial_jira_issue(
+                    tenant_id,
+                    integration_id,
+                    provider_id,
+                    finding_uid,
+                    finding_id,
+                )
+        if row is None:
+            record_lost_attempt(finding_id, identity, previous_issue_id)
+            continue
+        existing[identity] = row
+
+        delivery_attempt_token = row.delivery_attempt_token
+        if delivery_attempt_token is None or not _start_jira_delivery_attempt(
+            tenant_id, row, delivery_attempt_token
+        ):
+            record_lost_attempt(finding_id, identity, previous_issue_id)
+            continue
+        creation_result = _send_reserved_jira_finding(
+            jira_integration, payload, delivery_attempt_token
+        )
+        if creation_result.outcome == JiraCreationOutcome.CONFIRMED_SUCCESS:
+            linked = _link_jira_issue(
+                tenant_id,
+                row,
+                delivery_attempt_token,
+                issue_id=creation_result.issue_id,
+                issue_key=creation_result.issue_key,
+                issue_url=creation_result.issue_url,
+                project_key=project_key,
+                finding_id=finding_id,
+            )
+            if linked:
+                created_count += 1
                 logger.info(
                     "Finding %s sent to Jira as %s",
                     finding_id,
-                    result.get("key") if isinstance(result, dict) else result,
+                    creation_result.issue_key,
                 )
             else:
-                error_message = JIRA_GENERIC_SEND_ERROR
-                logger.error(error_message)
-                error_messages.append(error_message)
+                failed_count += 1
+                error_messages.append(JIRA_GENERIC_SEND_ERROR)
+            continue
+
+        failed_count += 1
+        error_messages.append(_jira_creation_error(creation_result))
+        if creation_result.outcome in {
+            JiraCreationOutcome.CONFIRMED_REJECTION,
+            JiraCreationOutcome.RETRYABLE_FAILURE,
+        }:
+            _release_jira_delivery_attempt(tenant_id, row, delivery_attempt_token)
+            if row.issue_id is None:
+                existing.pop(identity, None)
 
     result = {
-        "created_count": num_tickets_created,
-        "failed_count": len(finding_ids) - num_tickets_created,
+        "created_count": created_count,
+        "deferred_count": deferred_count,
+        "failed_count": failed_count,
+        "skipped_count": skipped_count,
     }
     if error_messages:
-        result["error"] = "; ".join(dict.fromkeys(error_messages))
-
+        result["error"] = "; ".join(dict.fromkeys(error_messages))[
+            :JIRA_ERROR_REPORT_MAX_LENGTH
+        ]
+    if skipped:
+        result["skipped"] = skipped
     return result

api/src/backend/tasks/tasks.py

@@ -1378,8 +1378,8 @@ def security_hub_integration_task(
     return upload_security_hub_integration(tenant_id, provider_id, scan_id)
 
 
-# acks_late=False: Jira sends are not deduplicated and the task is not auto-recovered,
-# so a crashed send is dropped rather than redelivered (avoids duplicate Jira issues).
+# A Jira POST can remain ambiguous after worker loss, so this manual task stays
+# early-acknowledged and relies on a later explicit send for marker recovery.
 @shared_task(
     base=RLSTask,
     name="integration-jira",
@@ -1392,9 +1392,15 @@ def jira_integration_task(
     project_key: str,
     issue_type: str,
     finding_ids: list[str],
+    force_retry: bool = False,
 ):
     return send_findings_to_jira(
-        tenant_id, integration_id, project_key, issue_type, finding_ids
+        tenant_id,
+        integration_id,
+        project_key,
+        issue_type,
+        finding_ids,
+        force_retry=force_retry,
     )
 
 

api/src/backend/tasks/tests/test_deletion.py

@@ -2,7 +2,7 @@ from unittest.mock import MagicMock, call, patch
 
 import pytest
 from api.attack_paths import database as graph_database
-from api.models import Provider, Tenant, TenantComplianceSummary
+from api.models import JiraIssue, Provider, Tenant, TenantComplianceSummary
 from django.core.exceptions import ObjectDoesNotExist
 from tasks.jobs.deletion import delete_provider, delete_tenant
 
@@ -33,6 +33,22 @@ class TestDeleteProvider:
                 str(instance.id),
             )
 
+    def test_delete_provider_removes_jira_issues(self, jira_issues_fixture):
+        linked, other_provider_issue, reservation = jira_issues_fixture
+        provider = linked.provider
+        tenant_id = str(provider.tenant_id)
+        with (
+            patch("tasks.jobs.deletion.graph_database.get_database_name"),
+            patch("tasks.jobs.deletion.graph_database.drop_subgraph"),
+        ):
+            delete_provider(tenant_id, provider.id)
+
+        remaining = set(JiraIssue.objects.values_list("id", flat=True))
+        assert linked.id not in remaining
+        assert reservation.id not in remaining
+        # Issues of other providers are untouched
+        assert other_provider_issue.id in remaining
+
     def test_delete_provider_does_not_exist(self, tenants_fixture):
         with (
             patch(

docs/user-guide/tutorials/prowler-app-jira-integration.mdx

@@ -136,6 +136,24 @@ Every Jira issue created from a single Finding carries a stable reference back t
 
 Prowler Cloud always includes the Finding URL. In Prowler Local Server, set `DJANGO_UI_BASE_URL` in the API environment (for example, `https://prowler.example.com`) to enable it. When the variable is empty, the issue is created without the link.
 
+### Sending a Finding That Already Has a Jira Issue
+
+<VersionBadge version="5.42.0" />
+
+Prowler remembers each confirmed Jira issue created for a Finding, keyed by the Finding UID, so sending the same Finding again does not create a duplicate issue:
+
+* If the linked issue is still open in Jira, the Finding is skipped. The task result reports `skipped_count` and the skipped Finding IDs; use `GET /api/v1/jira-issues` to read confirmed Jira references.
+* If the linked issue moved to a new Jira key, Prowler verifies its issue ID, refreshes the saved key and URL, and skips the Finding while the issue remains open.
+* If the linked issue has a Jira status in the **Done** category, Prowler creates a replacement and links the Finding to the new issue after Jira confirms it.
+* If Jira reports that the issue is missing or forbidden, or its status cannot be determined safely, Prowler keeps the existing link and skips the Finding.
+* If another task already owns the delivery, Prowler reports the Finding as deferred instead of treating it as already ticketed. Run the same send action after the active task finishes.
+
+The link, and the last status observed in Jira, are available through the API at `GET /api/v1/jira-issues` (filter by `finding_uid`, `finding_uid__in`, `provider_id`, `integration`, `issue_key`, `issue_status_category` or `issue_status_category__in`). Each Jira integration keeps its own links, so the same Finding can have one issue per integration.
+
+Prowler uses a private delivery marker to check whether Jira created an issue when a delivery result is uncertain. A reservation that stopped before the delivery attempt began can be resumed safely by a later explicit send. After an attempt starts, Prowler searches Jira by the marker: exactly one matching issue completes the link, while no match, multiple matches, or a lookup error keeps the reservation and prevents an automatic retry.
+
+An authorized API caller can repeat a dispatch with `force_retry: true` when an attempt has been pending for at least 15 minutes. Prowler retries only when Jira returns zero matches for the saved marker, and it reuses that marker. This action accepts a duplicate-issue risk because an empty Jira search cannot prove that Jira never received the original request. Multiple matches and lookup errors remain blocked for manual investigation.
+
 ## Integration Status
 
 Monitor and manage your Jira integrations through the management interface:
@@ -237,7 +255,9 @@ If you don't have `jq` installed, run the command without `| jq`.
         "state": "completed",
         "result": {
           "created_count": 0,
-          "failed_count": 1
+          "deferred_count": 0,
+          "failed_count": 1,
+          "skipped_count": 0
         },
         "task_args": {
           "integration_id": "a476c2c0-0a00-4720-bfb9-286e9eb5c7bd",
@@ -260,7 +280,9 @@ If you don't have `jq` installed, run the command without `| jq`.
         "state": "completed",
         "result": {
           "created_count": 1,
-          "failed_count": 0
+          "deferred_count": 0,
+          "failed_count": 0,
+          "skipped_count": 0
         },
         "task_args": {
           "integration_id": "a476c2c0-0a00-4720-bfb9-286e9eb5c7bd",
@@ -288,4 +310,6 @@ If you don't have `jq` installed, run the command without `| jq`.
 How to read it:
 
 * "created_count": number of Jira issues successfully created.
+* "deferred_count": number of Findings owned by another delivery task. Run the same send action after that task finishes.
 * "failed_count": number of Jira issues that could not be created. If `failed_count > 0` or the issue does not appear in Jira, please contact us so we can assist while detailed logs are not available through the UI.
+* "skipped_count": number of Findings not sent because an existing Jira issue does not need replacement, a pending delivery remains unresolved, or an equivalent Finding was already processed in the request.