fix(platform): bind AI runs to canonical correlation

apps/api/src/services/platform_runtime.py

@@ -19,17 +19,27 @@ from __future__ import annotations
 
 import hashlib
 import json
-import uuid
 from datetime import datetime, timedelta, timezone
 from typing import Any
+from uuid import UUID
 
 import structlog
-from sqlalchemy import select
-from sqlalchemy.dialects.postgresql import insert as pg_insert
+from sqlalchemy import select, text
 
-from src.db.awooop_models import AwoooPRunIdempotency, AwoooPRunState, AwoooPRunStepJournal
+from src.db.awooop_models import (
+    AwoooPRunIdempotency,
+    AwoooPRunState,
+    AwoooPRunStepJournal,
+)
 from src.db.base import get_db_context
 from src.services.run_state_machine import LEASE_TTL_SECONDS, transition
+from src.services.runtime_correlation import (
+    canonical_traceparent,
+    canonical_work_item_id,
+    is_canonical_traceparent,
+    new_runtime_correlation,
+    new_uuid7,
+)
 
 logger = structlog.get_logger(__name__)
 
@@ -46,26 +56,20 @@ _DESTRUCTIVE_TOOL_KEYWORDS = frozenset({
 # UUID v7（時間有序）
 # ─────────────────────────────────────────────────────────────────────────────
 
-def _uuid7() -> uuid.UUID:
-    """
-    生成 UUID v7（時間有序，適合資料庫 PK）。
-    格式：48-bit Unix timestamp ms + version(7) + 74-bit random
-    """
-    now_ms = int(datetime.now(timezone.utc).timestamp() * 1000)
-    rand_bits = int.from_bytes(uuid.uuid4().bytes[6:], "big") & 0x3FFFFFFFFFFFFFFF
-    val = (now_ms << 80) | (0x7 << 76) | (0x8 << 72) | rand_bits
-    return uuid.UUID(int=val)
+def _uuid7() -> UUID:
+    """Compatibility wrapper for the canonical RFC 9562 generator."""
+
+    return new_uuid7()
 
 
 # ─────────────────────────────────────────────────────────────────────────────
 # W3C traceparent 生成
 # ─────────────────────────────────────────────────────────────────────────────
 
-def _new_trace_id() -> str:
-    """生成 W3C traceparent-compatible trace_id"""
-    trace_id_hex = uuid.uuid4().hex + uuid.uuid4().hex[:16]  # 128-bit
-    span_id_hex = uuid.uuid4().hex[:16]                       # 64-bit
-    return f"00-{trace_id_hex}-{span_id_hex}-01"
+def _new_trace_id(run_id: UUID | None = None) -> str:
+    """Return the canonical W3C traceparent bound to one durable run."""
+
+    return canonical_traceparent(run_id or _uuid7())
 
 
 # ─────────────────────────────────────────────────────────────────────────────
@@ -74,13 +78,14 @@ def _new_trace_id() -> str:
 
 _IDEMPOTENCY_REDIS_PREFIX = "awooop:run:idem:"
 _IDEMPOTENCY_REDIS_TTL = 86400  # 24h
+_IDEMPOTENCY_LOCK_PREFIX = "awooop:run:idem:claim:"
 
 
 async def check_idempotency(
     project_id: str,
     channel_type: str,
     provider_event_id: str,
-) -> uuid.UUID | None:
+) -> UUID | None:
     """
     檢查 (project_id, channel_type, provider_event_id) 是否已有對應 run_id。
 
@@ -91,24 +96,20 @@ async def check_idempotency(
     """
     idem_key = f"{_IDEMPOTENCY_REDIS_PREFIX}{project_id}:{channel_type}:{provider_event_id}"
 
-    # Layer 1: Redis
+    cached_run_id: UUID | None = None
+    # Redis is only a lookup hint. PostgreSQL remains the durable authority so
+    # an old or interrupted cache write can never select a different run.
     try:
         from src.core.redis_client import get_redis
         redis = get_redis()
         cached = await redis.get(idem_key)
         if cached:
             run_id_str = cached.decode() if isinstance(cached, bytes) else cached
-            logger.info(
-                "idempotency_hit_redis",
-                project_id=project_id,
-                provider_event_id=provider_event_id,
-                run_id=run_id_str,
-            )
-            return uuid.UUID(run_id_str)
+            cached_run_id = UUID(run_id_str)
     except Exception as exc:
         logger.warning("idempotency_redis_check_failed", error=str(exc))
 
-    # Layer 2: PostgreSQL
+    # PostgreSQL is the only authoritative mapping.
     try:
         async with get_db_context(project_id) as db:
             result = await db.execute(
@@ -120,44 +121,43 @@ async def check_idempotency(
             )
             row = result.fetchone()
             if row:
-                return uuid.UUID(str(row[0]))
+                durable_run_id = UUID(str(row[0]))
+                if cached_run_id is not None and cached_run_id != durable_run_id:
+                    logger.warning(
+                        "idempotency_redis_durable_mismatch_ignored",
+                        project_id=project_id,
+                        provider_event_id=provider_event_id,
+                    )
+                return durable_run_id
     except Exception as exc:
         logger.warning("idempotency_pg_check_failed", error=str(exc))
 
+    if cached_run_id is not None:
+        logger.warning(
+            "idempotency_redis_orphan_hint_ignored",
+            project_id=project_id,
+            provider_event_id=provider_event_id,
+        )
     return None
 
 
-async def _register_idempotency(
+async def _cache_idempotency(
     project_id: str,
     channel_type: str,
     provider_event_id: str,
-    run_id: uuid.UUID,
+    run_id: UUID,
 ) -> None:
-    """寫入 idempotency 記錄（Redis + PostgreSQL）"""
+    """Cache a PostgreSQL-selected winner; never create durable authority."""
     idem_key = f"{_IDEMPOTENCY_REDIS_PREFIX}{project_id}:{channel_type}:{provider_event_id}"
     run_id_str = str(run_id)
 
-    # Redis NX（若已有其他 worker 寫入，NX 失敗，無害）
     try:
         from src.core.redis_client import get_redis
         redis = get_redis()
-        await redis.set(idem_key, run_id_str, ex=_IDEMPOTENCY_REDIS_TTL, nx=True)
+        await redis.set(idem_key, run_id_str, ex=_IDEMPOTENCY_REDIS_TTL)
     except Exception as exc:
         logger.warning("idempotency_redis_write_failed", error=str(exc))
 
-    # PostgreSQL（INSERT OR IGNORE）
-    try:
-        async with get_db_context(project_id) as db:
-            stmt = pg_insert(AwoooPRunIdempotency).values(
-                project_id=project_id,
-                channel_type=channel_type,
-                provider_event_id=provider_event_id,
-                run_id=run_id,
-            ).on_conflict_do_nothing(constraint="uix_run_idempotency_key")
-            await db.execute(stmt)
-    except Exception as exc:
-        logger.warning("idempotency_pg_write_failed", error=str(exc))
-
 
 # ─────────────────────────────────────────────────────────────────────────────
 # Shadow destructive tool check
@@ -192,7 +192,7 @@ async def create_run(
     channel_type: str | None = None,
     provider_event_id: str | None = None,
     timeout_seconds: int = 600,
-) -> tuple[uuid.UUID, bool]:
+) -> tuple[UUID, bool]:
     """
     建立新 run（或返回既有 run，若重複事件）。
 
@@ -201,22 +201,14 @@ async def create_run(
 
     Shadow mode：is_shadow=True，不產生 user response。
     """
-    # Idempotency 檢查
-    if channel_type and provider_event_id:
-        existing_run_id = await check_idempotency(project_id, channel_type, provider_event_id)
-        if existing_run_id:
-            logger.info(
-                "run_creation_idempotent",
-                project_id=project_id,
-                channel_type=channel_type,
-                provider_event_id=provider_event_id,
-                existing_run_id=str(existing_run_id),
-            )
-            return existing_run_id, True
-
-    run_id = _uuid7()
-    trace_id = _new_trace_id()
-    timeout_at = datetime.now(timezone.utc) + timedelta(seconds=timeout_seconds)
+    correlation = new_runtime_correlation(project_id)
+    run_id = correlation.run_id
+    trace_id = correlation.trace_id
+    # AwoooPRunState timestamps use PostgreSQL TIMESTAMP WITHOUT TIME ZONE.
+    # Normalize UTC before binding so asyncpg does not reject an aware value.
+    timeout_at = datetime.now(timezone.utc).replace(tzinfo=None) + timedelta(
+        seconds=timeout_seconds
+    )
 
     # 計算 input_sha256
     input_sha256 = None
@@ -224,24 +216,85 @@ async def create_run(
         canonical = json.dumps(input_payload, sort_keys=True, separators=(",", ":"))
         input_sha256 = hashlib.sha256(canonical.encode()).hexdigest()
 
-    async with get_db_context(project_id) as db:
-        run = AwoooPRunState(
-            run_id=run_id,
-            project_id=project_id,
-            agent_id=agent_id,
-            state="pending",
-            trace_id=trace_id,
-            trigger_type=trigger_type,
-            trigger_ref=trigger_ref,
-            is_shadow=_SHADOW_ENABLED,
-            input_sha256=input_sha256,
-            timeout_at=timeout_at,
-        )
-        db.add(run)
+    run = AwoooPRunState(
+        run_id=run_id,
+        project_id=project_id,
+        agent_id=agent_id,
+        state="pending",
+        trace_id=trace_id,
+        trigger_type=trigger_type,
+        trigger_ref=trigger_ref,
+        is_shadow=_SHADOW_ENABLED,
+        input_sha256=input_sha256,
+        timeout_at=timeout_at,
+    )
 
-    # 寫入 idempotency 記錄
     if channel_type and provider_event_id:
-        await _register_idempotency(project_id, channel_type, provider_event_id, run_id)
+        lock_key = (
+            f"{_IDEMPOTENCY_LOCK_PREFIX}{project_id}:"
+            f"{channel_type}:{provider_event_id}"
+        )
+        async with get_db_context(project_id) as db:
+            await db.execute(
+                text("SELECT pg_advisory_xact_lock(hashtext(:lock_key))"),
+                {"lock_key": lock_key},
+            )
+            existing_result = await db.execute(
+                select(AwoooPRunIdempotency.run_id).where(
+                    AwoooPRunIdempotency.project_id == project_id,
+                    AwoooPRunIdempotency.channel_type == channel_type,
+                    AwoooPRunIdempotency.provider_event_id
+                    == provider_event_id,
+                )
+            )
+            existing_row = existing_result.fetchone()
+            if existing_row is not None:
+                selected_run_id = UUID(str(existing_row[0]))
+                is_duplicate = True
+            else:
+                db.add(run)
+                # These mapped classes intentionally have no ORM relationship,
+                # so SQLAlchemy cannot infer the foreign-key insert order. Keep
+                # both writes in this transaction, but materialize the run
+                # before its idempotency mapping.
+                await db.flush()
+                db.add(
+                    AwoooPRunIdempotency(
+                        project_id=project_id,
+                        channel_type=channel_type,
+                        provider_event_id=provider_event_id,
+                        run_id=run_id,
+                    )
+                )
+                await db.flush()
+                selected_run_id = run_id
+                is_duplicate = False
+
+        await _cache_idempotency(
+            project_id,
+            channel_type,
+            provider_event_id,
+            selected_run_id,
+        )
+        if is_duplicate:
+            logger.info(
+                "run_creation_idempotent",
+                project_id=project_id,
+                channel_type=channel_type,
+                provider_event_id=provider_event_id,
+                existing_run_id=str(selected_run_id),
+            )
+            return selected_run_id, True
+    else:
+        async with get_db_context(project_id) as db:
+            db.add(run)
+
+    if channel_type and provider_event_id:
+        # The atomic path can reach here only for the transaction winner.
+        assert selected_run_id == run_id and is_duplicate is False
+    else:
+        selected_run_id = run_id
+        is_duplicate = False
 
     logger.info(
         "run_created",
@@ -250,9 +303,10 @@ async def create_run(
         agent_id=agent_id,
         is_shadow=_SHADOW_ENABLED,
         trace_id=trace_id,
+        work_item_id=correlation.work_item_id,
         trigger_type=trigger_type,
     )
-    return run_id, False
+    return selected_run_id, is_duplicate
 
 
 # ─────────────────────────────────────────────────────────────────────────────
@@ -341,7 +395,7 @@ async def shadow_execute(run: AwoooPRunState) -> None:
         )
 
 
-async def get_run_status(run_id: uuid.UUID, project_id: str) -> dict[str, Any] | None:
+async def get_run_status(run_id: UUID, project_id: str) -> dict[str, Any] | None:
     """
     查詢單一 run 的 FSM 狀態。回傳 None 表示不存在。
     Router 層透過此 service 函數存取，不直接操作 DB。
@@ -358,6 +412,8 @@ async def get_run_status(run_id: uuid.UUID, project_id: str) -> dict[str, Any] |
     if run is None:
         return None
 
+    work_item_id = canonical_work_item_id(run.project_id, run.run_id)
+    trace_id = str(run.trace_id or "")
     return {
         "run_id": str(run.run_id),
         "project_id": run.project_id,
@@ -365,6 +421,12 @@ async def get_run_status(run_id: uuid.UUID, project_id: str) -> dict[str, Any] |
         "state": run.state,
         "is_shadow": run.is_shadow,
         "trace_id": run.trace_id,
+        "work_item_id": work_item_id,
+        "correlation_status": (
+            "canonical"
+            if is_canonical_traceparent(trace_id, run_id=run.run_id)
+            else "legacy_or_invalid_trace"
+        ),
         "attempt_count": run.attempt_count,
         "cost_usd": float(run.cost_usd),
         "step_count": run.step_count,