From e307a18225ca1f813e5509e89510ac83de9d056a Mon Sep 17 00:00:00 2001
From: Your Name <you@example.com>
Date: Thu, 25 Jun 2026 11:10:16 +0800
Subject: [PATCH] =?UTF-8?q?fix(awooop):=20=E7=A9=A9=E5=AE=9A=20source=20co?=
 =?UTF-8?q?rrelation=20=E8=AE=80=E5=9B=9E=E6=8E=92=E5=BA=8F?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 .../services/channel_event_dossier_service.py | 56 +++++++++++++++----
 .../test_channel_event_dossier_service.py     | 43 ++++++++++++++
 2 files changed, 87 insertions(+), 12 deletions(-)

diff --git a/apps/api/src/services/channel_event_dossier_service.py b/apps/api/src/services/channel_event_dossier_service.py
index df7d9f65..dc1832fb 100644
--- a/apps/api/src/services/channel_event_dossier_service.py
+++ b/apps/api/src/services/channel_event_dossier_service.py
@@ -2052,7 +2052,25 @@ async def _fetch_source_review_decisions_by_work_item(
 
     result = await db.execute(
         text("""
-            SELECT DISTINCT ON (context->>'work_item_id')
+            WITH ranked AS (
+                SELECT
+                    id,
+                    incident_id,
+                    actor,
+                    action_detail,
+                    success,
+                    context,
+                    created_at,
+                    ROW_NUMBER() OVER (
+                        PARTITION BY context->>'work_item_id'
+                        ORDER BY created_at DESC
+                    ) AS rn
+                FROM alert_operation_log
+                WHERE actor = :actor
+                  AND context->>'schema_version' = :schema_version
+                  AND COALESCE(context->>'project_id', :project_id) = :project_id
+            )
+            SELECT
                 id,
                 incident_id,
                 actor,
@@ -2060,11 +2078,9 @@ async def _fetch_source_review_decisions_by_work_item(
                 success,
                 context,
                 created_at
-            FROM alert_operation_log
-            WHERE actor = :actor
-              AND context->>'schema_version' = :schema_version
-              AND COALESCE(context->>'project_id', :project_id) = :project_id
-            ORDER BY context->>'work_item_id', created_at DESC
+            FROM ranked
+            WHERE rn = 1
+            ORDER BY created_at DESC
             LIMIT :limit
         """),
         {
@@ -2094,7 +2110,25 @@ async def _fetch_source_applies_by_work_item(
 
     result = await db.execute(
         text("""
-            SELECT DISTINCT ON (context->>'work_item_id')
+            WITH ranked AS (
+                SELECT
+                    id,
+                    incident_id,
+                    actor,
+                    action_detail,
+                    success,
+                    context,
+                    created_at,
+                    ROW_NUMBER() OVER (
+                        PARTITION BY context->>'work_item_id'
+                        ORDER BY created_at DESC
+                    ) AS rn
+                FROM alert_operation_log
+                WHERE actor = :actor
+                  AND context->>'schema_version' = :schema_version
+                  AND COALESCE(context->>'project_id', :project_id) = :project_id
+            )
+            SELECT
                 id,
                 incident_id,
                 actor,
@@ -2102,11 +2136,9 @@ async def _fetch_source_applies_by_work_item(
                 success,
                 context,
                 created_at
-            FROM alert_operation_log
-            WHERE actor = :actor
-              AND context->>'schema_version' = :schema_version
-              AND COALESCE(context->>'project_id', :project_id) = :project_id
-            ORDER BY context->>'work_item_id', created_at DESC
+            FROM ranked
+            WHERE rn = 1
+            ORDER BY created_at DESC
             LIMIT :limit
         """),
         {
diff --git a/apps/api/tests/test_channel_event_dossier_service.py b/apps/api/tests/test_channel_event_dossier_service.py
index e5da7d2b..bcb84d96 100644
--- a/apps/api/tests/test_channel_event_dossier_service.py
+++ b/apps/api/tests/test_channel_event_dossier_service.py
@@ -23,6 +23,49 @@ from src.services.channel_event_dossier_service import (
 )
 
 
+class _FakeMappings:
+    def all(self) -> list[dict]:
+        return []
+
+
+class _FakeResult:
+    def mappings(self) -> _FakeMappings:
+        return _FakeMappings()
+
+
+class _CaptureSqlDb:
+    def __init__(self) -> None:
+        self.statements: list[str] = []
+
+    async def execute(self, statement, params):  # noqa: ANN001, ANN201
+        self.statements.append(str(statement))
+        return _FakeResult()
+
+
+@pytest.mark.asyncio
+async def test_source_correlation_read_model_ranks_latest_rows_before_limit() -> None:
+    db = _CaptureSqlDb()
+
+    await channel_event_dossier_service._fetch_source_review_decisions_by_work_item(
+        db,
+        project_id="awoooi",
+        limit=300,
+    )
+    await channel_event_dossier_service._fetch_source_applies_by_work_item(
+        db,
+        project_id="awoooi",
+        limit=300,
+    )
+
+    assert len(db.statements) == 2
+    for sql in db.statements:
+        assert "ROW_NUMBER() OVER" in sql
+        assert "PARTITION BY context->>'work_item_id'" in sql
+        assert "WHERE rn = 1" in sql
+        assert "ORDER BY created_at DESC" in sql
+        assert "SELECT DISTINCT ON" not in sql
+
+
 def test_build_dossier_event_summarizes_source_envelope() -> None:
     event = build_dossier_event(
         {