434 lines
16 KiB
Python
434 lines
16 KiB
Python
"""
|
||
Platform Runtime（Shadow Mode Shell）
|
||
======================================
|
||
AwoooP Phase 4: 第一個 runtime shell，只跑 shadow，不改 legacy 行為（ADR-106）
|
||
2026-05-04 ogt + Claude Sonnet 4.6
|
||
|
||
Shadow Mode 保證：
|
||
1. 0 user-visible response（不發送 Telegram/Slack 任何訊息）
|
||
2. 0 destructive tool call（is_destructive=true 的工具全部攔截）
|
||
3. 所有執行記錄寫入 awooop_run_state + step_journal（可觀測）
|
||
4. budget_service hard kill 同樣作用（防止 shadow 跑出超額費用）
|
||
|
||
Idempotency（ADR-114）：
|
||
(project_id, channel_type, provider_event_id) 複合唯一
|
||
Redis NX 先攔（快），PG constraint 最後防（準確）
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import hashlib
|
||
import json
|
||
from datetime import datetime, timedelta, timezone
|
||
from typing import Any
|
||
from uuid import UUID
|
||
|
||
import structlog
|
||
from sqlalchemy import select, text
|
||
|
||
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,
|
||
correlation_readback_for_run,
|
||
new_runtime_correlation,
|
||
new_uuid7,
|
||
)
|
||
|
||
logger = structlog.get_logger(__name__)
|
||
|
||
# Shadow mode 設定
|
||
_SHADOW_ENABLED = True # Phase 4 固定 True；Phase 6+ 由 migration_mode 控制
|
||
_DESTRUCTIVE_TOOL_KEYWORDS = frozenset({
|
||
"delete", "drop", "remove", "kill", "terminate", "destroy",
|
||
"rollback", "revert", "patch", "apply", "exec", "execute",
|
||
"restart", "scale", "cordon", "drain",
|
||
})
|
||
|
||
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
# UUID v7（時間有序）
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
|
||
def _uuid7() -> UUID:
|
||
"""Compatibility wrapper for the canonical RFC 9562 generator."""
|
||
|
||
return new_uuid7()
|
||
|
||
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
# W3C traceparent 生成
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
|
||
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())
|
||
|
||
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
# Idempotency
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
|
||
_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 | None:
|
||
"""
|
||
檢查 (project_id, channel_type, provider_event_id) 是否已有對應 run_id。
|
||
|
||
Layer 1：Redis NX（快速攔截，TTL 24h）
|
||
Layer 2：PostgreSQL awooop_run_idempotency（準確）
|
||
|
||
Returns: 既有 run_id，或 None（尚未處理）
|
||
"""
|
||
idem_key = f"{_IDEMPOTENCY_REDIS_PREFIX}{project_id}:{channel_type}:{provider_event_id}"
|
||
|
||
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
|
||
cached_run_id = UUID(run_id_str)
|
||
except Exception as exc:
|
||
logger.warning("idempotency_redis_check_failed", error=str(exc))
|
||
|
||
# PostgreSQL is the only authoritative mapping.
|
||
try:
|
||
async with get_db_context(project_id) as db:
|
||
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,
|
||
)
|
||
)
|
||
row = result.fetchone()
|
||
if row:
|
||
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 _cache_idempotency(
|
||
project_id: str,
|
||
channel_type: str,
|
||
provider_event_id: str,
|
||
run_id: UUID,
|
||
) -> None:
|
||
"""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)
|
||
|
||
try:
|
||
from src.core.redis_client import get_redis
|
||
redis = get_redis()
|
||
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))
|
||
|
||
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
# Shadow destructive tool check
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
|
||
def is_destructive_tool(tool_name: str, is_destructive_flag: bool = False) -> bool:
|
||
"""
|
||
判斷 tool call 是否為破壞性操作。
|
||
Shadow mode 下一律攔截。
|
||
|
||
判斷邏輯：
|
||
1. MCP Gateway contract 的 is_destructive=True flag
|
||
2. tool_name 包含破壞性關鍵字（fallback，無 contract 時使用）
|
||
"""
|
||
if is_destructive_flag:
|
||
return True
|
||
tool_lower = tool_name.lower()
|
||
return any(kw in tool_lower for kw in _DESTRUCTIVE_TOOL_KEYWORDS)
|
||
|
||
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
# Run 建立
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
|
||
async def create_run(
|
||
*,
|
||
project_id: str,
|
||
agent_id: str,
|
||
trigger_type: str,
|
||
trigger_ref: str | None = None,
|
||
input_payload: dict[str, Any] | None = None,
|
||
channel_type: str | None = None,
|
||
provider_event_id: str | None = None,
|
||
timeout_seconds: int = 600,
|
||
) -> tuple[UUID, bool]:
|
||
"""
|
||
建立新 run（或返回既有 run，若重複事件）。
|
||
|
||
Returns:
|
||
(run_id, is_duplicate) — is_duplicate=True 表示冪等命中
|
||
|
||
Shadow mode：is_shadow=True，不產生 user response。
|
||
"""
|
||
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
|
||
if input_payload:
|
||
canonical = json.dumps(input_payload, sort_keys=True, separators=(",", ":"))
|
||
input_sha256 = hashlib.sha256(canonical.encode()).hexdigest()
|
||
|
||
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,
|
||
)
|
||
|
||
if channel_type and provider_event_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",
|
||
run_id=str(run_id),
|
||
project_id=project_id,
|
||
agent_id=agent_id,
|
||
is_shadow=_SHADOW_ENABLED,
|
||
trace_id=trace_id,
|
||
work_item_id=correlation.work_item_id,
|
||
trigger_type=trigger_type,
|
||
)
|
||
return selected_run_id, is_duplicate
|
||
|
||
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
# Shadow Execution（Phase 4 主邏輯）
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
|
||
async def shadow_execute(run: AwoooPRunState) -> None:
|
||
"""
|
||
Shadow mode 執行一個 run。
|
||
|
||
Phase 4 行為：
|
||
- 解析 agent contract（get_active()）
|
||
- 執行 tool calls（全部攔截，不實際執行）
|
||
- 記錄 step_journal
|
||
- 完成後 COMPLETED（無 user response）
|
||
|
||
Phase 6+ 才接真實 LLM + channel adapter。
|
||
"""
|
||
run_id = run.run_id
|
||
project_id = run.project_id
|
||
agent_id = run.agent_id
|
||
|
||
logger.info(
|
||
"shadow_execute_start",
|
||
run_id=str(run_id),
|
||
project_id=project_id,
|
||
agent_id=agent_id,
|
||
)
|
||
|
||
try:
|
||
# 解析 agent contract（取得工具清單）
|
||
from src.services.contract_service import get_active_body
|
||
agent_contract = await get_active_body(
|
||
project_id=project_id,
|
||
contract_family="agent",
|
||
contract_id=agent_id,
|
||
)
|
||
|
||
tools = agent_contract.get("tools", []) if agent_contract else []
|
||
|
||
# Shadow step journal：記錄每個工具會被攔截
|
||
step_seq = 0
|
||
async with get_db_context(project_id) as db:
|
||
for tool in tools:
|
||
tool_name = tool.get("tool_name", "unknown")
|
||
blocked = is_destructive_tool(tool_name)
|
||
step = AwoooPRunStepJournal(
|
||
run_id=run_id,
|
||
project_id=project_id,
|
||
step_seq=step_seq,
|
||
tool_name=tool_name,
|
||
mcp_gateway_id=tool.get("mcp_gateway_id"),
|
||
result_status="success" if not blocked else "pending",
|
||
was_blocked=blocked,
|
||
block_reason="shadow_mode_destructive_blocked" if blocked else None,
|
||
)
|
||
db.add(step)
|
||
step_seq += 1
|
||
|
||
# 完成 run（shadow mode：無 user response）
|
||
await transition(
|
||
run_id=run_id,
|
||
project_id=project_id,
|
||
to_state="completed",
|
||
step_count_delta=step_seq,
|
||
)
|
||
|
||
logger.info(
|
||
"shadow_execute_completed",
|
||
run_id=str(run_id),
|
||
steps=step_seq,
|
||
)
|
||
|
||
except Exception as exc:
|
||
logger.exception(
|
||
"shadow_execute_failed",
|
||
run_id=str(run_id),
|
||
error=str(exc),
|
||
)
|
||
await transition(
|
||
run_id=run_id,
|
||
project_id=project_id,
|
||
to_state="failed",
|
||
error_code="E-RUN-001",
|
||
error_detail=str(exc)[:500],
|
||
)
|
||
|
||
|
||
async def get_run_status(run_id: UUID, project_id: str) -> dict[str, Any] | None:
|
||
"""
|
||
查詢單一 run 的 FSM 狀態。回傳 None 表示不存在。
|
||
Router 層透過此 service 函數存取，不直接操作 DB。
|
||
"""
|
||
async with get_db_context(project_id) as db:
|
||
result = await db.execute(
|
||
select(AwoooPRunState).where(
|
||
AwoooPRunState.run_id == run_id,
|
||
AwoooPRunState.project_id == project_id,
|
||
)
|
||
)
|
||
run = result.scalar_one_or_none()
|
||
|
||
if run is None:
|
||
return None
|
||
|
||
correlation = correlation_readback_for_run(
|
||
run.project_id,
|
||
run.run_id,
|
||
run.trace_id,
|
||
)
|
||
return {
|
||
"run_id": str(run.run_id),
|
||
"project_id": run.project_id,
|
||
"agent_id": run.agent_id,
|
||
"state": run.state,
|
||
"is_shadow": run.is_shadow,
|
||
**correlation,
|
||
"attempt_count": run.attempt_count,
|
||
"cost_usd": float(run.cost_usd),
|
||
"step_count": run.step_count,
|
||
"error_code": run.error_code,
|
||
"created_at": run.created_at.isoformat() if run.created_at else None,
|
||
"started_at": run.started_at.isoformat() if run.started_at else None,
|
||
"completed_at": run.completed_at.isoformat() if run.completed_at else None,
|
||
}
|