From 18b5ba25fa9c6aeaeb66260efe937ee79fe461b8 Mon Sep 17 00:00:00 2001 From: Your Name Date: Wed, 22 Jul 2026 16:23:11 +0800 Subject: [PATCH] fix(platform): bind AI runs to canonical correlation --- apps/api/src/api/v1/platform/runs.py | 27 +- apps/api/src/services/channel_hub.py | 39 ++- .../paid_provider_canary_authorization.py | 8 +- .../paid_provider_canary_gate5_run.py | 17 +- .../paid_provider_canary_validation.py | 10 +- apps/api/src/services/platform_runtime.py | 226 +++++++++------ apps/api/src/services/runtime_correlation.py | 117 ++++++++ .../test_platform_run_idempotency_race.py | 162 +++++++++++ .../test_channel_hub_grouped_alert_events.py | 74 ++++- ...test_paid_provider_canary_authorization.py | 10 +- .../test_paid_provider_canary_gate5_run.py | 153 ++++++++++- .../test_paid_provider_canary_validation.py | 56 +++- apps/api/tests/test_runtime_correlation.py | 257 ++++++++++++++++++ 13 files changed, 1051 insertions(+), 105 deletions(-) create mode 100644 apps/api/src/services/runtime_correlation.py create mode 100644 apps/api/tests/integration/test_platform_run_idempotency_race.py create mode 100644 apps/api/tests/test_runtime_correlation.py diff --git a/apps/api/src/api/v1/platform/runs.py b/apps/api/src/api/v1/platform/runs.py index 8d3179329..20334371d 100644 --- a/apps/api/src/api/v1/platform/runs.py +++ b/apps/api/src/api/v1/platform/runs.py @@ -24,7 +24,12 @@ from fastapi import APIRouter, HTTPException, status from pydantic import BaseModel, Field from src.services.audit_sink import write_audit -from src.services.platform_runtime import create_run +from src.services.platform_runtime import ( + create_run, +) +from src.services.platform_runtime import ( + get_run_status as read_platform_run_status, +) router = APIRouter() @@ -57,6 +62,9 @@ class CreateRunResponse(BaseModel): """POST /v1/platform/runs response""" run_id: str + trace_id: str + work_item_id: str + correlation_status: str is_duplicate: bool = Field(description="True = 冪等命中，返回既有 run_id") is_shadow: bool = Field(True, description="Phase 4 固定 True") message: str @@ -93,6 +101,17 @@ async def create_platform_run( provider_event_id=request.provider_event_id, timeout_seconds=request.timeout_seconds, ) + identity = await read_platform_run_status(run_id, request.project_id) + if ( + identity is None + or not identity.get("trace_id") + or not identity.get("work_item_id") + or ( + not is_duplicate + and identity.get("correlation_status") != "canonical" + ) + ): + raise RuntimeError("Run correlation readback missing") except Exception as exc: raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, @@ -110,11 +129,17 @@ async def create_platform_run( "trigger_type": request.trigger_type, "is_duplicate": is_duplicate, "is_shadow": True, + "trace_id": identity["trace_id"], + "work_item_id": identity["work_item_id"], + "correlation_status": identity["correlation_status"], }, ) return CreateRunResponse( run_id=str(run_id), + trace_id=str(identity["trace_id"]), + work_item_id=str(identity["work_item_id"]), + correlation_status=str(identity["correlation_status"]), is_duplicate=is_duplicate, is_shadow=True, message="Run 已接受（shadow mode）" if not is_duplicate else "冪等命中，返回既有 run_id", diff --git a/apps/api/src/services/channel_hub.py b/apps/api/src/services/channel_hub.py index 43db5b280..52af6948c 100644 --- a/apps/api/src/services/channel_hub.py +++ b/apps/api/src/services/channel_hub.py @@ -41,7 +41,13 @@ from sqlalchemy.ext.asyncio import AsyncSession from src.db.awooop_models import AwoooPRunState from src.services.audit_sink import _redact_string, sanitize -from src.services.platform_runtime import create_run +from src.services.platform_runtime import ( + create_run, +) +from src.services.platform_runtime import ( + get_run_status as read_platform_run_status, +) +from src.services.runtime_correlation import correlation_for_run logger = structlog.get_logger(__name__) @@ -160,17 +166,18 @@ async def ensure_completed_shadow_run( 資料。這些事件不應重新觸發 runtime，但需要 run_state 當 Console 的 聚合錨點；因此這裡建立的是已完成的 shadow run，不會被 worker pick up。 """ + correlation = correlation_for_run(project_id, run_id) result = await db.execute( text(""" INSERT INTO awooop_run_state ( run_id, project_id, agent_id, state, - trigger_type, trigger_ref, is_shadow, + trace_id, trigger_type, trigger_ref, is_shadow, input_sha256, attempt_count, max_attempts, cost_usd, step_count, created_at, completed_at, timeout_at ) VALUES ( :run_id, :project_id, :agent_id, 'completed', - :trigger_type, :trigger_ref, TRUE, + :trace_id, :trigger_type, :trigger_ref, TRUE, :input_sha256, 0, 3, 0.0000, 0, NOW(), NOW(), NOW() @@ -180,6 +187,7 @@ async def ensure_completed_shadow_run( """), { "run_id": run_id, + "trace_id": correlation.trace_id, "project_id": project_id, "agent_id": agent_id, "trigger_type": trigger_type, @@ -193,6 +201,8 @@ async def ensure_completed_shadow_run( "completed_shadow_run_created", project_id=project_id, run_id=str(run_id), + trace_id=correlation.trace_id, + work_item_id=correlation.work_item_id, agent_id=agent_id, trigger_type=trigger_type, ) @@ -303,7 +313,10 @@ async def mirror_inbound_event( ) ON CONFLICT (project_id, channel_type, provider_event_id) DO UPDATE SET is_duplicate = TRUE, - run_id = COALESCE(EXCLUDED.run_id, awooop_conversation_event.run_id), + run_id = COALESCE( + awooop_conversation_event.run_id, + EXCLUDED.run_id + ), content_redacted = COALESCE( awooop_conversation_event.content_redacted, EXCLUDED.content_redacted @@ -1209,6 +1222,20 @@ async def handle_telegram_inbound( channel_type="telegram", provider_event_id=message_id, ) + identity = await read_platform_run_status(run_id, project_id) + if ( + identity is None + or not identity.get("trace_id") + or not identity.get("work_item_id") + or (not is_duplicate and identity.get("correlation_status") != "canonical") + ): + raise RuntimeError("telegram_inbound_run_correlation_missing") + runtime_correlation = { + "run_id": str(run_id), + "trace_id": str(identity["trace_id"]), + "work_item_id": str(identity["work_item_id"]), + "correlation_status": str(identity["correlation_status"]), + } # Step 2: Mirror event（含 run_id） event_id = await mirror_inbound_event( @@ -1220,6 +1247,7 @@ async def handle_telegram_inbound( channel_chat_id=chat_id, content_type="text" if text else "callback_query", raw_content=text, + source_envelope={"runtime_correlation": runtime_correlation}, run_id=run_id, is_duplicate=is_duplicate, ) @@ -1243,5 +1271,8 @@ async def handle_telegram_inbound( return { "event_id": str(event_id), "run_id": str(run_id), + "trace_id": runtime_correlation["trace_id"], + "work_item_id": runtime_correlation["work_item_id"], + "correlation_status": runtime_correlation["correlation_status"], "is_duplicate": is_duplicate, } diff --git a/apps/api/src/services/paid_provider_canary_authorization.py b/apps/api/src/services/paid_provider_canary_authorization.py index 42530703b..f872decf2 100644 --- a/apps/api/src/services/paid_provider_canary_authorization.py +++ b/apps/api/src/services/paid_provider_canary_authorization.py @@ -16,7 +16,6 @@ from __future__ import annotations import hashlib import json -import re from dataclasses import dataclass from datetime import UTC, datetime, timedelta from typing import Any, Protocol @@ -31,6 +30,7 @@ from src.services.audit_sink import ( AWOOOP_AUDIT_SCHEMA_VERSION, audit_idempotency_key_sha256, ) +from src.services.runtime_correlation import is_canonical_traceparent logger = structlog.get_logger(__name__) @@ -47,7 +47,6 @@ PAID_CANARY_APPROVAL_AUDIT_IDEMPOTENCY_PREFIX = ( PAID_CANARY_START_ACTION = "claude_gemini_sanitized_sre_canary" PAID_CANARY_APPROVAL_TTL_SECONDS = 900 PAID_CANARY_PERCENT = 5 -_SAFE_TRACE_ID = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:@+-]{0,191}$") def paid_provider_canary_approval_audit_idempotency_key( @@ -371,8 +370,9 @@ async def verify_paid_provider_canary_authorization( expected_input_sha256 = paid_provider_canary_authorization_input_sha256() checks = { "run_id_exact": evidence.run_id == run_id, - "trace_id_present": bool( - _SAFE_TRACE_ID.fullmatch(str(evidence.trace_id or "").strip()) + "trace_id_run_bound": is_canonical_traceparent( + str(evidence.trace_id or ""), + run_id=evidence.run_id, ), "authorization_run_selected": paid_provider_canary_run_selected( evidence.run_id diff --git a/apps/api/src/services/paid_provider_canary_gate5_run.py b/apps/api/src/services/paid_provider_canary_gate5_run.py index f94794963..e50f3f221 100644 --- a/apps/api/src/services/paid_provider_canary_gate5_run.py +++ b/apps/api/src/services/paid_provider_canary_gate5_run.py @@ -34,6 +34,7 @@ from src.services.paid_provider_canary_authorization import ( paid_provider_canary_run_selected, ) from src.services.platform_runtime import _new_trace_id, _uuid7 +from src.services.runtime_correlation import is_canonical_traceparent logger = structlog.get_logger(__name__) @@ -95,6 +96,8 @@ def _utc_naive(value: datetime) -> datetime: def _record_from_run(run: AwoooPRunState) -> PaidProviderCanaryGate5RunRecord: if run.timeout_at is None or run.input_sha256 is None or run.trace_id is None: raise RuntimeError("paid_provider_canary_gate5_run_incomplete") + if not is_canonical_traceparent(str(run.trace_id), run_id=run.run_id): + raise RuntimeError("paid_provider_canary_gate5_run_trace_unbound") return PaidProviderCanaryGate5RunRecord( run_id=run.run_id, project_id=run.project_id, @@ -190,6 +193,10 @@ class _SqlPaidProviderCanaryGate5RunRepository: "authorization_run_selected": ( paid_provider_canary_run_selected(existing.run_id) ), + "trace_run_bound": is_canonical_traceparent( + str(existing.trace_id or ""), + run_id=existing.run_id, + ), } active_contract_errors = [ name @@ -254,6 +261,10 @@ def paid_provider_canary_approval_contract_errors( "authorization_run_selected": paid_provider_canary_run_selected( run.run_id ), + "trace_run_bound": is_canonical_traceparent( + str(run.trace_id or ""), + run_id=run.run_id, + ), "timeout_present": run.timeout_at is not None, "timeout_not_expired": bool( run.timeout_at is not None @@ -285,7 +296,7 @@ async def create_paid_provider_canary_gate5_run( authorization_input = paid_provider_canary_authorization_input() input_sha256 = paid_provider_canary_authorization_input_sha256() run_id = _new_selected_paid_canary_run_id() - trace_id = _new_trace_id() + trace_id = _new_trace_id(run_id) writer = repository or _SqlPaidProviderCanaryGate5RunRepository() record, is_duplicate = await writer.create_or_get( @@ -295,6 +306,10 @@ async def create_paid_provider_canary_gate5_run( timeout_at=timeout_at, input_sha256=input_sha256, ) + if not paid_provider_canary_run_selected(record.run_id): + raise RuntimeError("paid_provider_canary_gate5_run_not_selected") + if not is_canonical_traceparent(record.trace_id, run_id=record.run_id): + raise RuntimeError("paid_provider_canary_gate5_run_trace_unbound") write_creation_audit = audit_writer or write_audit await write_creation_audit( diff --git a/apps/api/src/services/paid_provider_canary_validation.py b/apps/api/src/services/paid_provider_canary_validation.py index c82222040..995881233 100644 --- a/apps/api/src/services/paid_provider_canary_validation.py +++ b/apps/api/src/services/paid_provider_canary_validation.py @@ -43,9 +43,9 @@ from src.services.paid_provider_canary_authorization import ( paid_provider_canary_bucket, paid_provider_canary_run_selected, ) +from src.services.runtime_correlation import is_canonical_traceparent _SAFE_RUN_REF = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{0,95}$") -_SAFE_TRACE_ID = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:@+-]{0,191}$") _WORK_ITEM_ID = "AIA-SRE-013" _ACTIVATION_LEASE_SECONDS = 300 _PAID_CANARY_MAX_OUTPUT_TOKENS = 512 @@ -461,6 +461,12 @@ async def _run_paid_provider_canary_validation_core( # The executable Gate 5 row is the controlled-apply run. A caller-supplied # source label must never mint a second run identity after approval. run_id = clean_authorization_ref + try: + durable_run_id = UUID(run_id) + except ValueError as exc: + raise RuntimeError( + "paid_canary_authorization_same_run_binding_failed" + ) from exc canary_percent = min( settings.CLAUDE_CANARY_PERCENT, settings.GEMINI_CANARY_PERCENT, @@ -473,7 +479,7 @@ async def _run_paid_provider_canary_validation_core( or authorization_receipt.get("work_item_id") != _WORK_ITEM_ID or not paid_provider_canary_run_selected(run_id) or canary_bucket >= canary_percent - or not _SAFE_TRACE_ID.fullmatch(trace_id) + or not is_canonical_traceparent(trace_id, run_id=durable_run_id) ): raise RuntimeError("paid_canary_authorization_same_run_binding_failed") activation_owner = f"{run_id}.attempt-{uuid4().hex[:12]}" diff --git a/apps/api/src/services/platform_runtime.py b/apps/api/src/services/platform_runtime.py index 6dae4a6c3..29f2c41e6 100644 --- a/apps/api/src/services/platform_runtime.py +++ b/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, diff --git a/apps/api/src/services/runtime_correlation.py b/apps/api/src/services/runtime_correlation.py new file mode 100644 index 000000000..85a3c2e4d --- /dev/null +++ b/apps/api/src/services/runtime_correlation.py @@ -0,0 +1,117 @@ +"""Canonical identity helpers for durable AwoooP runtime runs. + +The durable ``awooop_run_state.run_id`` is the root identity. Trace and work +item identifiers are derived from it so retries, API readback, channel mirrors, +and downstream receipts cannot silently invent unrelated correlation values. +""" + +from __future__ import annotations + +import hashlib +import re +import secrets +import time +from dataclasses import dataclass +from uuid import UUID + +_PROJECT_ID_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_.-]{0,63}$") +_TRACEPARENT_RE = re.compile( + r"^00-(?!0{32})[0-9a-f]{32}-(?!0{16})[0-9a-f]{16}-01$" +) + + +@dataclass(frozen=True, slots=True) +class RuntimeCorrelation: + """Public-safe identity shared by every node in one runtime run.""" + + run_id: UUID + trace_id: str + work_item_id: str + + def public_dict(self) -> dict[str, str]: + return { + "run_id": str(self.run_id), + "trace_id": self.trace_id, + "work_item_id": self.work_item_id, + } + + +def new_uuid7( + *, + timestamp_ms: int | None = None, + entropy: int | None = None, +) -> UUID: + """Return an RFC 9562 UUIDv7 with the RFC 4122 variant bits set. + + Optional inputs make the bit layout deterministic in focused tests. They + are not accepted from API callers. + """ + + observed_ms = ( + int(time.time_ns() // 1_000_000) + if timestamp_ms is None + else int(timestamp_ms) + ) + if not 0 <= observed_ms < (1 << 48): + raise ValueError("uuid7_timestamp_out_of_range") + + random_bits = secrets.randbits(74) if entropy is None else int(entropy) + if not 0 <= random_bits < (1 << 74): + raise ValueError("uuid7_entropy_out_of_range") + + rand_a = random_bits >> 62 + rand_b = random_bits & ((1 << 62) - 1) + value = ( + (observed_ms << 80) + | (0x7 << 76) + | (rand_a << 64) + | (0b10 << 62) + | rand_b + ) + return UUID(int=value) + + +def canonical_traceparent(run_id: UUID) -> str: + """Derive a stable W3C traceparent from one durable run identifier.""" + + trace_hex = hashlib.sha256( + f"awooop-runtime-trace:{run_id}".encode("utf-8") + ).hexdigest()[:32] + span_hex = hashlib.sha256( + f"awooop-runtime-root-span:{run_id}".encode("utf-8") + ).hexdigest()[:16] + return f"00-{trace_hex}-{span_hex}-01" + + +def is_canonical_traceparent(value: str, *, run_id: UUID | None = None) -> bool: + """Validate shape and, when supplied, binding to the durable run.""" + + candidate = str(value or "").strip() + if _TRACEPARENT_RE.fullmatch(candidate) is None: + return False + return run_id is None or candidate == canonical_traceparent(run_id) + + +def canonical_work_item_id(project_id: str, run_id: UUID) -> str: + """Return the only generic work-item identity for one platform run.""" + + project = str(project_id or "").strip() + if _PROJECT_ID_RE.fullmatch(project) is None: + raise ValueError("runtime_correlation_project_id_invalid") + return f"platform-run:{project}:{run_id}" + + +def correlation_for_run(project_id: str, run_id: UUID) -> RuntimeCorrelation: + """Build the canonical public-safe correlation projection for a run.""" + + return RuntimeCorrelation( + run_id=run_id, + trace_id=canonical_traceparent(run_id), + work_item_id=canonical_work_item_id(project_id, run_id), + ) + + +def new_runtime_correlation(project_id: str) -> RuntimeCorrelation: + """Allocate one new run root and all identifiers derived from it.""" + + return correlation_for_run(project_id, new_uuid7()) diff --git a/apps/api/tests/integration/test_platform_run_idempotency_race.py b/apps/api/tests/integration/test_platform_run_idempotency_race.py new file mode 100644 index 000000000..0abbaacfd --- /dev/null +++ b/apps/api/tests/integration/test_platform_run_idempotency_race.py @@ -0,0 +1,162 @@ +from __future__ import annotations + +import asyncio +import os +from contextlib import asynccontextmanager + +import pytest +from sqlalchemy import make_url, text +from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine +from sqlalchemy.pool import NullPool + +from src.services import platform_runtime + +_TEST_DATABASE_ENV = "AWOOOI_RUNTIME_IDENTITY_TEST_DATABASE_URL" +_TEST_DATABASE_NAME = "awoooi_runtime_identity_test" +_LOCAL_DATABASE_HOSTS = frozenset({"127.0.0.1", "localhost"}) + + +def _validated_test_database_url() -> str: + raw_url = os.getenv(_TEST_DATABASE_ENV, "").strip() + if not raw_url: + pytest.skip(f"{_TEST_DATABASE_ENV} is required for the real PostgreSQL race test") + + parsed = make_url(raw_url) + if ( + parsed.get_backend_name() != "postgresql" + or parsed.get_driver_name() != "asyncpg" + or parsed.host not in _LOCAL_DATABASE_HOSTS + or parsed.database != _TEST_DATABASE_NAME + ): + pytest.fail( + f"{_TEST_DATABASE_ENV} must use postgresql+asyncpg on localhost " + f"with database {_TEST_DATABASE_NAME!r}" + ) + return raw_url + + +_RUN_STATE_DDL = """ +CREATE TABLE awooop_run_state ( + run_id UUID PRIMARY KEY, + project_id VARCHAR(64) NOT NULL, + agent_id VARCHAR(128) NOT NULL, + state VARCHAR(32) NOT NULL DEFAULT 'pending', + lease_until TIMESTAMP, + next_attempt_at TIMESTAMP, + heartbeat_at TIMESTAMP, + worker_id VARCHAR(128), + attempt_count SMALLINT NOT NULL DEFAULT 0, + max_attempts SMALLINT NOT NULL DEFAULT 3, + trace_id VARCHAR(128), + trigger_type VARCHAR(32), + trigger_ref VARCHAR(256), + is_shadow BOOLEAN NOT NULL DEFAULT TRUE, + input_sha256 VARCHAR(64), + output_sha256 VARCHAR(64), + cost_usd NUMERIC(10, 4) NOT NULL DEFAULT 0, + step_count SMALLINT NOT NULL DEFAULT 0, + error_code VARCHAR(64), + error_detail TEXT, + created_at TIMESTAMP NOT NULL DEFAULT NOW(), + started_at TIMESTAMP, + completed_at TIMESTAMP, + timeout_at TIMESTAMP +) +""" + +_RUN_IDEMPOTENCY_DDL = """ +CREATE TABLE awooop_run_idempotency ( + idempotency_id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + project_id VARCHAR(64) NOT NULL, + channel_type VARCHAR(32) NOT NULL, + provider_event_id VARCHAR(256) NOT NULL, + run_id UUID NOT NULL REFERENCES awooop_run_state(run_id), + created_at TIMESTAMP NOT NULL DEFAULT NOW(), + CONSTRAINT uix_run_idempotency_key + UNIQUE (project_id, channel_type, provider_event_id) +) +""" + + +@pytest.mark.asyncio +async def test_concurrent_provider_event_has_one_postgresql_winner( + monkeypatch: pytest.MonkeyPatch, +) -> None: + engine = create_async_engine(_validated_test_database_url(), poolclass=NullPool) + factory = async_sessionmaker(engine, expire_on_commit=False, autoflush=False) + tables_created = False + + @asynccontextmanager + async def test_db_context(project_id: str): + assert project_id == "awoooi" + async with factory() as session: + try: + yield session + await session.commit() + except Exception: + await session.rollback() + raise + + async def skip_redis_cache(*_args: object) -> None: + return None + + try: + async with engine.begin() as connection: + existing_tables = ( + await connection.execute( + text( + "SELECT to_regclass('awooop_run_state'), " + "to_regclass('awooop_run_idempotency')" + ) + ) + ).one() + if any(existing_tables): + pytest.fail( + "dedicated runtime identity test database must not contain " + "pre-existing run tables" + ) + await connection.execute(text(_RUN_STATE_DDL)) + await connection.execute(text(_RUN_IDEMPOTENCY_DDL)) + tables_created = True + + monkeypatch.setattr(platform_runtime, "get_db_context", test_db_context) + monkeypatch.setattr(platform_runtime, "_cache_idempotency", skip_redis_cache) + + first, second = await asyncio.gather( + platform_runtime.create_run( + project_id="awoooi", + agent_id="runtime-agent", + trigger_type="channel_event", + channel_type="telegram", + provider_event_id="same-real-postgresql-event", + ), + platform_runtime.create_run( + project_id="awoooi", + agent_id="runtime-agent", + trigger_type="channel_event", + channel_type="telegram", + provider_event_id="same-real-postgresql-event", + ), + ) + + async with engine.connect() as connection: + run_count = await connection.scalar( + text("SELECT COUNT(*) FROM awooop_run_state") + ) + mapping_count = await connection.scalar( + text("SELECT COUNT(*) FROM awooop_run_idempotency") + ) + durable_run_id = await connection.scalar( + text("SELECT run_id FROM awooop_run_idempotency") + ) + + assert first[0] == second[0] == durable_run_id + assert sorted((first[1], second[1])) == [False, True] + assert run_count == 1 + assert mapping_count == 1 + finally: + if tables_created: + async with engine.begin() as connection: + await connection.execute(text("DROP TABLE awooop_run_idempotency")) + await connection.execute(text("DROP TABLE awooop_run_state")) + await engine.dispose() diff --git a/apps/api/tests/test_channel_hub_grouped_alert_events.py b/apps/api/tests/test_channel_hub_grouped_alert_events.py index d7f628ede..33d58443c 100644 --- a/apps/api/tests/test_channel_hub_grouped_alert_events.py +++ b/apps/api/tests/test_channel_hub_grouped_alert_events.py @@ -2,8 +2,11 @@ from __future__ import annotations import inspect import json -from uuid import uuid4 +from uuid import UUID, uuid4 +import pytest + +from src.services import channel_hub from src.services.channel_hub import ( _db_timestamp_now, _send_telegram_interim, @@ -17,9 +20,11 @@ from src.services.channel_hub import ( format_alertmanager_event_content, format_grouped_alert_digest_text, format_grouped_alert_event_content, + handle_telegram_inbound, mirror_inbound_event, record_outbound_message, ) +from src.services.runtime_correlation import correlation_for_run from src.services.telegram_gateway import _telegram_destination_binding _FAKE_SECRET_SHAPED_VALUE = "1234567890:" + "abcdefghijklmnopqrstuvwxyzABCDEFGH" @@ -357,6 +362,69 @@ async def test_completed_shadow_run_sets_run_state_not_null_defaults() -> None: assert "0, 3, 0.0000, 0" in session.statement assert session.params["project_id"] == "awoooi" assert session.params["run_id"] == run_id + correlation = correlation_for_run("awoooi", run_id) + assert session.params["trace_id"] == correlation.trace_id + assert "trace_id, trigger_type" in session.statement + + +@pytest.mark.asyncio +async def test_telegram_inbound_persists_and_returns_same_run_correlation( + monkeypatch: pytest.MonkeyPatch, +) -> None: + run_id = UUID("019f88d5-f6c9-7a18-b764-df0df0cbe222") + identity = correlation_for_run("awoooi", run_id) + mirrored: dict[str, object] = {} + + async def fake_create_run(**_kwargs: object) -> tuple[UUID, bool]: + return run_id, False + + async def fake_read_run_status( + received_run_id: UUID, + project_id: str, + ) -> dict[str, object]: + assert received_run_id == run_id + assert project_id == "awoooi" + return { + **identity.public_dict(), + "correlation_status": "canonical", + } + + async def fake_mirror(_db: object, **kwargs: object) -> UUID: + mirrored.update(kwargs) + return UUID("019f88d5-f6c9-7a18-b764-df0df0cbe223") + + async def fake_schedule(**_kwargs: object) -> None: + return None + + monkeypatch.setattr(channel_hub, "create_run", fake_create_run) + monkeypatch.setattr( + channel_hub, + "read_platform_run_status", + fake_read_run_status, + ) + monkeypatch.setattr(channel_hub, "mirror_inbound_event", fake_mirror) + monkeypatch.setattr(channel_hub, "schedule_interim_feedback", fake_schedule) + + result = await handle_telegram_inbound( + object(), # type: ignore[arg-type] + project_id="awoooi", + agent_id="openclaw-sre", + message_id="42", + user_id="7", + chat_id="-1007", + text="status", + ) + + assert result["run_id"] == str(run_id) + assert result["trace_id"] == identity.trace_id + assert result["work_item_id"] == identity.work_item_id + assert result["correlation_status"] == "canonical" + source_envelope = mirrored["source_envelope"] + assert isinstance(source_envelope, dict) + assert source_envelope["runtime_correlation"] == { + **identity.public_dict(), + "correlation_status": "canonical", + } async def test_mirror_inbound_event_writes_redacted_replay_fields() -> None: @@ -385,6 +453,10 @@ async def test_mirror_inbound_event_writes_redacted_replay_fields() -> None: assert "content_redacted" in session.statement assert "source_envelope" in session.statement + assert ( + "COALESCE(\n awooop_conversation_event.run_id," + in session.statement + ) assert session.params["redaction_version"] == "audit_sink_v1" assert len(json.loads(session.params["source_envelope"])["content_sha256"]) == 64 assert "1234567890:" not in session.params["content_redacted"] diff --git a/apps/api/tests/test_paid_provider_canary_authorization.py b/apps/api/tests/test_paid_provider_canary_authorization.py index 8964167da..0c997859f 100644 --- a/apps/api/tests/test_paid_provider_canary_authorization.py +++ b/apps/api/tests/test_paid_provider_canary_authorization.py @@ -22,9 +22,10 @@ from src.services.paid_provider_canary_authorization import ( # noqa: E402 require_paid_provider_canary_authorization, verify_paid_provider_canary_authorization, ) +from src.services.runtime_correlation import canonical_traceparent # noqa: E402 _SELECTED_RUN_ID = UUID("267a3d26-3c29-470f-8fe3-388a2f408f39") -_TRACE_ID = "00-0123456789abcdef0123456789abcdef-0123456789abcdef-01" +_TRACE_ID = canonical_traceparent(_SELECTED_RUN_ID) class _Reader: @@ -56,9 +57,10 @@ def _evidence( now: datetime, run_id: UUID | None = None, ) -> PaidProviderCanaryAuthorizationEvidence: + selected_run_id = run_id or _SELECTED_RUN_ID return PaidProviderCanaryAuthorizationEvidence( - run_id=run_id or _SELECTED_RUN_ID, - trace_id=_TRACE_ID, + run_id=selected_run_id, + trace_id=canonical_traceparent(selected_run_id), project_id=PAID_CANARY_PROJECT_ID, agent_id=PAID_CANARY_AGENT_ID, state="running", @@ -167,7 +169,7 @@ async def test_authorization_ref_must_be_canonical_uuid_without_db_read() -> Non ("trigger_type", "schedule", "trigger_type_exact"), ("trigger_ref", "paid-provider-canary:OTHER", "trigger_ref_exact"), ("input_sha256", "0" * 64, "input_sha256_exact"), - ("trace_id", None, "trace_id_present"), + ("trace_id", None, "trace_id_run_bound"), ("approval_step_count", 0, "approval_step_exactly_once"), ("approval_step_status", "failed", "approval_step_success"), ("approval_step_was_blocked", True, "approval_step_not_blocked"), diff --git a/apps/api/tests/test_paid_provider_canary_gate5_run.py b/apps/api/tests/test_paid_provider_canary_gate5_run.py index 71d4e9331..2eb5a3e68 100644 --- a/apps/api/tests/test_paid_provider_canary_gate5_run.py +++ b/apps/api/tests/test_paid_provider_canary_gate5_run.py @@ -3,7 +3,7 @@ from __future__ import annotations import os from contextlib import asynccontextmanager from datetime import UTC, datetime, timedelta -from uuid import UUID +from uuid import RFC_4122, UUID import pytest from fastapi import HTTPException @@ -25,6 +25,7 @@ from src.services.paid_provider_canary_authorization import ( # noqa: E402 PAID_CANARY_TRIGGER_REF, PAID_CANARY_TRIGGER_TYPE, paid_provider_canary_authorization_input_sha256, + paid_provider_canary_run_selected, ) from src.services.paid_provider_canary_gate5_run import ( # noqa: E402 PAID_CANARY_AUTHORIZATION_SCOPE, @@ -32,14 +33,28 @@ from src.services.paid_provider_canary_gate5_run import ( # noqa: E402 create_paid_provider_canary_gate5_run, paid_provider_canary_approval_contract_errors, ) +from src.services.runtime_correlation import ( # noqa: E402 + canonical_traceparent, + is_canonical_traceparent, +) _SELECTED_RUN_ID = UUID("267a3d26-3c29-470f-8fe3-388a2f408f39") +_UNSELECTED_RUN_ID = UUID("00000000-0000-4000-8000-000000000001") class _Repository: - def __init__(self, *, duplicate: bool = False, state: str = "waiting_approval") -> None: + def __init__( + self, + *, + duplicate: bool = False, + state: str = "waiting_approval", + mismatched_duplicate_trace: bool = False, + unselected_duplicate_run: bool = False, + ) -> None: self.duplicate = duplicate self.state = state + self.mismatched_duplicate_trace = mismatched_duplicate_trace + self.unselected_duplicate_run = unselected_duplicate_run self.calls: list[dict[str, object]] = [] async def create_or_get(self, **kwargs: object) -> tuple[ @@ -55,8 +70,15 @@ class _Repository: assert isinstance(timeout_at, datetime) assert isinstance(input_sha256, str) assert isinstance(trace_id, str) + assert is_canonical_traceparent(trace_id, run_id=run_id) is True if self.duplicate: - run_id = _SELECTED_RUN_ID + run_id = ( + _UNSELECTED_RUN_ID + if self.unselected_duplicate_run + else _SELECTED_RUN_ID + ) + if not self.mismatched_duplicate_trace: + trace_id = canonical_traceparent(run_id) return ( PaidProviderCanaryGate5RunRecord( run_id=run_id, @@ -160,6 +182,45 @@ async def test_duplicate_active_authorization_is_reused_without_scope_expansion( ) +@pytest.mark.asyncio +async def test_duplicate_active_authorization_rejects_unbound_trace() -> None: + repository = _Repository( + duplicate=True, + state="running", + mismatched_duplicate_trace=True, + ) + + with pytest.raises( + RuntimeError, + match="paid_provider_canary_gate5_run_trace_unbound", + ): + await create_paid_provider_canary_gate5_run( + operator_id="ops@example.com", + repository=repository, + now=datetime(2026, 7, 16, 9, 0, tzinfo=UTC), + ) + + +@pytest.mark.asyncio +async def test_injected_repository_rejects_unselected_execution_run() -> None: + assert paid_provider_canary_run_selected(_UNSELECTED_RUN_ID) is False + repository = _Repository( + duplicate=True, + state="running", + unselected_duplicate_run=True, + ) + + with pytest.raises( + RuntimeError, + match="paid_provider_canary_gate5_run_not_selected", + ): + await create_paid_provider_canary_gate5_run( + operator_id="ops@example.com", + repository=repository, + now=datetime(2026, 7, 16, 9, 0, tzinfo=UTC), + ) + + def _run( *, now: datetime, @@ -171,7 +232,7 @@ def _run( project_id=PAID_CANARY_PROJECT_ID, agent_id=agent_id, state="waiting_approval", - trace_id="00-0123456789abcdef0123456789abcdef-0123456789abcdef-01", + trace_id=canonical_traceparent(_SELECTED_RUN_ID), trigger_type=PAID_CANARY_TRIGGER_TYPE, trigger_ref=trigger_ref, is_shadow=False, @@ -473,21 +534,51 @@ async def test_generic_runtime_service_persists_shadow_true( assert len(captured) == 1 assert captured[0].is_shadow is True assert captured[0].state == "pending" + assert captured[0].run_id.version == 7 + assert captured[0].run_id.variant == RFC_4122 + assert is_canonical_traceparent( + str(captured[0].trace_id), + run_id=captured[0].run_id, + ) is True @pytest.mark.asyncio async def test_generic_public_run_creation_remains_shadow_only( monkeypatch: pytest.MonkeyPatch, ) -> None: + run_id = UUID("018f2d04-4c37-7a18-b764-df0df0cbe222") + async def fake_create_run(**_: object) -> tuple[UUID, bool]: - return UUID("018f2d04-4c37-7a18-b764-df0df0cbe222"), False + return run_id, False + + async def fake_read_run_status( + received_run_id: UUID, + project_id: str, + ) -> dict[str, object]: + assert received_run_id == run_id + assert project_id == "awoooi" + return { + "trace_id": "00-0123456789abcdef0123456789abcdef-0123456789abcdef-01", + "work_item_id": f"platform-run:awoooi:{run_id}", + "correlation_status": "canonical", + } async def fake_write_audit(**kwargs: object) -> None: details = kwargs["details"] assert isinstance(details, dict) assert details["is_shadow"] is True + assert details["trace_id"] == ( + "00-0123456789abcdef0123456789abcdef-0123456789abcdef-01" + ) + assert details["work_item_id"] == f"platform-run:awoooi:{run_id}" + assert details["correlation_status"] == "canonical" monkeypatch.setattr(public_runs, "create_run", fake_create_run) + monkeypatch.setattr( + public_runs, + "read_platform_run_status", + fake_read_run_status, + ) monkeypatch.setattr(public_runs, "write_audit", fake_write_audit) response = await public_runs.create_platform_run( @@ -499,4 +590,56 @@ async def test_generic_public_run_creation_remains_shadow_only( ) assert response.is_shadow is True + assert response.trace_id.startswith("00-") + assert response.work_item_id == f"platform-run:awoooi:{run_id}" + assert response.correlation_status == "canonical" assert response.message == "Run 已接受（shadow mode）" + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "identity_readback", + [ + None, + { + "trace_id": "legacy-trace", + "work_item_id": ( + "platform-run:awoooi:019f88d5-f6c9-7a18-b764-df0df0cbe222" + ), + "correlation_status": "legacy_or_invalid_trace", + }, + ], +) +async def test_generic_public_run_creation_fails_closed_without_canonical_identity_readback( + monkeypatch: pytest.MonkeyPatch, + identity_readback: dict[str, str] | None, +) -> None: + run_id = UUID("019f88d5-f6c9-7a18-b764-df0df0cbe222") + + async def fake_create_run(**_: object) -> tuple[UUID, bool]: + return run_id, False + + async def fake_read_run_status( + _run_id: UUID, + _project_id: str, + ) -> dict[str, str] | None: + return identity_readback + + monkeypatch.setattr(public_runs, "create_run", fake_create_run) + monkeypatch.setattr( + public_runs, + "read_platform_run_status", + fake_read_run_status, + ) + + with pytest.raises(HTTPException) as exc_info: + await public_runs.create_platform_run( + public_runs.CreateRunRequest( + project_id="awoooi", + agent_id="generic-agent", + trigger_type="api", + ) + ) + + assert exc_info.value.status_code == 500 + assert "correlation readback missing" in str(exc_info.value.detail) diff --git a/apps/api/tests/test_paid_provider_canary_validation.py b/apps/api/tests/test_paid_provider_canary_validation.py index 724bf5ab2..bf36085b5 100644 --- a/apps/api/tests/test_paid_provider_canary_validation.py +++ b/apps/api/tests/test_paid_provider_canary_validation.py @@ -19,9 +19,10 @@ from src.services.paid_provider_canary_validation import ( run_paid_provider_canary_validation, select_claude_canary_run_id, ) +from src.services.runtime_correlation import canonical_traceparent _AUTHORIZATION_REF = "267a3d26-3c29-470f-8fe3-388a2f408f39" -_AUTHORIZATION_TRACE_ID = "00-0123456789abcdef0123456789abcdef-0123456789abcdef-01" +_AUTHORIZATION_TRACE_ID = canonical_traceparent(UUID(_AUTHORIZATION_REF)) def test_ollama_canary_uses_bounded_production_diagnose_timeout( @@ -592,6 +593,59 @@ async def test_optional_terminal_aol_projection_cannot_reopen_terminal_run() -> assert record_id is None +@pytest.mark.asyncio +async def test_injected_authorization_receipt_rejects_trace_unbound_to_run( + monkeypatch, +) -> None: + from src.services import paid_provider_canary_validation as module + + monkeypatch.setattr( + module, + "settings", + SimpleNamespace( + CLAUDE_EXECUTION_MODE="canary", + CLAUDE_CANARY_PERCENT=5, + GEMINI_EXECUTION_MODE="canary", + GEMINI_CANARY_PERCENT=5, + ALERT_OLLAMA_MODEL="qwen3:14b", + ), + ) + execution_claim_attempted = False + + async def _unexpected_claim(_run_id: str, _owner: str) -> bool: + nonlocal execution_claim_attempted + execution_claim_attempted = True + return True + + monkeypatch.setattr( + module, + "_acquire_paid_canary_execution_claim", + _unexpected_claim, + ) + unbound_trace = canonical_traceparent( + UUID("267a3d26-3c29-470f-8fe3-388a2f408f3a") + ) + assert unbound_trace != _AUTHORIZATION_TRACE_ID + + with pytest.raises( + RuntimeError, + match="paid_canary_authorization_same_run_binding_failed", + ): + await module._run_paid_provider_canary_validation_core( + run_ref="source-sha-unbound-trace", + operator_id="test-operator", + authorization_ref=_AUTHORIZATION_REF, + verified_authorization_receipt={ + "authorization_ref": _AUTHORIZATION_REF, + "run_id": _AUTHORIZATION_REF, + "work_item_id": "AIA-SRE-013", + "trace_id": unbound_trace, + }, + ) + + assert execution_claim_attempted is False + + @pytest.mark.asyncio async def test_durable_authorization_blocks_before_receipt_or_provider_call( monkeypatch, diff --git a/apps/api/tests/test_runtime_correlation.py b/apps/api/tests/test_runtime_correlation.py new file mode 100644 index 000000000..6b8aaf29a --- /dev/null +++ b/apps/api/tests/test_runtime_correlation.py @@ -0,0 +1,257 @@ +from __future__ import annotations + +import asyncio +from contextlib import asynccontextmanager +from datetime import UTC, datetime +from decimal import Decimal +from types import SimpleNamespace +from uuid import RFC_4122, UUID + +import pytest + +from src.db.awooop_models import AwoooPRunIdempotency, AwoooPRunState +from src.services import platform_runtime +from src.services.runtime_correlation import ( + canonical_traceparent, + canonical_work_item_id, + correlation_for_run, + is_canonical_traceparent, + new_uuid7, +) + + +def test_uuid7_has_rfc_version_variant_and_timestamp_layout() -> None: + timestamp_ms = 1_753_155_200_123 + generated = new_uuid7(timestamp_ms=timestamp_ms, entropy=(1 << 74) - 1) + + assert generated.version == 7 + assert generated.variant == RFC_4122 + assert generated.int >> 80 == timestamp_ms + + +@pytest.mark.parametrize( + ("timestamp_ms", "entropy"), + [(-1, 0), (1 << 48, 0), (0, -1), (0, 1 << 74)], +) +def test_uuid7_rejects_out_of_range_inputs( + timestamp_ms: int, + entropy: int, +) -> None: + with pytest.raises(ValueError): + new_uuid7(timestamp_ms=timestamp_ms, entropy=entropy) + + +def test_runtime_correlation_is_stable_and_bound_to_run() -> None: + run_id = UUID("019f88d5-f6c9-7a18-b764-df0df0cbe222") + correlation = correlation_for_run("awoooi", run_id) + + assert correlation.run_id == run_id + assert correlation.trace_id == canonical_traceparent(run_id) + assert correlation.work_item_id == f"platform-run:awoooi:{run_id}" + assert is_canonical_traceparent(correlation.trace_id, run_id=run_id) is True + assert ( + is_canonical_traceparent( + correlation.trace_id, + run_id=UUID("019f88d5-f6c9-7a18-b764-df0df0cbe223"), + ) + is False + ) + + +def test_runtime_correlation_rejects_invalid_project_id() -> None: + with pytest.raises(ValueError, match="project_id_invalid"): + canonical_work_item_id( + "awoooi:forged", + UUID("019f88d5-f6c9-7a18-b764-df0df0cbe222"), + ) + + +@pytest.mark.asyncio +async def test_run_status_exposes_canonical_identity_readback( + monkeypatch: pytest.MonkeyPatch, +) -> None: + run_id = UUID("019f88d5-f6c9-7a18-b764-df0df0cbe222") + trace_id = canonical_traceparent(run_id) + row = SimpleNamespace( + run_id=run_id, + project_id="awoooi", + agent_id="runtime-agent", + state="running", + is_shadow=True, + trace_id=trace_id, + attempt_count=1, + cost_usd=Decimal("0.0000"), + step_count=2, + error_code=None, + created_at=datetime(2026, 7, 22, tzinfo=UTC), + started_at=datetime(2026, 7, 22, tzinfo=UTC), + completed_at=None, + ) + + class _Result: + def scalar_one_or_none(self) -> SimpleNamespace: + return row + + class _Db: + async def execute(self, _statement: object) -> _Result: + return _Result() + + @asynccontextmanager + async def fake_db_context(project_id: str): + assert project_id == "awoooi" + yield _Db() + + monkeypatch.setattr(platform_runtime, "get_db_context", fake_db_context) + + payload = await platform_runtime.get_run_status(run_id, "awoooi") + + assert payload is not None + assert payload["run_id"] == str(run_id) + assert payload["trace_id"] == trace_id + assert payload["work_item_id"] == f"platform-run:awoooi:{run_id}" + assert payload["correlation_status"] == "canonical" + + +@pytest.mark.asyncio +async def test_run_status_marks_legacy_unbound_trace_without_rewriting_it( + monkeypatch: pytest.MonkeyPatch, +) -> None: + run_id = UUID("019f88d5-f6c9-7a18-b764-df0df0cbe222") + row = SimpleNamespace( + run_id=run_id, + project_id="awoooi", + agent_id="legacy-agent", + state="completed", + is_shadow=True, + trace_id="legacy-trace", + attempt_count=1, + cost_usd=Decimal("0.0000"), + step_count=0, + error_code=None, + created_at=None, + started_at=None, + completed_at=None, + ) + + class _Result: + def scalar_one_or_none(self) -> SimpleNamespace: + return row + + class _Db: + async def execute(self, _statement: object) -> _Result: + return _Result() + + @asynccontextmanager + async def fake_db_context(_project_id: str): + yield _Db() + + monkeypatch.setattr(platform_runtime, "get_db_context", fake_db_context) + + payload = await platform_runtime.get_run_status(run_id, "awoooi") + + assert payload is not None + assert payload["trace_id"] == "legacy-trace" + assert payload["correlation_status"] == "legacy_or_invalid_trace" + + +@pytest.mark.asyncio +async def test_concurrent_provider_event_creation_returns_one_durable_winner( + monkeypatch: pytest.MonkeyPatch, +) -> None: + class _Result: + def __init__(self, row: tuple[UUID] | None = None) -> None: + self._row = row + + def fetchone(self) -> tuple[UUID] | None: + return self._row + + class _Store: + def __init__(self) -> None: + self.claim_lock = asyncio.Lock() + self.mapping: UUID | None = None + self.runs: dict[UUID, AwoooPRunState] = {} + self.lock_claim_count = 0 + + store = _Store() + + class _Db: + def __init__(self) -> None: + self.lock_held = False + self.pending_run: AwoooPRunState | None = None + self.pending_mapping: AwoooPRunIdempotency | None = None + + async def execute( + self, + statement: object, + _params: dict[str, str] | None = None, + ) -> _Result: + sql = str(statement) + if "pg_advisory_xact_lock" in sql: + await store.claim_lock.acquire() + self.lock_held = True + store.lock_claim_count += 1 + await asyncio.sleep(0) + return _Result() + if "awooop_run_idempotency" in sql: + return _Result( + (store.mapping,) if store.mapping is not None else None + ) + raise AssertionError(f"unexpected statement: {sql}") + + def add(self, row: object) -> None: + if isinstance(row, AwoooPRunState): + self.pending_run = row + elif isinstance(row, AwoooPRunIdempotency): + self.pending_mapping = row + else: + raise AssertionError(f"unexpected row: {type(row).__name__}") + + async def flush(self) -> None: + assert self.lock_held is True + assert self.pending_run is not None + if self.pending_mapping is None: + assert store.runs == {} + store.runs[self.pending_run.run_id] = self.pending_run + return + assert store.mapping is None + assert self.pending_mapping.run_id in store.runs + store.mapping = self.pending_mapping.run_id + + @asynccontextmanager + async def fake_db_context(project_id: str): + assert project_id == "awoooi" + db = _Db() + try: + yield db + finally: + if db.lock_held: + store.claim_lock.release() + + async def fake_cache(*_args: object) -> None: + return None + + monkeypatch.setattr(platform_runtime, "get_db_context", fake_db_context) + monkeypatch.setattr(platform_runtime, "_cache_idempotency", fake_cache) + + first, second = await asyncio.gather( + platform_runtime.create_run( + project_id="awoooi", + agent_id="runtime-agent", + trigger_type="channel_event", + channel_type="telegram", + provider_event_id="same-provider-event", + ), + platform_runtime.create_run( + project_id="awoooi", + agent_id="runtime-agent", + trigger_type="channel_event", + channel_type="telegram", + provider_event_id="same-provider-event", + ), + ) + + assert first[0] == second[0] == store.mapping + assert sorted((first[1], second[1])) == [False, True] + assert list(store.runs) == [store.mapping] + assert next(iter(store.runs.values())).timeout_at.tzinfo is None + assert store.lock_claim_count == 2