diff --git a/agent99-register-tasks.ps1 b/agent99-register-tasks.ps1 index de6c6a91b..cc3a93bd5 100644 --- a/agent99-register-tasks.ps1 +++ b/agent99-register-tasks.ps1 @@ -177,6 +177,51 @@ Register-ScheduledTask ` -Description "Agent99 backup freshness, integrity-status, cron-contract, and Telegram alert guard." ` -Force | Out-Null +# Superseded by the canonical all-host performance guard. An older Host188-only +# task sent directly to a product bot and duplicated the controlled Perf lane. +# Quarantine only the one known action fingerprint; an unknown action fails +# closed instead of being disabled blindly. +$legacyHost188TaskName = "Wooo-Agent99-Host188-Performance-Guard" +$legacyHost188ActionSha256 = "a2f98560673ebed5aa1f828b4e577c8a34d1c277b2e498044c3526a4ffe15379" +$legacyHost188Task = Get-ScheduledTask ` + -TaskName $legacyHost188TaskName ` + -TaskPath "\" ` + -ErrorAction SilentlyContinue +if ($legacyHost188Task) { + $legacyHost188Actions = @($legacyHost188Task.Actions) + if ($legacyHost188Actions.Count -ne 1) { + throw "legacy_host188_sender_action_count_mismatch" + } + $sha256 = [Security.Cryptography.SHA256]::Create() + try { + $argumentBytes = [Text.Encoding]::UTF8.GetBytes( + [string]$legacyHost188Actions[0].Arguments + ) + $actualLegacyActionSha256 = (($sha256.ComputeHash($argumentBytes) | + ForEach-Object { $_.ToString("x2") }) -join "") + } finally { + $sha256.Dispose() + } + if ($actualLegacyActionSha256 -ne $legacyHost188ActionSha256) { + throw "legacy_host188_sender_action_identity_mismatch" + } + if ([string]$legacyHost188Task.State -eq "Running") { + throw "legacy_host188_sender_running_refuse_quarantine" + } + if ([bool]$legacyHost188Task.Settings.Enabled) { + Disable-ScheduledTask ` + -TaskName $legacyHost188TaskName ` + -TaskPath "\" | Out-Null + } + $legacyHost188TaskAfter = Get-ScheduledTask ` + -TaskName $legacyHost188TaskName ` + -TaskPath "\" ` + -ErrorAction Stop + if ([bool]$legacyHost188TaskAfter.Settings.Enabled) { + throw "legacy_host188_sender_quarantine_post_verifier_failed" + } +} + $taskNames = @( "Wooo-Agent99-Startup-Recovery", "Wooo-Agent99-Heartbeat", diff --git a/apps/api/src/core/config.py b/apps/api/src/core/config.py index 086b33b75..2a9795f48 100644 --- a/apps/api/src/core/config.py +++ b/apps/api/src/core/config.py @@ -699,6 +699,19 @@ class Settings(BaseSettings): ) GEMINI_API_KEY: str = Field(default="", description="Google Gemini API key") CLAUDE_API_KEY: str = Field(default="", description="Anthropic Claude API key") + ENABLE_PAID_PROVIDER_CANARY_WORKER: bool = Field( + default=False, + description=( + "Let the singleton signal worker execute an already-approved, " + "single-use AIA-SRE-013 paid-provider canary run." + ), + ) + PAID_PROVIDER_CANARY_WORKER_INTERVAL_SECONDS: int = Field( + default=2, + ge=1, + le=30, + description="Polling interval for the bounded paid-provider canary worker.", + ) LOCAL_CODE_REVIEW_ALLOW_GEMINI_FALLBACK: bool = Field( default=False, description=( diff --git a/apps/api/src/jobs/paid_provider_canary_worker.py b/apps/api/src/jobs/paid_provider_canary_worker.py new file mode 100644 index 000000000..8d8ea5adb --- /dev/null +++ b/apps/api/src/jobs/paid_provider_canary_worker.py @@ -0,0 +1,157 @@ +"""Run one approved paid-provider canary from the singleton signal worker. + +The Gate 5 API only creates and approves the durable run. This worker is the +bounded executor handoff: it claims the exact ``running`` AIA-SRE-013 row and +invokes the existing single-use validation contract with the same run, +authorization, trace, and work-item identity. It never receives API keys as +arguments and never persists prompts or responses. +""" + +from __future__ import annotations + +import asyncio +from collections.abc import Awaitable, Callable +from datetime import UTC, datetime +from typing import Any + +import structlog +from sqlalchemy import select + +from src.db.awooop_models import AwoooPRunState +from src.db.base import get_db_context +from src.services.paid_provider_canary_authorization import ( + PAID_CANARY_AGENT_ID, + PAID_CANARY_PROJECT_ID, + PAID_CANARY_TRIGGER_REF, +) + +logger = structlog.get_logger(__name__) + +_WORKER_OPERATOR_ID = "awoooi-paid-provider-canary-worker" +_DEFAULT_INTERVAL_SECONDS = 2.0 + + +async def _load_next_approved_run_id() -> str | None: + now = datetime.now(UTC).replace(tzinfo=None) + async with get_db_context(PAID_CANARY_PROJECT_ID) as db: + result = await db.execute( + select(AwoooPRunState.run_id) + .where( + AwoooPRunState.project_id == PAID_CANARY_PROJECT_ID, + AwoooPRunState.agent_id == PAID_CANARY_AGENT_ID, + AwoooPRunState.trigger_ref == PAID_CANARY_TRIGGER_REF, + AwoooPRunState.state == "running", + AwoooPRunState.is_shadow.is_(False), + AwoooPRunState.timeout_at.is_not(None), + AwoooPRunState.timeout_at > now, + ) + .order_by(AwoooPRunState.created_at.asc()) + .limit(1) + ) + run_id = result.scalar_one_or_none() + return str(run_id) if run_id is not None else None + + +async def _execute_canary(**kwargs: Any) -> dict[str, Any]: + from src.services.paid_provider_canary_validation import ( + run_paid_provider_canary_validation, + ) + + return await run_paid_provider_canary_validation(**kwargs) + + +async def run_paid_provider_canary_worker_once( + *, + run_loader: Callable[[], Awaitable[str | None]] | None = None, + executor: Callable[..., Awaitable[dict[str, Any]]] | None = None, +) -> dict[str, Any]: + """Execute at most one exact approved run and return a non-secret receipt.""" + + selected_run_loader = run_loader or _load_next_approved_run_id + selected_executor = executor or _execute_canary + run_id = await selected_run_loader() + if not run_id: + return { + "schema_version": "paid_provider_canary_worker_receipt_v1", + "status": "idle_no_approved_run", + "execution_attempted": False, + "raw_prompt_persisted": False, + "raw_response_persisted": False, + "secret_value_read_or_returned": False, + } + + try: + receipt = await selected_executor( + run_ref=run_id, + authorization_ref=run_id, + operator_id=_WORKER_OPERATOR_ID, + ) + except asyncio.CancelledError: + raise + except BaseException as exc: + logger.error( + "paid_provider_canary_worker_execution_failed", + run_id=run_id, + error_code=f"executor_{type(exc).__name__}"[:120], + ) + return { + "schema_version": "paid_provider_canary_worker_receipt_v1", + "status": "blocked_with_safe_next_action", + "run_id": run_id, + "execution_attempted": True, + "error_code": f"executor_{type(exc).__name__}"[:120], + "raw_prompt_persisted": False, + "raw_response_persisted": False, + "secret_value_read_or_returned": False, + } + + verifier = receipt.get("independent_verifier") or {} + return { + "schema_version": "paid_provider_canary_worker_receipt_v1", + "status": ( + "terminal_verified" + if verifier.get("passed") is True + else "terminal_failed_verifier" + ), + "run_id": run_id, + "execution_attempted": True, + "same_run_authorization": receipt.get("run_id") == run_id, + "independent_verifier_passed": verifier.get("passed") is True, + "raw_prompt_persisted": False, + "raw_response_persisted": False, + "secret_value_read_or_returned": False, + } + + +async def run_paid_provider_canary_worker_loop( + *, interval_seconds: float = _DEFAULT_INTERVAL_SECONDS, +) -> None: + """Poll for an approved run; single-use claims prevent duplicate calls.""" + + delay = max(1.0, min(30.0, float(interval_seconds))) + logger.info( + "paid_provider_canary_worker_started", + interval_seconds=delay, + project_id=PAID_CANARY_PROJECT_ID, + ) + while True: + try: + receipt = await run_paid_provider_canary_worker_once() + if receipt["execution_attempted"] is True: + logger.info( + "paid_provider_canary_worker_cycle_terminal", + run_id=receipt.get("run_id"), + status=receipt.get("status"), + independent_verifier_passed=receipt.get( + "independent_verifier_passed" + ), + ) + except asyncio.CancelledError: + raise + except BaseException as exc: + logger.error( + "paid_provider_canary_worker_cycle_failed", + error_code=f"cycle_{type(exc).__name__}"[:120], + ) + await asyncio.sleep(delay) + diff --git a/apps/api/src/models/ai.py b/apps/api/src/models/ai.py index 37e867dfa..47301f8c7 100644 --- a/apps/api/src/models/ai.py +++ b/apps/api/src/models/ai.py @@ -38,6 +38,7 @@ class AIRiskLevel(str, Enum): """AI 風險評估等級""" LOW = "low" MEDIUM = "medium" + HIGH = "high" CRITICAL = "critical" diff --git a/apps/api/src/services/ai_control.py b/apps/api/src/services/ai_control.py index b3b3bdad6..7344df4f3 100644 --- a/apps/api/src/services/ai_control.py +++ b/apps/api/src/services/ai_control.py @@ -22,6 +22,7 @@ Usage: """ import hashlib +import json import re from typing import Any @@ -585,21 +586,144 @@ async def get_status_summary() -> str: primary = await get_primary_provider() primary_str = f"🎯 {primary}" if primary else "🤖 智慧路由 (AIRouter 決定)" - # Provider 狀態 + # Provider 狀態：只讀既有 health/control receipts，不呼叫模型或端點。 + ollama_specs = ( + ("ollama_gcp_a", "GCP-A", "OLLAMA_URL"), + ("ollama_gcp_b", "GCP-B", "OLLAMA_SECONDARY_URL"), + ("ollama_local", "111", "OLLAMA_FALLBACK_URL"), + ) + ollama_health = {provider: "not_observed" for provider, _, _ in ollama_specs} + try: + from src.services.ollama_health_monitor import make_cache_key + + health_redis = await _get_redis() + for provider, _, setting_name in ollama_specs: + endpoint = str(getattr(settings, setting_name, "") or "").strip() + if not endpoint: + ollama_health[provider] = "not_configured" + continue + try: + raw = await health_redis.get(make_cache_key(endpoint)) + if raw: + if isinstance(raw, bytes): + raw = raw.decode() + receipt = json.loads(raw) + ollama_health[provider] = str( + receipt.get("status") or "invalid_receipt" + ) + except Exception as exc: + ollama_health[provider] = "readback_unavailable" + logger.warning( + "ai_control_ollama_health_receipt_read_failed", + provider=provider, + error_type=type(exc).__name__, + ) + except Exception as exc: + logger.warning( + "ai_control_ollama_health_readback_unavailable", + error_type=type(exc).__name__, + ) + provider_lines = [] - for p in [ - "ollama_gcp_a", - "ollama_gcp_b", - "ollama_local", - "claude", - "gemini", - ]: + for provider, label, _ in ollama_specs: try: - disabled = await is_provider_disabled(p) + control = "disabled" if await is_provider_disabled(provider) else "enabled" except Exception: - disabled = True - icon = "❌" if disabled else "✅" - provider_lines.append(f" {icon} {p}") + control = "readback_unavailable" + health = ollama_health[provider] + if control == "disabled": + icon = "⏸️" + elif health == "healthy": + icon = "🟢" + elif health in {"slow", "degraded"}: + icon = "🟠" + elif health == "offline": + icon = "🔴" + else: + icon = "⚪" + provider_lines.append( + f" {icon} {label} | health={health} | control={control}" + ) + + paid_control: dict[str, Any] = {} + route_lease = "readback_unavailable" + try: + paid_control = await get_paid_provider_canary_control_readback() + if paid_control.get("legacy_persistent_state_detected") is True: + route_lease = "legacy_persistent_state_detected" + elif ( + paid_control.get("lock_present") is True + and paid_control.get("claude_disabled") is False + and paid_control.get("gemini_disabled") is False + and int(paid_control.get("lock_ttl_seconds") or 0) > 0 + ): + route_lease = "active_run_owned_lease" + elif paid_control.get("disabled_verified") is True: + route_lease = "待單次 run-owned lease" + else: + route_lease = "inconsistent_fail_closed" + except Exception as exc: + logger.warning( + "ai_control_paid_control_status_readback_unavailable", + error_type=type(exc).__name__, + ) + + authentication: dict[str, dict[str, Any]] = {} + try: + from src.services.ai_rate_limiter import get_ai_rate_limiter + + limiter = get_ai_rate_limiter() + for provider in ("claude", "gemini"): + try: + authentication[provider] = ( + await limiter.get_provider_authentication_status(provider) + ) + except Exception: + authentication[provider] = { + "configured": None, + "authenticated": None, + "authentication_verified": False, + "verification_status": "authentication_readback_unavailable", + } + except Exception as exc: + logger.warning( + "ai_control_paid_authentication_status_readback_unavailable", + error_type=type(exc).__name__, + ) + + def _truth_text(value: Any) -> str: + if value is True: + return "true" + if value is False: + return "false" + return "unknown" + + for provider, label in (("claude", "Claude"), ("gemini", "Gemini")): + auth = authentication.get(provider) or {} + configured = auth.get("configured") + mode = str( + getattr(settings, f"{provider.upper()}_EXECUTION_MODE", "shadow") + ).lower() + percent = int( + getattr(settings, f"{provider.upper()}_CANARY_PERCENT", 0) or 0 + ) + if configured is True and auth.get("authenticated") is False: + icon = "⚠️" + elif configured is True and route_lease == "待單次 run-owned lease": + icon = "🧪" + elif configured is True and route_lease == "active_run_owned_lease": + icon = "🟢" + elif configured is True: + icon = "⚠️" + else: + icon = "⚪" + provider_lines.append( + f" {icon} {label} | configured={_truth_text(configured)} | " + f"authenticated={_truth_text(auth.get('authenticated'))} | " + "verification=" + f"{auth.get('verification_status', 'authentication_readback_unavailable')} | " + f"canary={mode} {percent}% | route_lease={route_lease}" + ) provider_lines.extend( f" ⚪ {provider} (shadow metadata only)" for provider in sorted(SHADOW_METADATA_ONLY_PROVIDERS) @@ -608,15 +732,31 @@ async def get_status_summary() -> str: # 費用統計 cost_lines = [] try: - r = await _get_redis() - for p in ["claude", "gemini"]: - val = await r.get(f"ai_rate:total_cost:{p}") - if val: - cost = float(val) - if cost > 0: - cost_lines.append(f" 💰 {p}: ${cost:.4f}") - except Exception: - cost_lines.append(" (費用統計不可用)") + from src.services.ai_rate_limiter import get_ai_rate_limiter + + limiter = get_ai_rate_limiter() + for provider in ("claude", "gemini"): + usage = await limiter.get_usage_stats(provider) + cost = float( + (usage.get("total_cost_usd") or {}).get("current") or 0.0 + ) + accounting = usage.get("accounting") or {} + receipt_backed = bool( + usage.get("readback_status") == "verified" + and accounting.get("complete") is True + and accounting.get("status") == "receipt_backed" + and accounting.get("degraded") is not True + ) + cost_lines.append( + f" 💰 {provider}: ${cost:.6f} | " + f"receipt={'verified' if receipt_backed else 'degraded'}" + ) + except Exception as exc: + logger.warning( + "ai_control_status_cost_readback_unavailable", + error_type=type(exc).__name__, + ) + cost_lines.append(" (accounting receipt readback unavailable)") providers_str = "\n".join(provider_lines) costs_str = "\n".join(cost_lines) if cost_lines else " (尚無費用記錄)" @@ -629,7 +769,8 @@ async def get_status_summary() -> str: f"費用統計:\n{costs_str}\n\n" f"/ai primary <provider> 切換\n" f"/ai router on/off 開關 AIRouter\n" - f"/ai enable/disable <provider> 控制 Provider" + f"/ai enable/disable <ollama provider> 控制 Ollama\n" + f"Claude/Gemini 僅由單次 run-owned canary lease 啟用" ) @@ -709,23 +850,49 @@ async def handle_ai_command(text: str) -> str: return f"⚠️ {provider} 已停用" if ok else "❌ 操作失敗" elif sub == "cost": - lines = ["💰 AI 費用統計\n"] + lines = ["💰 AI 費用 receipt\n"] try: - r = await _get_redis() + from src.services.ai_rate_limiter import get_ai_rate_limiter + + limiter = get_ai_rate_limiter() total = 0.0 - for p in ["claude", "gemini"]: - val = await r.get(f"ai_rate:total_cost:{p}") - if val: - cost = float(val) - total += cost - if cost > 0: - lines.append(f" {p}: ${cost:.4f}") - if total > 0: - lines.append(f"\n 總計: ${total:.4f}") - else: - lines.append(" (尚無費用記錄)") - except Exception as e: - lines.append(f" ⚠️ 費用查詢失敗: {e}") + all_receipt_backed = True + for provider in ("claude", "gemini"): + usage = await limiter.get_usage_stats(provider) + cost = float( + (usage.get("total_cost_usd") or {}).get("current") or 0.0 + ) + accounting = usage.get("accounting") or {} + receipt_backed = bool( + usage.get("readback_status") == "verified" + and accounting.get("complete") is True + and accounting.get("status") == "receipt_backed" + and accounting.get("degraded") is not True + ) + total += cost + all_receipt_backed = all_receipt_backed and receipt_backed + lines.append( + f" {provider}: ${cost:.6f} | " + f"receipt={'verified' if receipt_backed else 'degraded'}" + ) + lines.extend( + [ + f"\n 總計: ${total:.6f}", + ( + " ✅ receipt-backed accounting" + if all_receipt_backed + else " ⚠️ accounting receipt degraded；禁止假性估算" + ), + ] + ) + except Exception as exc: + logger.warning( + "ai_control_cost_readback_unavailable", + error_type=type(exc).__name__, + ) + lines.append( + " ⚠️ accounting readback unavailable；未以舊 Redis key 或估算值替代。" + ) return "\n".join(lines) elif sub == "help": diff --git a/apps/api/src/services/ai_providers/claude.py b/apps/api/src/services/ai_providers/claude.py index 93277c733..65181a8af 100644 --- a/apps/api/src/services/ai_providers/claude.py +++ b/apps/api/src/services/ai_providers/claude.py @@ -136,8 +136,10 @@ def _analysis_tool() -> dict[str, Any]: return { "name": "submit_analysis", "description": "Submit the RCA analysis result in structured format", + "strict": True, "input_schema": { "type": "object", + "additionalProperties": False, "properties": { "action_title": {"type": "string"}, "description": {"type": "string"}, @@ -159,6 +161,7 @@ def _analysis_tool() -> dict[str, Any]: }, "blast_radius": { "type": "object", + "additionalProperties": False, "properties": { "affected_pods": {"type": "integer"}, "estimated_downtime": {"type": "string"}, @@ -180,7 +183,11 @@ def _analysis_tool() -> dict[str, Any]: }, "reasoning": {"type": "string"}, "deviation_analysis": {"type": "string"}, - "confidence": {"type": "number"}, + "confidence": { + "type": "number", + "minimum": 0.0, + "maximum": 1.0, + }, "affected_services": { "type": "array", "items": {"type": "string"}, diff --git a/apps/api/src/services/ai_providers/ollama.py b/apps/api/src/services/ai_providers/ollama.py index 03c3b3603..17a4168b8 100644 --- a/apps/api/src/services/ai_providers/ollama.py +++ b/apps/api/src/services/ai_providers/ollama.py @@ -13,6 +13,7 @@ from __future__ import annotations import hashlib import json +import re import time from collections.abc import Awaitable, Callable @@ -54,6 +55,24 @@ def _is_gcp_alert_lane(endpoint_url: str) -> bool: CloudTransportReceiptVerifier = Callable[..., Awaitable[str | None]] +def _safe_ollama_error_code(error: BaseException) -> str: + """Classify provider failures without returning endpoint or body text.""" + + if isinstance(error, httpx.TimeoutException): + return "provider_timeout" + if isinstance(error, httpx.HTTPStatusError): + status_code = int(error.response.status_code) + if 100 <= status_code <= 599: + return f"http_{status_code}" + return "provider_refusal" + if isinstance(error, httpx.TransportError | ConnectionError): + return "provider_transport_error" + error_type = type(error).__name__ + if re.fullmatch(r"[A-Za-z][A-Za-z0-9_]{0,79}", error_type): + return f"unhandled_{error_type}" + return "provider_error_redacted" + + def _gcp_provider_identity(endpoint_url: str) -> str | None: endpoint = _normalized_url(endpoint_url) if endpoint == _normalized_url(getattr(settings, "OLLAMA_URL", "")): @@ -63,6 +82,51 @@ def _gcp_provider_identity(endpoint_url: str) -> str | None: return None +def _sre_chat_sanitization_receipt_block_reason( + *, + prompt: str, + context: dict, +) -> str | None: + """Mirror the router's formal real-chat receipt at the provider boundary.""" + + receipt = context.get("sre_chat_sanitization_receipt") + if not isinstance(receipt, dict): + return "sre_chat_sanitization_receipt_invalid" + + prompt_sha256 = hashlib.sha256(prompt.encode("utf-8")).hexdigest() + if receipt.get("prompt_sha256") != prompt_sha256: + return "sre_chat_sanitized_prompt_mismatch" + + correlation_fields = ("trace_id", "run_id", "work_item_id") + if ( + receipt.get("schema_version") != "sre_chat_sanitization_receipt_v1" + or receipt.get("status") != "verified" + or receipt.get("sanitizer") + != "src.services.sanitization_service.sanitize" + or receipt.get("source_label") != "awoooi_sre_group_chat" + or receipt.get("data_classification") != "sanitized" + or receipt.get("raw_input_forwarded") is not False + or receipt.get("raw_log_payload_forwarded") is not False + or receipt.get("secret_value_exposed") is not False + or receipt.get("private_network_value_exposed") is not False + or receipt.get("infrastructure_remediation_write_allowed") is not False + or receipt.get("provider_timeout_seconds") != 35 + or context.get("data_classification") != "sanitized" + or context.get("infrastructure_remediation_write_allowed") is not False + or context.get("provider_timeout_seconds") != 35 + or context.get("executor_invocation_allowed") is not False + or context.get("agent99_dispatch_allowed") is not False + or context.get("cross_domain_execution_allowed") is not False + or any( + not str(context.get(field) or "") + or receipt.get(field) != context.get(field) + for field in correlation_fields + ) + ): + return "sre_chat_sanitization_receipt_invalid" + return None + + async def _gcp_cloud_input_block_reason( *, prompt: str, @@ -96,6 +160,12 @@ async def _gcp_cloud_input_block_reason( ): return None return "gcp_synthetic_prompt_receipt_invalid" + if "sre_chat_sanitization_receipt" in cloud_context: + chat_block = _sre_chat_sanitization_receipt_block_reason( + prompt=prompt, + context=cloud_context, + ) + return f"gcp_{chat_block}" if chat_block else None receipt = cloud_context.get("cloud_sanitization_receipt") if not isinstance(receipt, dict): return "gcp_cloud_sanitization_receipt_missing" @@ -278,13 +348,37 @@ class OllamaProvider: except httpx.TimeoutException as e: latency = (time.perf_counter() - start) * 1000 - logger.warning("ollama_provider_timeout", error=str(e), latency_ms=round(latency, 1)) - return AIResult(raw_response="", success=False, provider=self.name, latency_ms=latency, error=f"Timeout: {e}") + error_code = _safe_ollama_error_code(e) + logger.warning( + "ollama_provider_timeout", + error_code=error_code, + error_type=type(e).__name__, + latency_ms=round(latency, 1), + ) + return AIResult( + raw_response="", + success=False, + provider=self.name, + latency_ms=latency, + error=error_code, + ) except Exception as e: latency = (time.perf_counter() - start) * 1000 - logger.warning("ollama_provider_failed", error=str(e), latency_ms=round(latency, 1)) - return AIResult(raw_response="", success=False, provider=self.name, latency_ms=latency, error=str(e)) + error_code = _safe_ollama_error_code(e) + logger.warning( + "ollama_provider_failed", + error_code=error_code, + error_type=type(e).__name__, + latency_ms=round(latency, 1), + ) + return AIResult( + raw_response="", + success=False, + provider=self.name, + latency_ms=latency, + error=error_code, + ) async def analyze_with_tools( self, @@ -429,11 +523,13 @@ class OllamaProvider: except Exception as e: latency = (time.perf_counter() - start) * 1000 + error_code = _safe_ollama_error_code(e) logger.warning( "ollama_agent_loop_failed", provider=self.name, agent_role=agent_role, - error=str(e), + error_code=error_code, + error_type=type(e).__name__, latency_ms=round(latency, 1), ) return AIResult( @@ -442,7 +538,7 @@ class OllamaProvider: provider=f"{self.name}_agent_loop", tokens=total_tokens, latency_ms=latency, - error=str(e), + error=error_code, ) async def health_check(self) -> bool: @@ -566,13 +662,37 @@ class OllamaLocalProvider(OllamaProvider): except httpx.TimeoutException as e: latency = (time.perf_counter() - start) * 1000 - logger.warning("ollama_local_provider_timeout", error=str(e), latency_ms=round(latency, 1)) - return AIResult(raw_response="", success=False, provider=self.name, latency_ms=latency, error=f"Timeout: {e}") + error_code = _safe_ollama_error_code(e) + logger.warning( + "ollama_local_provider_timeout", + error_code=error_code, + error_type=type(e).__name__, + latency_ms=round(latency, 1), + ) + return AIResult( + raw_response="", + success=False, + provider=self.name, + latency_ms=latency, + error=error_code, + ) except Exception as e: latency = (time.perf_counter() - start) * 1000 - logger.warning("ollama_local_provider_failed", error=str(e), latency_ms=round(latency, 1)) - return AIResult(raw_response="", success=False, provider=self.name, latency_ms=latency, error=str(e)) + error_code = _safe_ollama_error_code(e) + logger.warning( + "ollama_local_provider_failed", + error_code=error_code, + error_type=type(e).__name__, + latency_ms=round(latency, 1), + ) + return AIResult( + raw_response="", + success=False, + provider=self.name, + latency_ms=latency, + error=error_code, + ) async def health_check(self) -> bool: fallback_url = getattr(settings, "OLLAMA_FALLBACK_URL", "") diff --git a/apps/api/src/services/ai_router.py b/apps/api/src/services/ai_router.py index e838ebbea..084fa6609 100644 --- a/apps/api/src/services/ai_router.py +++ b/apps/api/src/services/ai_router.py @@ -1190,6 +1190,54 @@ class AIRouterExecutor: content = f"{prompt}{ctx_hash}" return f"llm_cache:{hashlib.sha256(content.encode()).hexdigest()[:16]}" + @staticmethod + def _sre_chat_sanitization_receipt_block_reason( + prompt: str, + context: dict | None, + ) -> str | None: + """Verify the formal prompt-bound receipt for real SRE conversation.""" + + root = context or {} + receipt = root.get("sre_chat_sanitization_receipt") + if not isinstance(receipt, dict): + # Preserve the established generic cloud boundary code for callers + # that provide neither an alert nor a real-chat receipt. + return "cloud_sanitization_receipt_missing" + + prompt_sha256 = hashlib.sha256(prompt.encode("utf-8")).hexdigest() + if receipt.get("prompt_sha256") != prompt_sha256: + return "sre_chat_sanitized_prompt_mismatch" + + correlation_fields = ("trace_id", "run_id", "work_item_id") + if ( + receipt.get("schema_version") + != "sre_chat_sanitization_receipt_v1" + or receipt.get("status") != "verified" + or receipt.get("sanitizer") + != "src.services.sanitization_service.sanitize" + or receipt.get("source_label") != "awoooi_sre_group_chat" + or receipt.get("data_classification") != "sanitized" + or receipt.get("raw_input_forwarded") is not False + or receipt.get("raw_log_payload_forwarded") is not False + or receipt.get("secret_value_exposed") is not False + or receipt.get("private_network_value_exposed") is not False + or receipt.get("infrastructure_remediation_write_allowed") is not False + or receipt.get("provider_timeout_seconds") != 35 + or root.get("data_classification") != "sanitized" + or root.get("infrastructure_remediation_write_allowed") is not False + or root.get("provider_timeout_seconds") != 35 + or root.get("executor_invocation_allowed") is not False + or root.get("agent99_dispatch_allowed") is not False + or root.get("cross_domain_execution_allowed") is not False + or any( + not str(root.get(field) or "") + or receipt.get(field) != root.get(field) + for field in correlation_fields + ) + ): + return "sre_chat_sanitization_receipt_invalid" + return None + @staticmethod def _paid_alert_execution_inputs( prompt: str, @@ -1202,16 +1250,24 @@ class AIRouterExecutor: for key in ("alert_type", "alertname", "alert_name", "incident_id", "signals") ) if not is_alert: - if ( - (context or {}).get("synthetic_validation") is True - and (context or {}).get("data_classification") == "sanitized" - and (context or {}).get("infrastructure_remediation_write_allowed") - is False - and (context or {}).get("synthetic_prompt_sha256") - == hashlib.sha256(prompt.encode("utf-8")).hexdigest() - ): + root = context or {} + if root.get("synthetic_validation") is True: + if ( + root.get("data_classification") == "sanitized" + and root.get("infrastructure_remediation_write_allowed") is False + and root.get("synthetic_prompt_sha256") + == hashlib.sha256(prompt.encode("utf-8")).hexdigest() + ): + return prompt, context, None + return None, None, "cloud_synthetic_prompt_receipt_invalid" + + chat_block = AIRouterExecutor._sre_chat_sanitization_receipt_block_reason( + prompt, + context, + ) + if chat_block is None: return prompt, context, None - return None, None, "cloud_sanitization_receipt_missing" + return None, None, chat_block cloud_context = (context or {}).get("paid_cloud_context") paid_prompt = (context or {}).get("paid_cloud_prompt") if not isinstance(cloud_context, dict) or not isinstance(paid_prompt, str): diff --git a/apps/api/src/services/awooop_ansible_audit_service.py b/apps/api/src/services/awooop_ansible_audit_service.py index 211f5a136..6f4efcd46 100644 --- a/apps/api/src/services/awooop_ansible_audit_service.py +++ b/apps/api/src/services/awooop_ansible_audit_service.py @@ -623,23 +623,22 @@ _CATALOG: tuple[dict[str, Any], ...] = ( "host_188", ], "keywords": [ - "openclawdown", - "clawbotdown", + "telegramcallbackingressunverified", + "telegramcallbackforwarderdrift", "openclaw callback forwarder", "telegram callback forwarder", "external callback forward unverified", ], "activation_keywords": [ - "openclawdown", - "clawbotdown", - "telegram callback forwarder", + "telegramcallbackingressunverified", + "telegramcallbackforwarderdrift", ], "supports_check_mode": True, "auto_apply_enabled": True, "approval_required": False, "risk_level": "medium", "canonical_asset_id": "service:openclaw:host188", - "catalog_revision": "2026-07-17-openclaw-callback-forwarder-v1", + "catalog_revision": "2026-07-17-openclaw-callback-overlay-v2", "cross_domain_fallback_allowed": False, }, { diff --git a/apps/api/src/services/awooop_ansible_post_verifier.py b/apps/api/src/services/awooop_ansible_post_verifier.py index 6408f6fac..578d94e3b 100644 --- a/apps/api/src/services/awooop_ansible_post_verifier.py +++ b/apps/api/src/services/awooop_ansible_post_verifier.py @@ -24,6 +24,11 @@ _SAFE_REMOTE_ID_RE = re.compile(r"^[A-Za-z0-9_.-]+$") _FIXED_PROXY_JUMP_BY_HOST = { "host_111": "wooo@192.168.0.110", } +_RUN_SCOPED_CLOSURE_BLOCKER_BY_CATALOG = { + "ansible:188-openclaw-callback-forwarder": ( + "fresh_callback_receipt_run_scope_unavailable_in_static_postcondition_registry" + ), +} @dataclass(frozen=True) @@ -240,41 +245,136 @@ _POSTCONDITION_REGISTRY: dict[str, tuple[AssetPostcondition, ...]] = { ), "ansible:188-openclaw-callback-forwarder": ( AssetPostcondition( - "host_188_openclaw_immutable_source_identity", + "host_188_openclaw_overlay_manifest_identity", "configuration", "host_188", - "wd=$(systemctl show clawbot.service --property=WorkingDirectory " - "--value); case \"$wd\" in /opt/openclaw|/home/ollama/clawbot-v5) ;; " - "*) exit 1 ;; esac; " - "test \"$(git -C \"$wd\" symbolic-ref --short HEAD)\" = main && " - "test \"$(git -C \"$wd\" rev-parse --verify HEAD)\" = " - "ffc45311b6403b30cb6ae6e1aa83dae1a047e9c9 && " - "test \"$(git -C \"$wd\" remote get-url origin)\" = " - "ssh://git@192.168.0.110:2222/wooo/clawbot-v5.git && " - "git -C \"$wd\" diff --quiet && " - "git -C \"$wd\" diff --cached --quiet && " - "test -z \"$(git -C \"$wd\" ls-files --others --exclude-standard)\"", + "set -euo pipefail; " + "root=/var/lib/awoooi-controlled-apply/openclaw-callback-forwarder/" + "payloads/ffc45311b6403b30cb6ae6e1aa83dae1a047e9c9 && " + "manifest=\"$root/manifest.json\" && " + "test -d \"$root\" && test ! -L \"$root\" && " + "test \"$(stat -c '%u:%g:%a' \"$root\")\" = '0:0:755' && " + "test -f \"$manifest\" && test ! -L \"$manifest\" && " + "test \"$(stat -c '%u:%g:%a' \"$manifest\")\" = '0:0:444' && " + "test \"$(sha256sum \"$manifest\" | cut -d' ' -f1)\" = " + "f06df6f3163e5ae7bf999624b011a3ae0bff096c9d0c8cb4c43f8b40be43ae2b", ), AssetPostcondition( - "host_188_openclaw_callback_dropin_enabled", + "host_188_openclaw_callback_overlay_enabled", "configuration", "host_188", - "test -r /etc/systemd/system/clawbot.service.d/" + "set -euo pipefail; " + "drop=/etc/systemd/system/clawbot.service.d/" "20-awoooi-callback-forwarder.conf && " - "grep -Fqx 'Environment=TELEGRAM_CALLBACK_FORWARD_ENABLED=true' " - "/etc/systemd/system/clawbot.service.d/" - "20-awoooi-callback-forwarder.conf", + "override=/etc/awoooi/openclaw-callback-forwarder/" + "docker-compose.callback.yml && " + "COMPOSE_FILE_key=COMPOSE_FILE && " + "test \"$COMPOSE_FILE_key\" = 'COMPOSE_FILE' && " + "wd=$(systemctl show clawbot.service --property=WorkingDirectory " + "--value) && " + "test -f \"$drop\" && test ! -L \"$drop\" && " + "test \"$(stat -c '%u:%g:%a' \"$drop\")\" = '0:0:644' && " + "drop_sha=$(sha256sum \"$drop\" | cut -d' ' -f1) && " + "case \"$wd:$drop_sha\" in " + "'/home/ollama/clawbot-v5:" + "419fb4f249f7e1370fbca87086fd725e57b1f5cf54da2f84caf9a21fea7111e3'" + "|'/opt/openclaw:" + "4a78d414e4b591f54232fab04529861a6042718a7627c4e17ccd5cc0cbd485f9') " + "true ;; *) false ;; esac && " + "test -f \"$override\" && test ! -L \"$override\" && " + "test \"$(stat -c '%u:%g:%a' \"$override\")\" = '0:0:444' && " + "test \"$(sha256sum \"$override\" | cut -d' ' -f1)\" = " + "f6fa7f6724c35f86f1c8dafac07e2680231dfa29a5d45534ed0515c58ac2de77", + ), + AssetPostcondition( + "host_188_openclaw_callback_payload_runtime", + "configuration", + "host_188", + "set -euo pipefail; " + "root=/var/lib/awoooi-controlled-apply/openclaw-callback-forwarder/" + "payloads/ffc45311b6403b30cb6ae6e1aa83dae1a047e9c9 && " + "telegram=\"$root/app/bot/telegram.py\" && " + "config=\"$root/app/core/config.py\" && " + "forwarder=\"$root/app/services/telegram_callback_forwarder.py\" && " + "test -f \"$telegram\" && test ! -L \"$telegram\" && " + "test \"$(stat -c '%u:%g:%a' \"$telegram\")\" = '0:0:444' && " + "test \"$(sha256sum \"$telegram\" | cut -d' ' -f1)\" = " + "af9600f32305ae73184651a8d3437063d7c8d35e3c3eb0910a2e27ac874c9302 && " + "test -f \"$config\" && test ! -L \"$config\" && " + "test \"$(stat -c '%u:%g:%a' \"$config\")\" = '0:0:444' && " + "test \"$(sha256sum \"$config\" | cut -d' ' -f1)\" = " + "e1b8b633f7b4780d3078d6c42a55e49d200255d3597a34e0cd2fd649b84eaea1 && " + "test -f \"$forwarder\" && test ! -L \"$forwarder\" && " + "test \"$(stat -c '%u:%g:%a' \"$forwarder\")\" = '0:0:444' && " + "test \"$(sha256sum \"$forwarder\" | cut -d' ' -f1)\" = " + "0d7693e34f63db4491d974f9f76a3a1c793d32cda193cb55ed3d8c86297b4fe3 && " + "ids=$(docker ps -q " + "--filter label=com.docker.compose.project=clawbot " + "--filter label=com.docker.compose.service=clawbot) && " + "test \"$(printf '%s\\n' \"$ids\" | sed '/^$/d' | wc -l | " + "tr -d ' ')\" = 1 && " + "id=\"$ids\" && " + "test \"$(docker inspect --format='{{.Config.WorkingDir}}' \"$id\")\" " + "= '/app' && " + "test \"$(docker exec \"$ids\" sha256sum " + "/app/app/bot/telegram.py | cut -d' ' -f1)\" = " + "af9600f32305ae73184651a8d3437063d7c8d35e3c3eb0910a2e27ac874c9302 && " + "test \"$(docker exec \"$ids\" sha256sum " + "/app/app/core/config.py | cut -d' ' -f1)\" = " + "e1b8b633f7b4780d3078d6c42a55e49d200255d3597a34e0cd2fd649b84eaea1 && " + "test \"$(docker exec \"$ids\" sha256sum " + "/app/app/services/telegram_callback_forwarder.py | cut -d' ' -f1)\" = " + "0d7693e34f63db4491d974f9f76a3a1c793d32cda193cb55ed3d8c86297b4fe3 && " + "test \"$(docker inspect --format='{{range .Mounts}}{{if eq " + ".Destination \"/app/app/bot/telegram.py\"}}{{.Source}}|" + "{{.Destination}}|{{.Type}}|{{.RW}}{{end}}{{end}}' \"$id\")\" = " + "\"$telegram|/app/app/bot/telegram.py|bind|false\" && " + "test \"$(docker inspect --format='{{range .Mounts}}{{if eq " + ".Destination \"/app/app/core/config.py\"}}{{.Source}}|" + "{{.Destination}}|{{.Type}}|{{.RW}}{{end}}{{end}}' \"$id\")\" = " + "\"$config|/app/app/core/config.py|bind|false\" && " + "test \"$(docker inspect --format='{{range .Mounts}}{{if eq " + ".Destination \"/app/app/services/telegram_callback_forwarder.py\"}}" + "{{.Source}}|{{.Destination}}|{{.Type}}|{{.RW}}{{end}}{{end}}' " + "\"$id\")\" = \"$forwarder|/app/app/services/" + "telegram_callback_forwarder.py|bind|false\"", + ), + AssetPostcondition( + "host_188_openclaw_callback_runtime_setting", + "configuration", + "host_188", + "set -euo pipefail; " + "ids=$(docker ps -q " + "--filter label=com.docker.compose.project=clawbot " + "--filter label=com.docker.compose.service=clawbot) && " + "test \"$(printf '%s\\n' \"$ids\" | sed '/^$/d' | wc -l | " + "tr -d ' ')\" = 1 && " + "docker exec \"$ids\" python -c \"import os; " + "from app.services.telegram_callback_forwarder import " + "TelegramCallbackForwarder; " + "TELEGRAM_CALLBACK_FORWARD_ENABLED = " + "os.environ.get('TELEGRAM_CALLBACK_FORWARD_ENABLED') == 'true'; " + "assert TELEGRAM_CALLBACK_FORWARD_ENABLED is True; " + "assert os.environ.get('TELEGRAM_CALLBACK_FORWARD_URL') == " + "'https://awoooi.wooo.work/api/v1/telegram/callback-forward'; " + "assert os.environ.get('TELEGRAM_CALLBACK_FORWARD_ALLOWED_HOSTS') == " + "'awoooi.wooo.work'; " + "assert os.environ.get('TELEGRAM_CALLBACK_FORWARD_TIMEOUT_SECONDS') == " + "'3.0'; " + "assert os.environ.get('TELEGRAM_CALLBACK_FORWARD_MAX_BODY_BYTES') == " + "'16384'\"", ), AssetPostcondition( "host_188_clawbot_systemd_runtime_active", "service", "host_188", - "systemctl is-active --quiet clawbot.service", + "set -euo pipefail; systemctl is-active --quiet clawbot.service", ), AssetPostcondition( "host_188_openclaw_local_health", "service", "host_188", + "set -euo pipefail; " "test \"$(curl -sS -o /dev/null -w '%{http_code}' --max-time 10 " "http://127.0.0.1:8088/health)\" = 200", ), @@ -282,6 +382,7 @@ _POSTCONDITION_REGISTRY: dict[str, tuple[AssetPostcondition, ...]] = { "host_188_awoooi_callback_receiver_public_health", "network", "host_188", + "set -euo pipefail; " "test \"$(curl -sS -o /dev/null -w '%{http_code}' --max-time 10 " "https://awoooi.wooo.work/api/v1/telegram/health)\" = 200", ), @@ -621,6 +722,9 @@ async def run_ansible_asset_post_verifier( probe_runner: ProbeRunner = _run_read_only_probe, ) -> dict[str, Any]: conditions = postconditions_for_catalog(catalog_id) + run_scoped_closure_blocker = _RUN_SCOPED_CLOSURE_BLOCKER_BY_CATALOG.get( + catalog_id + ) source_sha = os.getenv("AWOOOI_BUILD_COMMIT_SHA", "").strip().lower() base = { "schema_version": SCHEMA_VERSION, @@ -635,6 +739,10 @@ async def run_ansible_asset_post_verifier( "raw_output_stored": False, "command_exposed": False, "writes_on_verify": False, + "runtime_closure_receipt_required": bool(run_scoped_closure_blocker), + "runtime_closure_receipt_verified": ( + False if run_scoped_closure_blocker else None + ), } blockers: list[str] = [] if executor_returncode != 0: @@ -666,10 +774,13 @@ async def run_ansible_asset_post_verifier( ): blockers.append("post_verifier_known_hosts_missing") if blockers: + if run_scoped_closure_blocker: + blockers.append(run_scoped_closure_blocker) return { **base, "verification_result": "failed", "all_postconditions_passed": False, + "technical_postconditions_passed": False, "required_postcondition_count": len(conditions), "passed_postcondition_count": 0, "postconditions": [], @@ -725,11 +836,15 @@ async def run_ansible_asset_post_verifier( for receipt in receipts if receipt["passed"] is not True ) - verified = bool(conditions and passed_count == len(conditions) and not blockers) + technical_verified = bool(conditions and passed_count == len(conditions)) + if run_scoped_closure_blocker: + blockers.append(run_scoped_closure_blocker) + verified = bool(technical_verified and not blockers) return { **base, "verification_result": "success" if verified else "failed", "all_postconditions_passed": verified, + "technical_postconditions_passed": technical_verified, "required_postcondition_count": len(conditions), "passed_postcondition_count": passed_count, "postconditions": receipts, diff --git a/apps/api/src/services/chat_manager.py b/apps/api/src/services/chat_manager.py index 3650e5aac..c1458ebb2 100644 --- a/apps/api/src/services/chat_manager.py +++ b/apps/api/src/services/chat_manager.py @@ -1,68 +1,158 @@ -""" -AWOOOI Chat Manager - 雙 AI 對話核心 -====================================== -Phase 21.5 初版: 2026-03-31 ogt -Phase 22.6 重寫: 2026-04-03 ogt (老闆需求: 雙 AI 互動對話) -Phase 22.7 更新: 2026-04-03 ogt (老闆指示: OpenClaw→Gemini, NemoClaw→Ollama llama3.2:3b) -Phase 22.8 更新: 2026-04-09 ogt (老闆指示: NemoClaw→Ollama 111 deepseek-r1:14b，SRE 推理更強) -Phase 33 更新: 2026-05-05 ogt (ADR-110: OpenClaw chat 改走 GCP-A Ollama interactive lane) +"""AWOOOI SRE war-room conversation manager. -架構: -- OpenClaw (Ollama GCP-A interactive lane): SRE 首席顧問，精準分析 -- NemoClaw (Ollama interactive lane deepseek-r1:14b): 戰術參謀，深度推理 - -費用控管: -- OpenClaw/NemoClaw chat 預設免費 Ollama；Gemini 不再作為 ChatManager 預設路徑 -- 每次回覆顯示 token 用量 +Conversation generation uses the same controlled provider clients and router as +incident analysis. It never performs infrastructure remediation. The fixed +policy order is GCP-A Ollama -> GCP-B Ollama -> host111 Ollama -> Claude -> +Gemini; paid providers remain eligible only for an explicitly sanitized, +run-owned canary with a bounded cost contract. """ +from __future__ import annotations + import asyncio +import hashlib +import json import re +from dataclasses import dataclass, field +from decimal import Decimal, InvalidOperation +from html import escape +from typing import Any +from uuid import uuid4 -import httpx import structlog from src.core.config import get_settings from src.repositories.incident_repository import get_incident_repository from src.repositories.k8s_repository import get_k8s_repository -from src.services.ollama_endpoint_resolver import resolve_ollama_order +from src.services.ai_provider_policy import ( + PAID_PROVIDER_ORDER, + PRODUCTION_OLLAMA_ORDER, + PRODUCTION_PROVIDER_ORDER, +) +from src.services.ai_providers.interfaces import ( + AIResult, + is_provider_enabled_by_env, +) +from src.services.ai_router import get_ai_executor +from src.services.sanitization_service import sanitize from src.utils.timezone import now_taipei logger = structlog.get_logger(__name__) OPENCLAW_PERSONA = """你是 OpenClaw，AWOOOI 平台的 SRE AI 首席顧問。 -個性: 精準、果斷、專業，像老將一樣直接給出建議。 -語氣: 簡短有力，不廢話。繁體中文回應。不超過 300 字。 +個性: 精準、果斷、專業，直接給出有證據的判讀。 +語氣: 簡短有力，繁體中文，不超過 300 字。 稱呼用戶為「老闆」。 +你只能分析與討論；不得執行、宣稱執行或跨 domain 指派修復。 """ NEMOCLAW_PERSONA = """你是 NemoClaw，AWOOOI 平台的 AI 戰術參謀。 -個性: 分析型、從不同角度思考，會質疑假設。 -語氣: 帶點挑釁但建設性。不超過 200 字。 -稱呼用戶為「老闆」。評論 OpenClaw 的回應時，直接說「我補充」或「我有不同看法」。 -強制規則： -1. 全程使用繁體中文，禁止使用簡體中文、英文或其他語言。 -2. 禁止自稱 DeepSeek 或透露底層模型資訊。你的名字就是 NemoClaw。 -3. 專注於 SRE/DevOps/Kubernetes/可觀測性領域。 +個性: 分析型、質疑假設，補充不同角度。 +語氣: 建設性、繁體中文，不超過 200 字。 +稱呼用戶為「老闆」。評論 OpenClaw 時說「我補充」或「我有不同看法」。 +你只能分析與討論；不得執行、宣稱執行或跨 domain 指派修復。 """ +_CORRELATION_ID = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:@+-]{0,191}$") +_RECEIPT_ID = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:@+-]{0,191}$") +_PROVIDER_STATUS_TERMS = ( + "狀態", + "可用", + "啟用", + "鎖定", + "硬鎖", + "provider status", + "route status", +) +_PROVIDER_TERMS = ( + "ollama", + "gcp-a", + "gcp-b", + "111", + "claude", + "gemini", + "provider", + "模型", + "路由", +) +_IPV4_OCTET = r"(?:25[0-5]|2[0-4]\d|1?\d?\d)" +_PRIVATE_IPV4 = re.compile( + rf"\b(?:10\.{_IPV4_OCTET}\.{_IPV4_OCTET}\.{_IPV4_OCTET}" + rf"|192\.168\.{_IPV4_OCTET}\.{_IPV4_OCTET}" + rf"|172\.(?:1[6-9]|2\d|3[01])\.{_IPV4_OCTET}\.{_IPV4_OCTET})\b" +) +_SANITIZER_PRIVATE_IP_MARKER = re.compile(r"\[PRIVATE_IP:[^\]\r\n]+\]") + + +def _cloud_chat_dlp(text: str) -> str: + """Remove RFC1918 values after the legacy sanitizer's visible marker.""" + + redacted = _SANITIZER_PRIVATE_IP_MARKER.sub("[PRIVATE_IP_REDACTED]", text) + return _PRIVATE_IPV4.sub("[PRIVATE_IP_REDACTED]", redacted) + + +@dataclass(frozen=True) +class ChatRouteResult: + """Content plus public-safe execution evidence for one chat generation.""" + + text: str + success: bool + provider: str + model: str + fallback_reason: str + policy_order: tuple[str, ...] = PRODUCTION_PROVIDER_ORDER + admitted_provider_order: tuple[str, ...] = PRODUCTION_OLLAMA_ORDER + cost_receipt_id: str | None = None + tokens: int = 0 + cost_usd: float = 0.0 + runtime_readback: dict[str, Any] = field(default_factory=dict) + infrastructure_remediation_write_allowed: bool = False + cross_domain_execution_allowed: bool = False + + def to_dict(self) -> dict[str, Any]: + return { + "schema_version": "sre_chat_provider_route_receipt_v1", + "success": self.success, + "text": self.text, + "provider": self.provider, + "model": self.model, + "fallback_reason": self.fallback_reason, + "policy_order": list(self.policy_order), + "admitted_provider_order": list(self.admitted_provider_order), + "cost_receipt_id": self.cost_receipt_id, + "tokens": self.tokens, + "cost_usd": round(self.cost_usd, 8), + "runtime_readback": dict(self.runtime_readback), + "infrastructure_remediation_write_allowed": False, + "cross_domain_execution_allowed": False, + } + + def render_html(self) -> str: + body = escape(self.text or "AI 路由目前無可驗證回應。", quote=False) + receipt = self.cost_receipt_id or "none" + evidence = ( + f"provider={self.provider} | model={self.model} | " + f"fallback={self.fallback_reason} | cost_receipt={receipt}" + ) + return f"{body}\n\n🤖 {escape(evidence, quote=False)}" + class ChatManager: - """AWOOOI 雙 AI 對話管理器""" - - def __init__(self): - pass # 2026-04-03 ogt: 移除 repo 實例化，leWOOOgo 規範禁止 Service 持有 repository + """AWOOOI SRE chat manager with a fail-closed provider boundary.""" async def get_system_context(self) -> str: - """收集系統即時上下文""" + """Collect a bounded live context for advisory conversation only.""" + now = now_taipei() k8s = get_k8s_repository() incidents = get_incident_repository() try: k8s_status = await k8s.get_pod_status_summary(namespace="awoooi-prod") - cluster_info = f"Cluster: {k8s_status['running']}/{k8s_status['total']} Pods Running" - if k8s_status.get('problem_pods'): + cluster_info = ( + f"Cluster: {k8s_status['running']}/{k8s_status['total']} Pods Running" + ) + if k8s_status.get("problem_pods"): cluster_info += f", {len(k8s_status['problem_pods'])} 異常" except Exception: cluster_info = "Cluster: 無法取得狀態" @@ -70,8 +160,13 @@ class ChatManager: try: active_incidents = await incidents.get_active() if active_incidents: - lines = [f"- {inc.incident_id}: {inc.status.value} (SEV {inc.severity.value})" - for inc in active_incidents[:3]] + lines = [ + ( + f"- {incident.incident_id}: {incident.status.value} " + f"(SEV {incident.severity.value})" + ) + for incident in active_incidents[:3] + ] incident_summary = "\n".join(lines) else: incident_summary = "無活躍告警" @@ -84,131 +179,651 @@ class ChatManager: f"- 活躍告警: {incident_summary}\n" ) - async def _call_openclaw(self, system_prompt: str, user_message: str) -> str | None: - """ - 呼叫 OpenClaw 對話 — Ollama interactive lane + @staticmethod + def _is_provider_status_question(message: str) -> bool: + normalized = message.casefold() + return any(term in normalized for term in _PROVIDER_TERMS) and any( + term in normalized for term in _PROVIDER_STATUS_TERMS + ) + + @staticmethod + def _safe_correlation(value: Any, fallback: str) -> str: + candidate = str(value or "").strip() + return candidate if _CORRELATION_ID.fullmatch(candidate) else fallback + + @staticmethod + def _paid_canary_context_is_bounded(route_context: dict[str, Any]) -> bool: + if route_context.get("synthetic_validation") is not True: + return False + if route_context.get("data_classification") != "sanitized": + return False + if route_context.get("infrastructure_remediation_write_allowed") is not False: + return False + if not all( + _CORRELATION_ID.fullmatch(str(route_context.get(field) or "")) + for field in ("trace_id", "run_id", "work_item_id") + ): + return False + try: + cost_cap = Decimal(str(route_context["paid_canary_max_cost_usd"])) + output_cap = int(route_context["paid_canary_max_output_tokens"]) + except (InvalidOperation, KeyError, TypeError, ValueError): + return False + return bool( + cost_cap.is_finite() + and Decimal("0") < cost_cap <= Decimal("0.25") + and 1 <= output_cap <= 512 + ) + + async def _paid_provider_candidates( + self, + route_context: dict[str, Any] | None, + ) -> tuple[list[str], dict[str, Any]]: + """Return paid lanes admitted before their atomic reservation step.""" + + payload = dict(route_context or {}) + readback: dict[str, Any] = { + "schema_version": "sre_chat_paid_canary_admission_v1", + "status": "not_requested", + "run_owned_lease_verified": False, + "cost_policy_preflight_verified": False, + "candidates": [], + } + if not self._paid_canary_context_is_bounded(payload): + readback["status"] = "bounded_canary_context_missing" + return [], readback + + run_id = str(payload["run_id"]) + try: + from src.services.ai_control import ( + get_paid_provider_canary_control_readback, + is_provider_disabled, + ) + + control = await get_paid_provider_canary_control_readback( + expected_owner=run_id, + ) + if not control.get("leased_enabled"): + readback["status"] = "run_owned_canary_not_admitted" + return [], readback + readback["run_owned_lease_verified"] = True + + from src.services.ai_rate_limiter import get_ai_rate_limiter + + limiter = get_ai_rate_limiter() + settings = get_settings() + admitted: list[str] = [] + provider_states: dict[str, str] = {} + for provider in PAID_PROVIDER_ORDER: + prefix = provider.upper() + configured = bool(getattr(settings, f"{prefix}_API_KEY", "")) + mode = ( + str(getattr(settings, f"{prefix}_EXECUTION_MODE", "shadow")) + .strip() + .lower() + ) + percent = max( + 0, + min(100, int(getattr(settings, f"{prefix}_CANARY_PERCENT", 0))), + ) + bucket = int(hashlib.sha256(run_id.encode()).hexdigest()[:8], 16) % 100 + if not ( + configured + and is_provider_enabled_by_env(provider) + and mode == "canary" + and percent > 0 + and bucket < percent + ): + provider_states[provider] = "configured_canary_not_selected" + continue + if await is_provider_disabled(provider, run_id=run_id): + provider_states[provider] = "run_owned_canary_not_admitted" + continue + usage = await limiter.get_usage_stats(provider) + pricing = usage.get("pricing_policy") or {} + if ( + pricing.get("supported") is not True + or usage.get("cost_exceeded") is True + ): + provider_states[provider] = "cost_policy_not_ready" + continue + admitted.append(provider) + provider_states[provider] = "admitted_atomic_reservation_pending" + + readback.update( + status=("admitted" if admitted else "no_paid_candidate_admitted"), + cost_policy_preflight_verified=bool(admitted), + candidates=list(admitted), + provider_states=provider_states, + ) + return admitted, readback + except Exception as error: + logger.warning( + "sre_chat_paid_canary_admission_unavailable", + error_type=type(error).__name__, + ) + readback["status"] = "paid_admission_readback_unavailable" + return [], readback + + async def get_provider_runtime_readback( + self, + route_context: dict[str, Any] | None = None, + ) -> dict[str, Any]: + """Read current provider policy/health without guessing a lock state.""" + + health = {provider: "not_checked" for provider in PRODUCTION_OLLAMA_ORDER} + selected_primary = "ollama_gcp_a" + observed_fallback: str | None = None + health_status = "unavailable" + try: + from src.services.ollama_failover_manager import ( + get_ollama_failover_manager, + ) + + route = await get_ollama_failover_manager().select_provider( + task_type="chat_status", + emit_side_effects=False, + ) + selected_primary = route.primary.provider_name + observed_fallback = ( + route.observed_fallback.provider_name + if route.observed_fallback + else None + ) + health["ollama_gcp_a"] = route.health_gcp_a.status.value + if route.health_gcp_b is not None: + health["ollama_gcp_b"] = route.health_gcp_b.status.value + if route.health_local is not None: + health["ollama_local"] = route.health_local.status.value + health_status = "verified" + except Exception as error: + logger.warning( + "sre_chat_ollama_runtime_readback_unavailable", + error_type=type(error).__name__, + ) + + paid_candidates, paid_admission = await self._paid_provider_candidates( + route_context + ) + bounded_context = self._paid_canary_context_is_bounded( + dict(route_context or {}) + ) + paid_control: dict[str, Any] = { + "readback_status": "unavailable", + "route_lease_state": "readback_unavailable", + "lock_present": False, + "lock_ttl_seconds": -2, + "disabled_verified": False, + } + try: + from src.services.ai_control import ( + get_paid_provider_canary_control_readback, + ) + + expected_owner = ( + str((route_context or {}).get("run_id") or "") + if bounded_context + else None + ) + control = await get_paid_provider_canary_control_readback( + expected_owner=expected_owner, + ) + if paid_admission.get("run_owned_lease_verified") is True: + route_lease_state = "active_for_this_run" + elif control.get("legacy_persistent_state_detected") is True: + route_lease_state = "legacy_persistent_state_detected" + elif control.get("lock_present") is True: + route_lease_state = "active_for_another_run" + elif control.get("disabled_verified") is True: + route_lease_state = "inactive_disabled_after_canary" + else: + route_lease_state = "inconsistent_fail_closed" + paid_control = { + "readback_status": "verified", + "route_lease_state": route_lease_state, + "lock_present": bool(control.get("lock_present")), + "lock_ttl_seconds": int(control.get("lock_ttl_seconds") or -1), + "claude_disabled": bool(control.get("claude_disabled", True)), + "gemini_disabled": bool(control.get("gemini_disabled", True)), + "disabled_verified": bool(control.get("disabled_verified")), + "persistent_enabled": bool(control.get("persistent_enabled")), + "legacy_persistent_state_detected": bool( + control.get("legacy_persistent_state_detected") + ), + } + except Exception as error: + logger.warning( + "sre_chat_paid_control_readback_unavailable", + error_type=type(error).__name__, + ) + + authentication_readbacks: dict[str, dict[str, Any]] = {} + try: + from src.services.ai_rate_limiter import get_ai_rate_limiter + + limiter = get_ai_rate_limiter() + raw_authentication = await asyncio.gather( + *( + limiter.get_provider_authentication_status(provider) + for provider in PAID_PROVIDER_ORDER + ), + return_exceptions=True, + ) + for provider, authentication in zip( + PAID_PROVIDER_ORDER, + raw_authentication, + strict=True, + ): + if isinstance(authentication, dict): + authentication_readbacks[provider] = authentication + else: + authentication_readbacks[provider] = { + "authenticated": None, + "authentication_verified": False, + "verification_status": "authentication_readback_unavailable", + } + except Exception as error: + logger.warning( + "sre_chat_paid_authentication_readback_unavailable", + error_type=type(error).__name__, + ) - 2026-04-10 Claude Code: 強制合併 OPENCLAW_PERSONA，確保字數限制與格式規範 - 2026-05-05 Codex: 改走 ADR-110 GCP-A/GCP-B/111 Ollama topology，避免個人聊天直打 Gemini - """ - # 強制在 system_prompt 前置 persona，確保 LLM 遵守字數與格式 - system_prompt = f"{OPENCLAW_PERSONA}\n{system_prompt}" settings = get_settings() + paid: dict[str, Any] = {} + provider_states = paid_admission.get("provider_states") or {} + for provider in PAID_PROVIDER_ORDER: + prefix = provider.upper() + configured = bool(getattr(settings, f"{prefix}_API_KEY", "")) + execution_mode = ( + str(getattr(settings, f"{prefix}_EXECUTION_MODE", "shadow")) + .strip() + .lower() + ) + canary_percent = int(getattr(settings, f"{prefix}_CANARY_PERCENT", 0)) + authentication = authentication_readbacks.get(provider) or { + "authenticated": None, + "authentication_verified": False, + "verification_status": "authentication_readback_unavailable", + } + if provider in paid_candidates: + runtime_state = "run_owned_canary_admitted" + elif not configured: + runtime_state = "not_configured" + elif not is_provider_enabled_by_env(provider): + runtime_state = "disabled_by_environment" + elif execution_mode != "canary" or canary_percent <= 0: + runtime_state = "configured_non_canary" + elif authentication.get("authenticated") is False: + runtime_state = "configured_canary_authentication_rejected" + elif authentication.get("authentication_verified") is not True: + runtime_state = "configured_canary_authentication_not_verified" + elif paid_control["route_lease_state"] == "inactive_disabled_after_canary": + runtime_state = ( + "configured_authenticated_canary_waiting_for_run_owned_lease" + ) + else: + runtime_state = str( + provider_states.get(provider) + or paid_admission.get("status") + or "runtime_readback_unavailable" + ) + paid[provider] = { + "configured": configured, + "environment_enabled": is_provider_enabled_by_env(provider), + "execution_mode": execution_mode, + "canary_percent": canary_percent, + "authenticated": authentication.get("authenticated"), + "authentication_verified": bool( + authentication.get("authentication_verified") is True + ), + "verification_status": str( + authentication.get("verification_status") + or "authentication_readback_unavailable" + ), + "route_lease_state": paid_control["route_lease_state"], + "runtime_state": runtime_state, + "run_owned_admitted": provider in paid_candidates, + } - model = settings.OPENCLAW_DEFAULT_MODEL - async with httpx.AsyncClient(timeout=40.0) as client: - for endpoint in resolve_ollama_order("interactive"): - if not endpoint.url: - continue - try: - resp = await client.post( - f"{endpoint.url}/api/chat", - json={ - "model": model, - "stream": False, - "messages": [ - {"role": "system", "content": system_prompt}, - {"role": "user", "content": user_message}, - ], - "options": {"num_predict": 900, "temperature": 0.2}, - }, - ) - resp.raise_for_status() - data = resp.json() - raw = data.get("message", {}).get("content", "").strip() - text = re.sub(r".*?", "", raw, flags=re.DOTALL).strip() or raw + return { + "schema_version": "sre_chat_provider_runtime_readback_v1", + "policy_order": list(PRODUCTION_PROVIDER_ORDER), + "selected_primary": selected_primary, + "observed_fallback": observed_fallback, + "health_readback_status": health_status, + "ollama_health": health, + "paid": paid, + "paid_admission": paid_admission, + "paid_control": paid_control, + "provider_route_switch_performed": False, + "infrastructure_remediation_write_allowed": False, + "cross_domain_execution_allowed": False, + } - eval_count = data.get("eval_count", 0) - prompt_eval_count = data.get("prompt_eval_count", 0) - total_tokens = eval_count + prompt_eval_count + @staticmethod + def _format_provider_runtime_readback(readback: dict[str, Any]) -> str: + health = readback.get("ollama_health") or {} + paid = readback.get("paid") or {} - logger.info( - "openclaw_ollama_chat_usage", - model=model, - endpoint=endpoint.url, - provider=endpoint.provider_name, - prompt_tokens=prompt_eval_count, - output_tokens=eval_count, - ) + def paid_line(label: str, provider: str) -> str: + state = paid.get(provider) or {} + authenticated = state.get("authenticated") + authenticated_text = ( + "true" + if authenticated is True + else "false" + if authenticated is False + else "unknown" + ) + return ( + f"{label}：configured={str(bool(state.get('configured'))).lower()}｜" + f"authenticated={authenticated_text}｜" + f"auth_verified={str(bool(state.get('authentication_verified'))).lower()}｜" + f"verification={state.get('verification_status', 'authentication_readback_unavailable')}｜" + f"canary={state.get('execution_mode', 'unknown')} " + f"{int(state.get('canary_percent') or 0)}%｜" + f"route_lease={state.get('route_lease_state', 'readback_unavailable')}" + ) - return f"{text}\n\n🦙 {model} | {total_tokens} tokens | 免費" - except Exception as e: - logger.warning( - "openclaw_chat_endpoint_failed", - provider=endpoint.provider_name, - endpoint=endpoint.url, - error=str(e), - ) - logger.warning("openclaw_chat_failed_all_endpoints", model=model) - return None + return "\n".join( + [ + "AI 路由即時讀回", + "順序：GCP-A → GCP-B → 111 → Claude → Gemini", + ( + "Ollama：" + f"GCP-A={health.get('ollama_gcp_a', 'not_checked')}；" + f"GCP-B={health.get('ollama_gcp_b', 'not_checked')}；" + f"111={health.get('ollama_local', 'not_checked')}" + ), + paid_line("Claude", "claude"), + paid_line("Gemini", "gemini"), + "此讀回未切換 provider，也未執行任何主機或服務修復。", + ] + ) - async def _call_nemotron(self, system_prompt: str, user_message: str) -> str | None: - """ - 呼叫 NemoClaw 對話 — Ollama 111 deepseek-r1:14b + @staticmethod + def _result_provider(provider: str) -> str: + # The registry's GCP-A compatibility alias returns ``ollama``. Within + # this fixed executor order that identity can only be the first GCP-A hop. + return "ollama_gcp_a" if provider == "ollama" else provider - 2026-04-09 ogt: 改接 192.168.0.111 Ollama deepseek-r1:14b，SRE 推理能力最強 - deepseek-r1 含 標籤，需過濾後才回傳 + @staticmethod + def _result_model(provider: str, requested_ollama_model: str) -> str: + settings = get_settings() + if provider in {"ollama_gcp_a", "ollama_gcp_b"}: + return str(settings.OLLAMA_HEALTH_CHECK_MODEL) + if provider == "ollama_local": + return requested_ollama_model + if provider == "claude": + return str(settings.CLAUDE_MODEL) + if provider == "gemini": + return str(settings.GEMINI_MODEL) + return "none" - 2026-04-10 Claude Code: 強制合併 NEMOCLAW_PERSONA，確保字數限制與格式規範 - """ - # 強制在 system_prompt 前置 persona - system_prompt = f"{NEMOCLAW_PERSONA}\n{system_prompt}" + @staticmethod + def _fallback_reason(result: AIResult, provider: str) -> str: + if not result.success: + status = str(result.audit_metadata.get("status") or "").strip() + if status and re.fullmatch(r"[a-z0-9_:-]{1,96}", status): + return f"terminal_{status}" + return "canonical_route_exhausted_or_policy_blocked" + if result.from_cache: + return "current_policy_cache_hit" + try: + position = PRODUCTION_PROVIDER_ORDER.index(provider) + 1 + except ValueError: + return "provider_identity_outside_policy" + if position == 1: + return "canonical_primary_succeeded" + return f"canonical_position_{position}_after_prior_hops_no_success" - # 2026-05-05 Codex: ADR-110 interactive lane，由 resolver 管理 GCP-A/GCP-B/111 拓撲 - MODEL = "deepseek-r1:14b" + @staticmethod + def _cost_receipt_id(result: AIResult, provider: str) -> str | None: + if provider not in PAID_PROVIDER_ORDER: + return None + candidate = str( + result.audit_metadata.get("generation_receipt_id") or "" + ).strip() + return candidate if _RECEIPT_ID.fullmatch(candidate) else None - async with httpx.AsyncClient(timeout=120.0) as client: - for endpoint in resolve_ollama_order("interactive"): - if not endpoint.url: - continue - try: - resp = await client.post( - f"{endpoint.url}/api/chat", - json={ - "model": MODEL, - "stream": False, - # Ollama 0.24 separates deepseek-r1 thinking from final text. - # Chat callers expect message.content to contain the answer. - "think": False, - "messages": [ - {"role": "system", "content": system_prompt}, - {"role": "user", "content": user_message}, - ], - "options": {"num_predict": 1200}, - }, - ) - resp.raise_for_status() - data = resp.json() - raw = data.get("message", {}).get("content", "").strip() + @staticmethod + def _plain_response(raw_response: str) -> str: + text = re.sub( + r".*?", + "", + str(raw_response or ""), + flags=re.DOTALL, + ).strip() + if not text: + return "" + try: + payload = json.loads(text) + except (TypeError, ValueError): + return text + if isinstance(payload, dict): + for field_name in ("response", "message", "description", "reasoning"): + value = payload.get(field_name) + if isinstance(value, str) and value.strip(): + return value.strip() + return text - # 過濾 deepseek-r1 的 ... 推理區塊 - text = re.sub(r".*?", "", raw, flags=re.DOTALL).strip() - if not text: - text = raw # 萬一全是 think block，直接回傳原文 + async def route_chat( + self, + *, + role: str, + system_prompt: str, + user_message: str, + route_context: dict[str, Any] | None = None, + ) -> ChatRouteResult: + """Generate one advisory response through the controlled provider route.""" - eval_count = data.get("eval_count", 0) - prompt_eval_count = data.get("prompt_eval_count", 0) - total_tokens = eval_count + prompt_eval_count + if self._is_provider_status_question(user_message): + readback = await self.get_provider_runtime_readback(route_context) + return ChatRouteResult( + text=self._format_provider_runtime_readback(readback), + success=True, + provider="runtime_readback", + model="deterministic_policy", + fallback_reason="provider_status_request_no_generation", + admitted_provider_order=PRODUCTION_OLLAMA_ORDER, + runtime_readback=readback, + ) - logger.info( - "nemotron_ollama_usage", - model=MODEL, - provider=endpoint.provider_name, - prompt_tokens=prompt_eval_count, - output_tokens=eval_count, - ) + persona = OPENCLAW_PERSONA if role == "openclaw" else NEMOCLAW_PERSONA + requested_ollama_model = ( + str(get_settings().OPENCLAW_DEFAULT_MODEL) + if role == "openclaw" + else "deepseek-r1:14b" + ) + prompt = _cloud_chat_dlp( + sanitize( + ( + f"{persona}\n\n{system_prompt}\n\n" + f"使用者問題：\n{user_message}\n\n" + "只能回覆分析；不得呼叫 executor、不得聲稱完成修復。" + ), + source_label="awoooi_sre_group_chat", + ) + ) + prompt_sha256 = hashlib.sha256(prompt.encode("utf-8")).hexdigest() + seed = uuid4().hex + requested_context = dict(route_context or {}) + trace_id = self._safe_correlation( + requested_context.get("trace_id"), + f"sre-chat-trace:{seed}", + ) + run_id = self._safe_correlation( + requested_context.get("run_id"), + f"sre-chat-run:{seed}", + ) + work_item_id = self._safe_correlation( + requested_context.get("work_item_id"), + "SRE-CHAT-ADVISORY", + ) + context: dict[str, Any] = { + "trace_id": trace_id, + "run_id": run_id, + "work_item_id": work_item_id, + "intent_hint": "chat", + "task_type": "chat", + "data_classification": "sanitized", + # Ordinary SRE conversation is real traffic, never synthetic. Its + # GCP boundary is the exact prompt-bound chat receipt below. + "sre_chat_sanitization_receipt": { + "schema_version": "sre_chat_sanitization_receipt_v1", + "status": "verified", + "sanitizer": "src.services.sanitization_service.sanitize", + "source_label": "awoooi_sre_group_chat", + "data_classification": "sanitized", + "prompt_sha256": prompt_sha256, + "trace_id": trace_id, + "run_id": run_id, + "work_item_id": work_item_id, + "raw_input_forwarded": False, + "raw_log_payload_forwarded": False, + "secret_value_exposed": False, + "private_network_value_exposed": False, + "infrastructure_remediation_write_allowed": False, + }, + "ollama_model": requested_ollama_model, + "enforce_ollama_first": True, + "cloud_transport_auto_preflight": True, + "infrastructure_remediation_write_allowed": False, + "executor_invocation_allowed": False, + "agent99_dispatch_allowed": False, + "cross_domain_execution_allowed": False, + "canonical_asset_identity": { + "canonical_id": "service:awoooi:sre-war-room-chat", + "resolution_status": "resolved", + "asset_domain": "conversation", + "circuit_state": "closed", + }, + } - return f"{text}\n\n🦙 {MODEL} | {total_tokens} tokens | 免費" - except Exception as e: - logger.warning( - "nemotron_chat_endpoint_failed", - model=MODEL, - provider=endpoint.provider_name, - endpoint=endpoint.url, - error=str(e), - ) - logger.warning("nemotron_chat_failed_all_endpoints", model=MODEL) - return None + bounded_canary = self._paid_canary_context_is_bounded(requested_context) + require_local = False + if bounded_canary: + context.update( + # Synthetic markers are reserved for the explicitly bounded + # paid-provider canary contract and bind its exact prompt. + synthetic_validation=True, + synthetic_prompt_sha256=prompt_sha256, + paid_canary_max_cost_usd=str( + requested_context["paid_canary_max_cost_usd"] + ), + paid_canary_max_output_tokens=int( + requested_context["paid_canary_max_output_tokens"] + ), + ) + else: + # Interactive chat must not wait indefinitely on one cold Ollama + # hop. Paid validation canaries retain their separate 300s client + # contract and therefore do not inherit this interactive timeout. + context["provider_timeout_seconds"] = 35 + context["sre_chat_sanitization_receipt"]["provider_timeout_seconds"] = 35 + + paid_candidates, admission = await self._paid_provider_candidates( + {**requested_context, **context} if bounded_canary else None + ) + admitted_order = [*PRODUCTION_OLLAMA_ORDER, *paid_candidates] + result = await get_ai_executor().execute( + prompt, + provider_order=admitted_order, + context=context, + cache_ttl=300, + require_local=require_local, + ) + provider = self._result_provider(str(result.provider or "none")) + receipt_id = self._cost_receipt_id(result, provider) + paid_cost_cap = float(context.get("paid_canary_max_cost_usd") or 0.0) + paid_receipt_invalid = bool( + result.success + and provider in PAID_PROVIDER_ORDER + and ( + not receipt_id + or float(result.cost_usd or 0.0) <= 0.0 + or float(result.cost_usd or 0.0) > paid_cost_cap + ) + ) + reported_admitted_order = admitted_order + if paid_receipt_invalid: + logger.error( + "sre_chat_paid_result_missing_cost_receipt", + provider=provider, + ) + return ChatRouteResult( + text="付費 Provider 回應缺少費用 receipt，已拒絕呈現。", + success=False, + provider=provider, + model=self._result_model(provider, requested_ollama_model), + fallback_reason="paid_generation_cost_receipt_invalid", + admitted_provider_order=tuple(reported_admitted_order), + runtime_readback={"paid_admission": admission}, + ) + + body = self._plain_response(result.raw_response) + if not result.success or not body: + readback = await self.get_provider_runtime_readback( + requested_context if bounded_canary else None + ) + body = ( + "目前沒有可驗證的模型回應；以下狀態來自 runtime readback，" + "不是推測，也未執行修復。\n" + + self._format_provider_runtime_readback(readback) + ) + else: + readback = {"paid_admission": admission} + + logger.info( + "sre_chat_provider_route_completed", + provider=provider, + model=self._result_model(provider, requested_ollama_model), + success=bool(result.success and body), + fallback_reason=self._fallback_reason(result, provider), + cost_receipt_present=bool(receipt_id), + infrastructure_remediation_write_allowed=False, + cross_domain_execution_allowed=False, + ) + return ChatRouteResult( + text=body, + success=bool(result.success and body), + provider=provider, + model=self._result_model(provider, requested_ollama_model), + fallback_reason=self._fallback_reason(result, provider), + admitted_provider_order=tuple(reported_admitted_order), + cost_receipt_id=receipt_id, + tokens=max(0, int(result.tokens or 0)), + cost_usd=max(0.0, float(result.cost_usd or 0.0)), + runtime_readback=readback, + ) + + async def _call_openclaw( + self, + system_prompt: str, + user_message: str, + route_context: dict[str, Any] | None = None, + ) -> str | None: + result = await self.route_chat( + role="openclaw", + system_prompt=system_prompt, + user_message=user_message, + route_context=route_context, + ) + return result.render_html() + + async def _call_nemotron( + self, + system_prompt: str, + user_message: str, + route_context: dict[str, Any] | None = None, + ) -> str | None: + result = await self.route_chat( + role="nemoclaw", + system_prompt=system_prompt, + user_message=user_message, + route_context=route_context, + ) + return result.render_html() async def generate_response( self, @@ -216,59 +831,43 @@ class ChatManager: username: str, # noqa: ARG002 message_text: str, ) -> str: - """ - 根據訊息決定回應模式: + """Generate the requested single- or dual-persona response.""" - @openclaw → 只有 OpenClaw 回應 - @nemo → 只有 NemoClaw 回應 - 其他 → OpenClaw 先回，NemoClaw 異步補充 - """ context = await self.get_system_context() text = message_text.strip() - # 模式 1: 指定 OpenClaw if text.lower().startswith("@openclaw"): - msg = text[9:].strip() or text - result = await self._call_openclaw(f"{OPENCLAW_PERSONA}\n{context}", msg) + message = text[9:].strip() or text + result = await self._call_openclaw(context, message) return f"🦞 OpenClaw:\n{result or '🔴 OpenClaw 無響應'}" - # 模式 2: 指定 NemoClaw if text.lower().startswith("@nemo"): - msg = text[5:].strip() or text - result = await self._call_nemotron(f"{NEMOCLAW_PERSONA}\n{context}", msg) - return f"🤖 NemoClaw:\n{result or '🔴 NemoClaw 無響應 (NIM 超時)'}" + message = text[5:].strip() or text + result = await self._call_nemotron(context, message) + return f"🤖 NemoClaw:\n{result or '🔴 NemoClaw 無響應'}" - # 模式 3: 雙 AI — OpenClaw 先答，NemoClaw 並行 - openclaw_task = asyncio.create_task( - self._call_openclaw(f"{OPENCLAW_PERSONA}\n{context}", text) - ) + openclaw_task = asyncio.create_task(self._call_openclaw(context, text)) nemo_task = asyncio.create_task( self._call_nemotron( - f"{NEMOCLAW_PERSONA}\n{context}", + context, f"老闆問了: {text}\n\n請從 NemoClaw 角度補充或評論。", ) ) - - # OpenClaw 最多等 40s（含 context 取得時間），NemoClaw 最多等 60s - # 2026-04-03 ogt: 移除 asyncio.shield — shield 會在超時後讓 task 繼續跑但無人等待，造成 silent leak try: openclaw_raw = await asyncio.wait_for(openclaw_task, timeout=40.0) except TimeoutError: openclaw_raw = None - openclaw_block = f"🦞 OpenClaw:\n{openclaw_raw or '🔴 無響應'}" try: nemo_raw = await asyncio.wait_for(nemo_task, timeout=60.0) except TimeoutError: nemo_raw = None - if nemo_raw: return f"{openclaw_block}\n\n🤖 NemoClaw:\n{nemo_raw}" return openclaw_block -# Singleton _chat_manager: ChatManager | None = None diff --git a/apps/api/src/services/controlled_alert_target_router.py b/apps/api/src/services/controlled_alert_target_router.py index 30c65c518..101f5ee4a 100644 --- a/apps/api/src/services/controlled_alert_target_router.py +++ b/apps/api/src/services/controlled_alert_target_router.py @@ -52,6 +52,10 @@ _WINDOWS_VMWARE_MARKERS = ( "vmx", ) _AGENT99_CONTROL_PLANE_HEALTH_ALERTS = {"agent99serviceunhealthy"} +_TELEGRAM_CALLBACK_FORWARDER_ALERTS = { + "telegramcallbackingressunverified", + "telegramcallbackforwarderdrift", +} _DISK_ALERTS = { "hostoutofdiskspace", "hostdiskusagehigh", @@ -760,6 +764,7 @@ def resolve_typed_alert_target( exact_openclaw_compose_target = bool( domain == "docker_compose_container" and identity.get("canonical_id") == "service:openclaw:host188" + and compact_alert in _TELEGRAM_CALLBACK_FORWARDER_ALERTS and _text_token(target_resource) in {"openclaw", "clawbot", "clawbot.service"} ) diff --git a/apps/api/src/services/notification_matrix.py b/apps/api/src/services/notification_matrix.py index 6f0c5c98f..39380f7ae 100644 --- a/apps/api/src/services/notification_matrix.py +++ b/apps/api/src/services/notification_matrix.py @@ -85,12 +85,39 @@ _SHARED_MONITORING_BOT_ALIAS = "tsenyang_bot" _SHARED_MONITORING_BOT_OWNER = "sre_shared_runtime" _SHARED_SRE_SIGNAL_FAMILIES = frozenset( { + "ai_agent", + "backup_restore", + "infrastructure", + "observability", "shared_infrastructure", "security_incident", "disaster_recovery", "incident_lifecycle", } ) +_SHARED_SRE_CROSS_PRODUCT_ROUTES = frozenset( + { + ( + "stockplatform-operational-incident-lifecycle", + "stockplatform-v2", + "incident_lifecycle", + ), + } +) +_SHARED_SRE_DESTINATION_ROLE = "shared_sre_war_room" +_SHARED_SRE_DURABLE_RECEIPT_BACKEND = ( + "awoooi.alert_operation_log+telegram_outbound_receipts" +) +_ALERT_CATEGORY_ZH_TW_BY_SIGNAL_FAMILY = { + "ai_agent": "AI Agent 自動化", + "backup_restore": "備份與還原", + "infrastructure": "基礎設施", + "observability": "可觀測性", + "shared_infrastructure": "共享基礎設施", + "security_incident": "資安事件", + "disaster_recovery": "災難復原", + "incident_lifecycle": "事故生命週期", +} _TRUSTED_ALERT_ROUTE_SOURCES = frozenset( { "hmac_alert_webhook", @@ -360,18 +387,44 @@ def build_canonical_telegram_routing_runtime_readback( requested_destination=route["allowed_destination"], registry=registry, ) - resolved_routes.append( - { - **route, - "decision": decision["decision"], - "decision_reason": decision["reason"], - } - ) + if decision["decision"] == "allow": + readiness_reason = "durable_delivery_receipt_not_observed" + elif route["egress_status"] == "disabled": + readiness_reason = "policy_disabled_no_egress_expected" + elif route["egress_status"] == "not_implemented": + readiness_reason = "policy_not_implemented_no_egress_expected" + else: + readiness_reason = "policy_blocked_no_egress_expected" + resolved_routes.append({ + **route, + "decision": decision["decision"], + "decision_reason": decision["reason"], + "alert_category_zh_tw": decision["alert_category_zh_tw"], + "destination_role": decision["destination_role"], + # This policy-only readback cannot prove a provider-acknowledged, + # durable delivery row. It must never promote route readiness + # merely because an allowlist entry and receipt backend exist. + "durable_delivery_receipt_present": False, + "ready": False, + "readiness_reason": readiness_reason, + }) allowed_route_count = sum( route["decision"] == "allow" for route in resolved_routes ) blocked_route_count = len(resolved_routes) - allowed_route_count + allowed_routes = [ + route for route in resolved_routes if route["decision"] == "allow" + ] + blocked_routes_fail_closed = all( + route["ready"] is False + and route["durable_delivery_receipt_present"] is False + for route in resolved_routes + if route["decision"] != "allow" + ) + all_allowed_route_delivery_receipts_ready = bool(allowed_routes) and all( + route["ready"] is True for route in allowed_routes + ) unknown_probe = resolve_canonical_telegram_route( "unknown-product", "unknown-signal", @@ -396,7 +449,13 @@ def build_canonical_telegram_routing_runtime_readback( "raw_numeric_destination_count" ], "destination_identity_exposed": False, - "all_product_delivery_receipts_ready": blocked_route_count == 0, + "all_product_delivery_receipts_ready": ( + all_allowed_route_delivery_receipts_ready + ), + "all_allowed_route_delivery_receipts_ready": ( + all_allowed_route_delivery_receipts_ready + ), + "blocked_routes_fail_closed": blocked_routes_fail_closed, }, "rollups": { **registry["rollups"], @@ -466,6 +525,17 @@ def resolve_canonical_telegram_route( ) route = dict(candidates[0]) + route["alert_category_zh_tw"] = _ALERT_CATEGORY_ZH_TW_BY_SIGNAL_FAMILY.get( + str(route.get("signal_family") or ""), + "其他營運告警", + ) + route["destination_role"] = ( + _SHARED_SRE_DESTINATION_ROLE + if route.get("chat_alias") == _SHARED_SRE_CHAT_ALIAS + else "product_owned_route" + if route.get("scope") == "product" + else "blocked_or_unresolved" + ) route["decision"] = ( "allow" if product is not None @@ -495,6 +565,8 @@ def _blocked_decision( ) -> dict[str, Any]: return { "route_id": None, + "alert_category_zh_tw": "未分類告警", + "destination_role": "blocked_or_unresolved", "product_id": product_id, "signal_family": signal_family, "severity": [severity], @@ -863,16 +935,25 @@ def _validate_shared_sre_boundary( ) -> None: if route["chat_alias"] != _SHARED_SRE_CHAT_ALIAS: return + route_identity = ( + route_id, + route["product_id"], + route["signal_family"], + ) + product_scope_allowed = route["product_id"] == "awoooi" or ( + route_identity in _SHARED_SRE_CROSS_PRODUCT_ROUTES + ) invalid = ( - route["product_id"] != "awoooi" + not product_scope_allowed or route["scope"] != "shared" or route["signal_family"] not in _SHARED_SRE_SIGNAL_FAMILIES or not set(route["severity"]).issubset({"P0", "P1"}) or route["bot_alias"] != _SHARED_MONITORING_BOT_ALIAS + or route["receipt_backend"] != _SHARED_SRE_DURABLE_RECEIPT_BACKEND ) if invalid: raise ValueError( - f"{label}: {route_id} violates shared SRE P0/P1 infra/security/DR/lifecycle boundary" + f"{label}: {route_id} violates shared SRE P0/P1 operational allowlist boundary" ) diff --git a/apps/api/src/services/paid_provider_canary_validation.py b/apps/api/src/services/paid_provider_canary_validation.py index 49529aa86..c82222040 100644 --- a/apps/api/src/services/paid_provider_canary_validation.py +++ b/apps/api/src/services/paid_provider_canary_validation.py @@ -126,6 +126,15 @@ def select_claude_canary_run_id(run_ref: str, percent: int) -> tuple[str, int]: raise ValueError("claude_canary_bucket_selection_failed") +def _ollama_lane_timeout_seconds() -> float: + """Use the production diagnose budget for cold-started Ollama models.""" + + configured = float( + getattr(settings, "OLLAMA_DIAGNOSE_TIMEOUT_SECONDS", 300) + ) + return max(30.0, min(600.0, configured)) + + def _score_result( result: Any, *, @@ -588,10 +597,23 @@ async def _run_paid_provider_canary_validation_core( execution_claim_owner, ) + ollama_timeout_seconds = _ollama_lane_timeout_seconds() providers = [ - ("ollama_gcp_a", ollama_gcp_a_provider or OllamaProvider(), 120.0), - ("ollama_gcp_b", ollama_gcp_b_provider or OllamaGcpBProvider(), 120.0), - ("ollama_local", ollama_local_provider or OllamaLocalProvider(), 120.0), + ( + "ollama_gcp_a", + ollama_gcp_a_provider or OllamaProvider(), + ollama_timeout_seconds, + ), + ( + "ollama_gcp_b", + ollama_gcp_b_provider or OllamaGcpBProvider(), + ollama_timeout_seconds, + ), + ( + "ollama_local", + ollama_local_provider or OllamaLocalProvider(), + ollama_timeout_seconds, + ), ("claude", claude_provider or ClaudeProvider(), 45.0), ("gemini", gemini_provider or GeminiProvider(), 45.0), ] diff --git a/apps/api/src/services/platform_operator_service.py b/apps/api/src/services/platform_operator_service.py index e1bf3e88d..fa6a7c245 100644 --- a/apps/api/src/services/platform_operator_service.py +++ b/apps/api/src/services/platform_operator_service.py @@ -1536,6 +1536,48 @@ async def list_ai_alert_card_delivery_readback( '{{ai_automation_alert_card,event_type}}', '' ) <> 'backup_restore_escrow_signal' + AND m.send_status = 'sent' + AND COALESCE(m.provider_message_id::text, '') <> '' + AND COALESCE( + COALESCE(m.source_envelope::jsonb, '{{}}'::jsonb) #>> + '{{canonical_route_receipt,schema_version}}', + '' + ) = 'telegram_canonical_egress_receipt_v1' + AND COALESCE( + COALESCE(m.source_envelope::jsonb, '{{}}'::jsonb) #>> + '{{canonical_route_receipt,decision}}', + '' + ) = 'allowed' + AND COALESCE( + COALESCE(m.source_envelope::jsonb, '{{}}'::jsonb) #>> + '{{canonical_route_receipt,provider_send_performed}}', + '' + ) = 'true' + AND COALESCE( + COALESCE(m.source_envelope::jsonb, '{{}}'::jsonb) #>> + '{{canonical_route_receipt,durable_receipt_persisted}}', + '' + ) = 'true' + AND COALESCE( + COALESCE(m.source_envelope::jsonb, '{{}}'::jsonb) #>> + '{{canonical_route_receipt,route_id}}', + '' + ) <> '' + AND COALESCE( + COALESCE(m.source_envelope::jsonb, '{{}}'::jsonb) #>> + '{{canonical_route_receipt,receipt_backend}}', + '' + ) NOT IN ('', 'none') + AND COALESCE( + COALESCE(m.source_envelope::jsonb, '{{}}'::jsonb) #>> + '{{canonical_route_receipt,alert_category_zh_tw}}', + '' + ) <> '' + AND COALESCE( + COALESCE(m.source_envelope::jsonb, '{{}}'::jsonb) #>> + '{{canonical_route_receipt,destination_role}}', + '' + ) <> '' ) AS learning_writeback_ready_total, MAX(m.sent_at) AS latest_sent_at, MAX(m.queued_at) AS latest_queued_at @@ -1620,6 +1662,8 @@ async def list_ai_alert_card_delivery_readback( -> 'ai_automation_alert_card' AS alert_card, COALESCE(m.source_envelope::jsonb, '{{}}'::jsonb) -> 'source_refs' AS source_refs, + COALESCE(m.source_envelope::jsonb, '{{}}'::jsonb) + -> 'canonical_route_receipt' AS canonical_route_receipt, COALESCE(m.source_envelope::jsonb, '{{}}'::jsonb) -> 'agent99_dispatch_receipt' AS agent99_dispatch_receipt, ar.error_detail AS agent99_current_receipt, @@ -1927,6 +1971,48 @@ async def _load_ai_alert_card_delivery_readback_direct( '{{ai_automation_alert_card,event_type}}', '' ) <> 'backup_restore_escrow_signal' + AND m.send_status = 'sent' + AND COALESCE(m.provider_message_id::text, '') <> '' + AND COALESCE( + COALESCE(m.source_envelope::jsonb, '{{}}'::jsonb) #>> + '{{canonical_route_receipt,schema_version}}', + '' + ) = 'telegram_canonical_egress_receipt_v1' + AND COALESCE( + COALESCE(m.source_envelope::jsonb, '{{}}'::jsonb) #>> + '{{canonical_route_receipt,decision}}', + '' + ) = 'allowed' + AND COALESCE( + COALESCE(m.source_envelope::jsonb, '{{}}'::jsonb) #>> + '{{canonical_route_receipt,provider_send_performed}}', + '' + ) = 'true' + AND COALESCE( + COALESCE(m.source_envelope::jsonb, '{{}}'::jsonb) #>> + '{{canonical_route_receipt,durable_receipt_persisted}}', + '' + ) = 'true' + AND COALESCE( + COALESCE(m.source_envelope::jsonb, '{{}}'::jsonb) #>> + '{{canonical_route_receipt,route_id}}', + '' + ) <> '' + AND COALESCE( + COALESCE(m.source_envelope::jsonb, '{{}}'::jsonb) #>> + '{{canonical_route_receipt,receipt_backend}}', + '' + ) NOT IN ('', 'none') + AND COALESCE( + COALESCE(m.source_envelope::jsonb, '{{}}'::jsonb) #>> + '{{canonical_route_receipt,alert_category_zh_tw}}', + '' + ) <> '' + AND COALESCE( + COALESCE(m.source_envelope::jsonb, '{{}}'::jsonb) #>> + '{{canonical_route_receipt,destination_role}}', + '' + ) <> '' ) AS learning_writeback_ready_total, MAX(m.sent_at) AS latest_sent_at, MAX(m.queued_at) AS latest_queued_at @@ -2028,6 +2114,10 @@ async def _load_ai_alert_card_delivery_readback_direct( COALESCE(m.source_envelope::jsonb, '{{}}'::jsonb) -> 'source_refs' )::text AS source_refs_json, + ( + COALESCE(m.source_envelope::jsonb, '{{}}'::jsonb) + -> 'canonical_route_receipt' + )::text AS canonical_route_receipt_json, ( COALESCE(m.source_envelope::jsonb, '{{}}'::jsonb) -> 'agent99_dispatch_receipt' @@ -2137,6 +2227,9 @@ def _ai_alert_card_direct_row(row: Mapping[str, Any]) -> dict[str, Any]: payload = dict(row) payload["alert_card"] = _json_object(payload.pop("alert_card_json", None)) payload["source_refs"] = _json_object(payload.pop("source_refs_json", None)) + payload["canonical_route_receipt"] = _json_object( + payload.pop("canonical_route_receipt_json", None) + ) payload["agent99_dispatch_receipt"] = _json_object( payload.pop("agent99_dispatch_receipt_json", None) ) @@ -2558,6 +2651,11 @@ def _ai_alert_card_delivery_item(row: Mapping[str, Any]) -> dict[str, Any]: """Convert one AI alert-card outbound mirror row into delivery evidence.""" alert_card = _as_dict(row.get("alert_card")) source_refs = _as_dict(row.get("source_refs")) + canonical_route_receipt = _as_dict(row.get("canonical_route_receipt")) + durable_delivery_receipt = _ai_alert_card_durable_delivery_receipt( + row, + canonical_route_receipt=canonical_route_receipt, + ) frozen_agent99_dispatch_receipt = _as_dict( row.get("agent99_dispatch_receipt") ) @@ -2580,6 +2678,7 @@ def _ai_alert_card_delivery_item(row: Mapping[str, Any]) -> dict[str, Any]: message_id=row.get("message_id"), run_id=run_id, project_id=project_id, + durable_delivery_receipt=durable_delivery_receipt, ) backup_restore_automation: dict[str, Any] | None = None gates = ( @@ -2702,6 +2801,16 @@ def _ai_alert_card_delivery_item(row: Mapping[str, Any]) -> dict[str, Any]: "event_type": event_type, "lane": lane, "target": target, + "alert_category_zh_tw": durable_delivery_receipt[ + "alert_category_zh_tw" + ], + "route_id": durable_delivery_receipt["route_id"], + "destination_role": durable_delivery_receipt["destination_role"], + "durable_delivery_receipt_present": durable_delivery_receipt[ + "durable_delivery_receipt_present" + ], + "delivery_receipt_status": durable_delivery_receipt["status"], + "delivery_receipt": durable_delivery_receipt, "gates": gates, "declared_gates": declared_gates, "declared_runtime_write_gate_count": ( @@ -2754,6 +2863,72 @@ def _ai_alert_card_delivery_item(row: Mapping[str, Any]) -> dict[str, Any]: } +def _ai_alert_card_durable_delivery_receipt( + row: Mapping[str, Any], + *, + canonical_route_receipt: Mapping[str, Any], +) -> dict[str, Any]: + """Project only provider-acknowledged, DB-persisted Telegram delivery truth.""" + + alert_category_zh_tw = str( + canonical_route_receipt.get("alert_category_zh_tw") or "" + ).strip() + route_id = str(canonical_route_receipt.get("route_id") or "").strip() + destination_role = str( + canonical_route_receipt.get("destination_role") or "" + ).strip() + receipt_backend = str( + canonical_route_receipt.get("receipt_backend") or "" + ).strip() + checks = { + "outbound_row_persisted": bool(row.get("message_id")), + "send_status_sent": str(row.get("send_status") or "") == "sent", + "provider_message_id_present": bool( + str(row.get("provider_message_id") or "").strip() + ), + "canonical_receipt_schema_valid": ( + canonical_route_receipt.get("schema_version") + == "telegram_canonical_egress_receipt_v1" + ), + "canonical_route_allowed": ( + canonical_route_receipt.get("decision") == "allowed" + ), + "provider_send_performed": ( + canonical_route_receipt.get("provider_send_performed") is True + ), + "durable_receipt_persisted": ( + canonical_route_receipt.get("durable_receipt_persisted") is True + ), + "alert_category_present": bool(alert_category_zh_tw), + "route_id_present": bool(route_id), + "destination_role_present": bool(destination_role), + "durable_receipt_backend_declared": bool( + receipt_backend and receipt_backend != "none" + ), + } + missing = [name for name, passed in checks.items() if not passed] + durable = not missing + return { + "schema_version": "telegram_ai_alert_card_delivery_receipt_readback_v1", + "status": "ready" if durable else "durable_delivery_receipt_missing", + "durable_delivery_receipt_present": durable, + "alert_category_zh_tw": alert_category_zh_tw, + "route_id": route_id, + "destination_role": destination_role, + "receipt_backend": receipt_backend, + "destination_alias": str( + canonical_route_receipt.get("destination_alias") or "" + ).strip(), + "sender_bot_alias": str( + canonical_route_receipt.get("sender_bot_alias") or "" + ).strip(), + "provider_message_id": str(row.get("provider_message_id") or "").strip(), + "checks": checks, + "missing_required_evidence": missing, + "ready": durable, + } + + def _slug_ref(value: Any, fallback: str) -> str: slug = re.sub(r"[^a-z0-9_.:-]+", "-", str(value or "").strip().lower()) slug = slug.strip("-") @@ -2767,12 +2942,23 @@ def _ai_alert_card_learning_writeback_refs( message_id: Any, run_id: Any, project_id: str, + durable_delivery_receipt: Mapping[str, Any], ) -> dict[str, Any]: """Build deterministic KM / PlayBook / RAG / MCP refs for one alert receipt.""" event_type = str(alert_card.get("event_type") or "") lane = str(alert_card.get("lane") or "") target = str(alert_card.get("target") or "") delivery_required = bool(alert_card.get("delivery_receipt_readback_required")) + alert_category_zh_tw = str( + durable_delivery_receipt.get("alert_category_zh_tw") or "" + ).strip() + route_id = str(durable_delivery_receipt.get("route_id") or "").strip() + destination_role = str( + durable_delivery_receipt.get("destination_role") or "" + ).strip() + durable_delivery_present = bool( + durable_delivery_receipt.get("durable_delivery_receipt_present") is True + ) event_slug = _slug_ref(event_type, "unknown-event") lane_slug = _slug_ref(lane, "unknown-lane") target_slug = _slug_ref(target, "unknown-target") @@ -2784,6 +2970,10 @@ def _ai_alert_card_learning_writeback_refs( ("event_type", event_type), ("lane", lane), ("delivery_receipt_readback_required", delivery_required), + ("alert_category_zh_tw", alert_category_zh_tw), + ("route_id", route_id), + ("destination_role", destination_role), + ("durable_delivery_receipt", durable_delivery_present), ) if value in ("", False, None) ] @@ -2799,6 +2989,10 @@ def _ai_alert_card_learning_writeback_refs( "receipt_id": str(message_id or ""), "run_id": str(run_id or ""), "project_id": project_id, + "alert_category_zh_tw": alert_category_zh_tw, + "route_id": route_id, + "destination_role": destination_role, + "durable_delivery_receipt_present": durable_delivery_present, "km_entry_ref": f"km://{project_id}/telegram-ai-alert-card/{event_slug}", "playbook_candidate_ref": ( f"playbook://{project_id}/telegram-ai-alert-card/{lane_slug}/{target_slug}" @@ -2819,11 +3013,11 @@ def _ai_alert_card_learning_writeback_refs( "alert_ids": alert_ids if isinstance(alert_ids, list) else [], "missing_required_fields": missing, "production_write_performed": False, - "runtime_send_performed": False, + "runtime_send_performed": durable_delivery_present, "next_action": ( "none" if status == "ready" - else "ensure_ai_alert_cards_include_event_type_lane_and_delivery_receipt" + else "persist_provider_acknowledged_canonical_delivery_receipt_with_card_route_metadata" ), } diff --git a/apps/api/src/services/sre_k3s_controlled_automation_work_items.py b/apps/api/src/services/sre_k3s_controlled_automation_work_items.py index c58ec14c7..859696a1a 100644 --- a/apps/api/src/services/sre_k3s_controlled_automation_work_items.py +++ b/apps/api/src/services/sre_k3s_controlled_automation_work_items.py @@ -5,6 +5,7 @@ from __future__ import annotations import json from pathlib import Path from typing import Any +from uuid import UUID from src.services.ai_provider_policy import PRODUCTION_PROVIDER_ORDER from src.services.snapshot_paths import default_operations_dir @@ -136,6 +137,31 @@ def load_sre_k3s_controlled_automation_work_items( raise ValueError(f"{path}: paid calls require explicit cost controls") if provider.get("production_provider_route_switch_allowed") is not False: raise ValueError(f"{path}: production provider switch must remain gated") + durable_gate5 = _dict(provider, "durable_gate5_authorization", path) + if durable_gate5.get("status") == "consumed_terminal_failed": + run_id = str(durable_gate5.get("run_id") or "") + authorization_ref = str( + durable_gate5.get("authorization_ref") or "" + ) + try: + UUID(run_id) + except ValueError as exc: + raise ValueError(f"{path}: durable Gate5 run_id invalid") from exc + if run_id != authorization_ref: + raise ValueError(f"{path}: durable Gate5 identity mismatch") + if durable_gate5.get("approval_durable") is not True: + raise ValueError(f"{path}: durable Gate5 approval not acknowledged") + if durable_gate5.get("single_use_claim_consumed") is not True: + raise ValueError(f"{path}: durable Gate5 claim not consumed") + if durable_gate5.get("terminal_state") != "failed": + raise ValueError(f"{path}: partial canary must remain failed") + latest_canary = _dict(provider, "latest_canary_receipt", path) + if latest_canary.get("run_id") != run_id: + raise ValueError(f"{path}: canary receipt run mismatch") + if latest_canary.get("paired_rollback_verified") is not True: + raise ValueError(f"{path}: paid-provider rollback not verified") + if latest_canary.get("persistent_paid_routes_enabled") is not False: + raise ValueError(f"{path}: paid routes must remain disabled") bridge = _dict(payload, "agent99_host_operations_bridge", path) if bridge.get("arbitrary_command_allowed") is not False: diff --git a/apps/api/src/services/telegram_gateway.py b/apps/api/src/services/telegram_gateway.py index 8f2088cd3..cb4e62e3d 100644 --- a/apps/api/src/services/telegram_gateway.py +++ b/apps/api/src/services/telegram_gateway.py @@ -1117,6 +1117,10 @@ def _telegram_send_delivery_succeeded(result: object) -> bool: != "telegram_canonical_egress_receipt_v1" or route_receipt.get("decision") != "allowed" or route_receipt.get("provider_send_performed") is not True + or ( + route_receipt.get("durable_receipt_required") is True + and route_receipt.get("durable_receipt_persisted") is not True + ) ): return False @@ -5823,6 +5827,13 @@ class TelegramGateway: "schema_version": "telegram_canonical_egress_receipt_v1", "decision": "blocked_no_egress", "classifier": classifier, + "route_id": None, + "receipt_backend": "none", + "alert_category_zh_tw": "未分類告警", + "destination_role": "blocked_or_unresolved", + "durable_receipt_required": False, + "durable_receipt_persisted": False, + "durable_receipt_status": "not_applicable_no_egress", "product_id": str(context.get("product_id") or "unknown")[:80], "signal_family": str(context.get("signal_family") or "unknown")[:80], "severity": str(context.get("severity") or "unknown")[:16], @@ -5839,6 +5850,7 @@ class TelegramGateway: route_context: object, legacy_notification_type: object, actual_bot_alias: str, + method: str, ) -> tuple[str | None, dict[str, object]]: context: Mapping[str, object] | None = ( route_context if isinstance(route_context, Mapping) else None @@ -5900,6 +5912,23 @@ class TelegramGateway: "sender_bot_alias_not_authorized_for_route", route_context=normalized_context, ) + route_id = str(decision.get("route_id") or "").strip() + receipt_backend = str(decision.get("receipt_backend") or "").strip() + alert_category_zh_tw = str( + decision.get("alert_category_zh_tw") or "" + ).strip() + destination_role = str(decision.get("destination_role") or "").strip() + if ( + not route_id + or not receipt_backend + or receipt_backend == "none" + or not alert_category_zh_tw + or not destination_role + ): + return None, self._blocked_route_receipt( + "canonical_route_card_readback_metadata_missing", + route_context=normalized_context, + ) canonical_chat_id = str(settings.SRE_GROUP_CHAT_ID or "") if not canonical_chat_id: @@ -5918,12 +5947,23 @@ class TelegramGateway: "schema_version": "telegram_canonical_egress_receipt_v1", "decision": "allowed", "classifier": "canonical_shared_route_allowed", + "route_id": route_id, + "receipt_backend": receipt_backend, + "alert_category_zh_tw": alert_category_zh_tw, + "destination_role": destination_role, **normalized_context, "destination_alias": "awoooi_sre_war_room", "destination_binding": _telegram_destination_binding( canonical_chat_id ), "sender_bot_alias": actual_bot_alias, + "durable_receipt_required": method in {"sendMessage", "sendPhoto"}, + "durable_receipt_persisted": False, + "durable_receipt_status": ( + "pending_provider_and_mirror" + if method in {"sendMessage", "sendPhoto"} + else "not_required_for_interaction_method" + ), "provider_send_performed": False, } @@ -6014,6 +6054,13 @@ class TelegramGateway: target_chat_id ), "sender_bot_alias": actual_bot_alias, + "durable_receipt_required": False, + "durable_receipt_persisted": False, + "durable_receipt_status": ( + "best_effort_mirror_pending" + if method in {"sendMessage", "sendPhoto"} + else "not_required_for_interaction_method" + ), "inbound_context_verified": True, "provider_send_performed": False, } @@ -6035,6 +6082,7 @@ class TelegramGateway: }, legacy_notification_type=None, actual_bot_alias="tsenyang_bot", + method="sendMessage", ) return chat_id or "" @@ -6492,13 +6540,23 @@ class TelegramGateway: triggered_by_state = :triggered_by_state, source_envelope = jsonb_set( jsonb_set( - source_envelope, - '{callback_reply,status}', - to_jsonb('callback_reply_sent'::text), + jsonb_set( + jsonb_set( + source_envelope, + '{callback_reply,status}', + to_jsonb('callback_reply_sent'::text), + true + ), + '{notification_policy,disposition}', + to_jsonb('sent'::text), + true + ), + '{canonical_route_receipt,durable_receipt_persisted}', + 'true'::jsonb, true ), - '{notification_policy,disposition}', - to_jsonb('sent'::text), + '{canonical_route_receipt,durable_receipt_status}', + to_jsonb('persisted'::text), true ) WHERE project_id = :project_id @@ -6829,6 +6887,7 @@ class TelegramGateway: route_context=route_context, legacy_notification_type=legacy_notification_type, actual_bot_alias=actual_bot_alias, + method=method, ) ) elif interaction_context is not None: @@ -7067,19 +7126,24 @@ class TelegramGateway: provider_message_id = str(provider_message_id_value) if canonical_route_receipt is not None: canonical_route_receipt["provider_send_performed"] = True + if canonical_route_receipt.get("durable_receipt_required") is True: + canonical_route_receipt["durable_receipt_status"] = ( + "provider_sent_mirror_pending" + ) result["_awoooi_canonical_route_receipt"] = ( canonical_route_receipt ) if not destination_binding_verified: - if controlled_reservation is not None: - result[ - "_awooop_outbound_mirror_acknowledged" - ] = False + result["_awooop_outbound_mirror_acknowledged"] = False result["_awooop_delivery_status"] = ( "provider_destination_mismatch" if provider_destination_binding else "provider_destination_unverified" ) + if canonical_route_receipt is not None: + canonical_route_receipt["durable_receipt_status"] = ( + "provider_destination_unverified" + ) result["_awooop_provider_send_performed"] = True elif ( controlled_reservation is not None @@ -7096,31 +7160,124 @@ class TelegramGateway: ) ) result["_awooop_outbound_mirror_acknowledged"] = finalized + if canonical_route_receipt is not None: + canonical_route_receipt["durable_receipt_persisted"] = ( + finalized + ) + canonical_route_receipt["durable_receipt_status"] = ( + "persisted" + if finalized + else "provider_sent_durable_receipt_pending" + ) result["_awooop_delivery_status"] = ( - "sent" if finalized else "pending_unknown" + "sent" + if finalized + else "provider_sent_durable_receipt_pending" ) result["_awooop_provider_send_performed"] = True else: - result["_awooop_delivery_status"] = "sent" result["_awooop_provider_send_performed"] = True - source_envelope_extra = ( - _acknowledge_callback_reply_source_envelope( - source_envelope_extra, - delivery_result=result, - provider_message_id=provider_message_id, + durable_receipt_required = bool( + canonical_route_receipt is not None + and canonical_route_receipt.get( + "durable_receipt_required" ) + is True ) - await self._mirror_outbound_message( - method=method, - payload=payload, - provider_message_id=provider_message_id, - source_envelope_extra=source_envelope_extra, - ) + if durable_receipt_required: + prospective_receipt = dict(canonical_route_receipt) + prospective_receipt["durable_receipt_persisted"] = True + prospective_receipt["durable_receipt_status"] = "persisted" + prospective_result = dict(result) + prospective_result["_awooop_delivery_status"] = "sent" + prospective_result[ + "_awooop_provider_send_performed" + ] = True + prospective_result[ + "_awoooi_canonical_route_receipt" + ] = prospective_receipt + prospective_source_envelope_extra = ( + _acknowledge_callback_reply_source_envelope( + source_envelope_extra, + delivery_result=prospective_result, + provider_message_id=provider_message_id, + ) + ) + mirror_persisted = await self._mirror_outbound_message( + method=method, + payload=payload, + provider_message_id=provider_message_id, + source_envelope_extra=( + prospective_source_envelope_extra + ), + ) + canonical_route_receipt[ + "durable_receipt_persisted" + ] = mirror_persisted + canonical_route_receipt["durable_receipt_status"] = ( + "persisted" + if mirror_persisted + else "provider_sent_durable_receipt_pending" + ) + result[ + "_awooop_outbound_mirror_acknowledged" + ] = mirror_persisted + result["_awooop_delivery_status"] = ( + "sent" + if mirror_persisted + else "provider_sent_durable_receipt_pending" + ) + if mirror_persisted: + source_envelope_extra = ( + prospective_source_envelope_extra + ) + else: + result["_awooop_delivery_status"] = "sent" + result["_awooop_provider_send_performed"] = True + if method in {"sendMessage", "sendPhoto"}: + source_envelope_extra = ( + _acknowledge_callback_reply_source_envelope( + source_envelope_extra, + delivery_result=result, + provider_message_id=provider_message_id, + ) + ) + mirror_persisted = ( + await self._mirror_outbound_message( + method=method, + payload=payload, + provider_message_id=provider_message_id, + source_envelope_extra=( + source_envelope_extra + ), + ) + ) + result[ + "_awooop_outbound_mirror_acknowledged" + ] = mirror_persisted + if canonical_route_receipt is not None: + canonical_route_receipt[ + "durable_receipt_persisted" + ] = mirror_persisted + canonical_route_receipt[ + "durable_receipt_status" + ] = ( + "best_effort_persisted" + if mirror_persisted + else "best_effort_mirror_failed" + ) + else: + result[ + "_awooop_outbound_mirror_acknowledged" + ] = False else: - if controlled_reservation is not None: - result["_awooop_outbound_mirror_acknowledged"] = False + result["_awooop_outbound_mirror_acknowledged"] = False if canonical_route_receipt is not None: canonical_route_receipt["provider_send_performed"] = True + canonical_route_receipt["durable_receipt_persisted"] = False + canonical_route_receipt["durable_receipt_status"] = ( + "provider_message_id_missing" + ) result["_awooop_delivery_status"] = ( "provider_message_id_missing" ) @@ -7225,15 +7382,23 @@ class TelegramGateway: payload: dict, provider_message_id: str, source_envelope_extra: dict[str, object] | None = None, - ) -> None: - """將 legacy Telegram 出站訊息鏡像到 AwoooP，不改變實際發送行為。""" - if method != "sendMessage": - return + ) -> bool: + """Persist provider-acknowledged message/photo truth into AwoooP.""" + if method not in {"sendMessage", "sendPhoto"}: + return False chat_id = str(payload.get("chat_id") or "") text = str(payload.get("text") or payload.get("caption") or "") - if not chat_id or not text: - return + if not chat_id or not text or not provider_message_id: + return False + + mirror_extra = dict(source_envelope_extra or {}) + route_receipt = mirror_extra.get("canonical_route_receipt") + if isinstance(route_receipt, dict): + persisted_route_receipt = dict(route_receipt) + persisted_route_receipt["durable_receipt_persisted"] = True + persisted_route_receipt["durable_receipt_status"] = "persisted" + mirror_extra["canonical_route_receipt"] = persisted_route_receipt try: from src.core.context import get_current_project_id @@ -7243,7 +7408,7 @@ class TelegramGateway: project_id = get_current_project_id() or "awoooi" run_id = _legacy_outbound_run_id(chat_id, provider_message_id) async with get_db_context(project_id) as db: - await record_outbound_message( + message_id = await record_outbound_message( db, project_id=project_id, run_id=run_id, @@ -7252,26 +7417,39 @@ class TelegramGateway: message_type=_infer_outbound_message_type( text, payload, - source_envelope_extra, + mirror_extra, ), content=text, source_envelope=_merge_outbound_source_envelope_extra( _outbound_source_envelope(method, payload), - source_envelope_extra, + mirror_extra, ), provider_message_id=provider_message_id, send_status="sent", triggered_by_state="legacy_gateway", is_shadow=False, ) + if not message_id: + return False + logger.info( + "telegram_outbound_mirror_persisted", + method=method, + destination_binding=_telegram_destination_binding(chat_id), + provider_message_id=provider_message_id, + durable_receipt_persisted=True, + ) + return True except Exception as exc: logger.warning( "telegram_outbound_mirror_failed", method=method, - chat_id=chat_id, + destination_binding=_telegram_destination_binding(chat_id), provider_message_id=provider_message_id, - error=str(exc), + error_code="durable_mirror_persistence_failed", + error_type=type(exc).__name__, + durable_receipt_persisted=False, ) + return False async def _record_callback_reply_failure( self, @@ -13435,18 +13613,35 @@ class TelegramGateway: def _clean_ai_reply(text: str, max_chars: int = 600) -> str: """清理 AI 回覆：移除 Markdown 語法，截斷超長內容""" import re - # 移除 Markdown bold/italic (**text**, *text*, __text__, _text_) + + # 移除 Markdown bold/italic。單底線是 provider / receipt ID 的 + # 合法字元，不能用 ``_(.+?)_`` 清掉，否則結構化路由證據會失真。 text = re.sub(r'\*\*(.+?)\*\*', r'\1', text) text = re.sub(r'\*(.+?)\*', r'\1', text) text = re.sub(r'__(.+?)__', r'\1', text) - text = re.sub(r'_(.+?)_', r'\1', text) # 移除 Markdown header (#, ##, ###) text = re.sub(r'^#{1,3}\s+', '', text, flags=re.MULTILINE) # 移除 標籤（deepseek-r1） text = re.sub(r'.*?', '', text, flags=re.DOTALL).strip() - # 截斷 + # 截斷時保留結構化 provider receipt，否則真實的長 runtime + # readback 會把 provider/model/fallback/cost_receipt 尾端截掉。 if len(text) > max_chars: - text = text[:max_chars].rsplit('\n', 1)[0] + '…' + footer_match = re.search( + r'(?P