diff --git a/.env.example b/.env.example index 21f95a5..0ee9933 100644 --- a/.env.example +++ b/.env.example @@ -408,6 +408,8 @@ RAG_EMBED_MODEL=bge-m3:latest RAG_EMBED_DIM=1024 RAG_EMBED_NORMALIZE=true EMBED_CONSISTENCY_INCLUDE_111=false +INTERNAL_RAG_CANDIDATE_CANARY_THRESHOLD=0.70 +INTERNAL_RAG_CANDIDATE_CANARY_SCHEDULED_ENABLED=true PPT_VISION_ENABLED=true PPT_VISION_MODEL=minicpm-v:latest PPT_VISION_TIMEOUT=120 @@ -684,6 +686,7 @@ PIXELRAG_SOURCE_CONTRACT_REPLAY_RECEIPT_ROOT=/app/data/pixelrag/receipts/source_ PIXELRAG_MARKETPLACE_ADAPTER_DRY_RUN_RECEIPT_ROOT=/app/data/pixelrag/receipts/adapter_dry_run PIXELRAG_MARKETPLACE_ADAPTER_PREFLIGHT_RECEIPT_ROOT=/app/data/pixelrag/receipts/adapter_preflight PIXELRAG_MARKETPLACE_CANDIDATE_KNOWLEDGE_REPLAY_RECEIPT_ROOT=/app/data/pixelrag/receipts/knowledge_replay +INTERNAL_RAG_CANDIDATE_CANARY_RECEIPT_ROOT=/app/data/pixelrag/receipts/internal_rag_canary PIXELRAG_MARKETPLACE_EMBEDDING_SIGNATURE_GUARD_REPLAY_RECEIPT_ROOT=/app/data/pixelrag/receipts/embedding_guard PIXELRAG_MARKETPLACE_IDENTITY_MATCHER_REPLAY_RECEIPT_ROOT=/app/data/pixelrag/receipts/identity_matcher PIXELRAG_MARKETPLACE_PROMOTION_GATE_REPLAY_RECEIPT_ROOT=/app/data/pixelrag/receipts/promotion_gate diff --git a/config.py b/config.py index c9bca1e..c0c4511 100644 --- a/config.py +++ b/config.py @@ -414,7 +414,7 @@ YOUTUBE_API_KEY = os.getenv('YOUTUBE_API_KEY', '') # ========================================== # 系統版本與路徑 # ========================================== -SYSTEM_VERSION = "V10.810" +SYSTEM_VERSION = "V10.811" LOG_FILE_PATH = os.path.join(BASE_DIR, 'logs/system.log') public_url = PUBLIC_URL # 用於模板顯示 diff --git a/docs/AI_INTELLIGENCE_MODULE_SOT.md b/docs/AI_INTELLIGENCE_MODULE_SOT.md index da92bf0..a4efa91 100644 --- a/docs/AI_INTELLIGENCE_MODULE_SOT.md +++ b/docs/AI_INTELLIGENCE_MODULE_SOT.md @@ -1,8 +1,18 @@ # PChome 業績成長自動化作戰系統 — AI 競價情報模組 Single Source of Truth -> **最後更新**: 2026-07-15 (台北時間) -> **狀態**: 🟠 Partial。V10.801 已在 production 以 exact Gitea object `c091c466c203d7e26c8ed576f573e7e930df34d1` 運行。Yahoo run `7b3b140c076b43be8c79bd85448fbc49` 掃描 20 件、取得 17 筆來源候選，將 1 筆 source-disallowed unit-price 保留為 candidate-only，並讓 3 筆 total-price exact 通過 check-mode、bounded write 與 3/3 exact DB readback；來源已由 paused 原子切換為 `active + enabled + write_enabled`，rollback 不需要執行。固定母體 fingerprint `7b74504fcc1e1801c2ca2b42` 未漂移，`pchome_comparison_coverage_v3` 已由 `21/50 = 42%` 提升至 `24/50 = 48%`，業績覆蓋由 `NT$180,667 / NT$354,062 = 51.0%` 提升至 `NT$204,975 / NT$354,062 = 57.9%`，正式外部平台 runtime 由 `1/15` 提升至 `2/15`。完整 regression 為 `2,106 passed / 9 skipped / 0 failed`。SEC-P0-001 已完成 production closure，SEC-P0-002 維持 production hybrid canary-ready；full asset reconciliation、software supply chain、統一 controlled-apply envelope、restore drill、internal RAG canary 與 MCP/RAG runtime closure 尚未全部完成。任何局部 receipt 不得代表整體閉環完成。 -> **適用版本**: V10.801 production runtime；Gitea Action #1133 仍因 `ewoooc-host` runner 離線而未提供 CD 執行證據，正式 fallback deploy、runtime 與 visible receipts 已獨立保存 +> **最後更新**: 2026-07-17 (台北時間) +> **狀態**: 🟠 Partial。Production runtime 目前仍是 V10.810；V10.811 source 新增 daily scheduled internal RAG candidate canary 與四 AI Agent 產品整合 truth readback，尚未完成 production deploy/canary，因此不得標示完整整合。既有 Yahoo structured source 維持正式啟用，PChome 固定 TOP50 為 `25 ready + 1 candidate + 24 unmatched`、商品覆蓋 `50%`、業績覆蓋 `59.782%`、正式平台 `2/15`。七日 production AI telemetry 只有 Hermes `1` 次（`1` error）與 OpenClaw `13` 次（`4` errors），NemoTron、ElephantAlpha、MCP 與 RAG query 均為 `0`；`MCP_ROUTER_ENABLED=false`、`RAG_ENABLED=false`。程式存在、fallback 可匯入或 smoke 顯示 ok 都不能覆蓋這個 runtime truth。 +> **適用版本**: V10.810 production runtime；V10.811 source candidate pending Gitea/deploy/runtime verification + +--- + +## 零之負五、AI Agent 產品整合 truth 與 internal RAG canary（V10.811 source） + +- `/api/ai-automation/agent-product-integration` 與 `scripts/ops/report_ai_agent_product_integration.py` 分開輸出四 Agent source/scheduler wiring、七日 `ai_calls` 實際呼叫與錯誤率、MCP/RAG telemetry、action plan/outcome、AutoHeal incident retry，以及九階段 closure。只有四 Agent 全部有健康 runtime、MCP/RAG 已啟用且有 telemetry、internal RAG canary 已通過、受控執行/驗證/重試/學習都有實證時才可回 `fully_integrated`。 +- `/api/ai-automation/internal-rag-candidate-canary` 只有 GET，永遠是 no-model/no-DB-write readback；production execute 由 `momo-scheduler` 每日 04:45 自動跑一次，CLI 只保留給 Windows 99 controlled apply / verifier。每次最多一筆 candidate knowledge receipt，使用 Ollama-first BGE-M3 與 GCP-A/GCP-B consistency probe，再於 PostgreSQL `SET TRANSACTION READ ONLY` 中執行 pgvector exact/semantic similarity，最後 rollback。 +- canary 不 INSERT/UPDATE `ai_insights`、`competitor_prices`、`external_offers` 或任何正式價格表；artifact 與 scheduler receipt 共用 `trace_id/run_id/work_item_id`，明確輸出 `transaction_read_only`、similarity、embedding signature、GCP-A/GCP-B reachability、Telegram acknowledgement 與 zero-write/rollback terminal。簽名漂移、任一主要 GCP embedding host 不可達、pgvector probe 失敗或 semantic threshold 未達都必須 fail closed。 +- `bge-m3:latest` 仍是 immutable supply-chain blocker；即使 canary 通過，也只能回 `canary_passed_activation_blocked`，必須先鎖定可重現模型版本，再以 Windows 99 controlled apply 啟用 RAG shadow/canary。MCP runtime 仍受 localhost-only、read-only tool contract與 required secret presence preflight 約束，不可因 registry 已存在就宣稱上線。 +- `AI Agent product integration truth` 與 `Internal RAG candidate canary` 已接入 `/api/ai-automation/smoke`。舊的 NemoTron/ElephantAlpha class/method smoke 保留為 source guard，但不再能單獨代表產品整合完成。 --- diff --git a/docs/guides/ai_automation_mainline_work_items.md b/docs/guides/ai_automation_mainline_work_items.md index 1fe3711..fdbf581 100644 --- a/docs/guides/ai_automation_mainline_work_items.md +++ b/docs/guides/ai_automation_mainline_work_items.md @@ -1,6 +1,6 @@ # AI Automation Mainline Work Items -> Updated: 2026-07-16 16:08 Asia/Taipei +> Updated: 2026-07-17 01:25 Asia/Taipei > Governance: `global_product_governance_v2` + ADR-038 > Current P0: `GROWTH-P0-001 comparison coverage truth + autonomous refresh` @@ -23,7 +23,7 @@ | 5 | `SUPPLY-P0-001` | In progress | Gitea-only secure software supply chain | Gitea-native checkout, secret-safe `.dockerignore`, commit-bound source receipt and governance gate are active. Exit: exact dependency lock, internal SAST/SCA/secret scan, SBOM, image digest/provenance, vulnerability SLA and production digest readback. | | 6 | `GOV-P0-001` | In progress | Canonical full asset graph + runtime reconciliation | `governance/ewoooc_asset_inventory.json` seeds hosts, services, data, AI, routes, supply chain, observability and recovery. Exit: same-run probe receipt for every asset; drift auto-creates work items. | | 7 | `GOV-P0-002` | Not started | Unified controlled-apply envelope | Introduce one `trace_id/run_id/work_item_id` across sensor, identity, SOT diff, decision, risk, dry-run, execution, verifier, rollback/retry and learning acknowledgement. Start with EventRouter + AutoHeal. | -| 8 | `RAG-P0-001` | Not started | Internal RAG candidate canary | V10.770 stops at candidate preview. Exit: bounded pgvector candidate canary, signature/readback/feedback receipt, no price write and no premature `ai_insights` promotion. Next: `run_internal_rag_candidate_canary`. | +| 8 | `RAG-P0-001` | In progress (`scheduled_source_ready_runtime_unproven`) | Internal RAG candidate canary | V10.811 source adds `internal_rag_candidate_canary_service`, read-only authenticated API, CLI, smoke adapter and a default-enabled daily 04:45 scheduler run. The bounded executor processes at most one candidate, requires both GCP-A/GCP-B, runs pgvector similarity inside `SET TRANSACTION READ ONLY`, always rolls back, writes only a canary artifact/scheduler receipt and records Telegram acknowledgement; it cannot write `ai_insights` or price tables. Exit still requires production candidate/executed receipts, immutable model reference, controlled `RAG_ENABLED` shadow activation, query/hit telemetry and feedback readback. Source/test readiness is not runtime closure. | | 9 | `MCP-P0-001` | In progress (`federation_source_ready`) | MCP/RAG production runtime closure | V10.796 source adds a strict public aggregate receipt for canonical `ewoooc` and `momo-pro-system` identities without opening authenticated internal APIs or exposing endpoint/tool payload data. Exit still requires V10.796 production `/health`, two fresh AWOOOI durable receipts with fingerprint recompute, live MCP servers/router/RAG, approved caller/tool boundary and production query canary. Current source readiness must not be reported as runtime closure. | | 10 | `SEC-P0-004` | Not started | Security operations lifecycle and metrics | Add durable security incident state and publish MTTA, MTTR, recurrence, false positive, human intervention, verifier pass, rollback and freshness. Exit: detect-to-learn production receipt. | | 11 | `REL-P0-001` | In progress (`runtime_verified_cd_degraded`) | Formal deploy and visible proof discipline | Production V10.810 runs exact Gitea object `911393190ded015e384e438c26b68faf50ec260c`; dev merge `e56c613399affb5cf0134eea44b7f78aa8134943` carries the same source. Host 110 still exposes only paused `vibework-dedicated-runner`, with no matching EwoooC runner, so this release has no formal CD execution receipt. The controlled fallback archive SHA-256 is `cd0607d593fd16c8281b5f01839d20d7eba493f49d363cb197c30966c25be5ee`; archive, checksum manifest and rollback (`02e73678125ee0c8908a42be8fff479287180fa2c06f6efe7dce70c2ce7339a4`) are retained at `/home/ollama/momo-deploy-backups/analytics-ui-20260716T080302Z-9113931`. Internal/external `/health` is healthy at V10.810; only `momo-app` was recreated, while scheduler ID `9af8e31fd0eb90f74ce5894af1ae1fbc511362ec90df142fde6442cf737d1181`, Telegram bot ID `1d4bd922f6ad1252177ba40f16ea172dcd5eb1de1b4c999c00f4a3141bd06ca5` and `momo-db` immutable ID `cd092451cb5fd555d0ffff70642e109f3b742882c418beeab631793d1e9dc55d` did not change. Runner recovery remains a release-governance gap and is not replaced by fallback evidence. | @@ -38,6 +38,17 @@ These lanes are one ordered current P0, not optional side work. They must advanc | B. Verified same-item evidence | In progress | TOP50 fixed cohort: `25` verified, `1` candidate/source validation, `24` unmatched. Count coverage is `50%`; revenue-weighted coverage is `NT$211,667 / NT$354,062 = 59.782%`. V10.810 preserves fingerprint `7b74504fcc1e1801c2ca2b42`; run `34999f5314054f09909663b46fc42ee2` added one independently verified offer (`+2.0pp` count, `+1.89pp` revenue), while run `e1508ec3a49241d1965bc99d428b0325` correctly ended no-write. Their durable artifact SHA-256 values are `994beba6dfa70ef8c50031f130916ee155657dd77ac8a2f59b6530b491d45d1e` and `e58be7bb07599871a8f318ad63885bd868c786b0144b81ba0980ae80cf556541`. | Retry unresolved candidates only after fresh evidence arrives, preserve deterministic identity/unit/variant gates, and publish count/revenue deltas against the same fingerprint per run. | | C. Platform runtime coverage | In progress (`runtime_canary_activated`) | V10.810 is live. Yahoo is durably active and exact-offer canary readback remains valid; an already-active canary now returns `already_active_verified`, `state_changed=false` and `writes_database_count=0`, while latest no-write receipts no longer erase durable activation truth. Formal runtime is `2/15`; PixelRAG remains evidence-only. | Continue bounded refresh on schedule, monitor expiry/recurrence/rollback signals, then implement the next approved structured source contract for Shopee, Coupang, ETMall, Friday or Rakuten without treating blocked pages as product data. | +### AI Agent Product Integration Acceptance + +This is an acceptance surface inside the current growth P0; it does not reorder the P0 queue. V10.811 source replaces class/method-presence optimism with a seven-day production telemetry readback. + +| Layer | Current status | Exit evidence | +|---|---|---| +| Source and scheduler wiring | Source ready (`4/4`) | Hermes, NemoTron, OpenClaw and ElephantAlpha source markers plus scheduler ownership are machine-read and reported separately from runtime. | +| Agent runtime activity | Production baseline partial (`2/4`) | The pre-release live readback found Hermes `1` call / `1` error, OpenClaw `13` calls / `4` errors, and zero NemoTron/ElephantAlpha calls in seven days. Exit requires all four active and healthy in the bounded window, without treating a configured fallback as activity. | +| MCP/RAG dependency | Runtime disabled / telemetry empty | Production currently reports `MCP_ROUTER_ENABLED=false`, `RAG_ENABLED=false`, zero `mcp_calls` and zero `rag_query_log` activity. Exit requires enabled approved routes, live health, non-zero agent/product telemetry and the internal RAG candidate canary. | +| Controlled automation closure | Runtime partial | `/api/ai-automation/agent-product-integration`, CLI and smoke must report Detect -> Normalize -> Correlate -> Decide -> Check -> Controlled Apply -> Verify -> Retry/Rollback -> Learn/Writeback. Completion requires bounded execution, linked outcome/incident verification and durable learning evidence; aggregate source presence is insufficient. | + ### Analytics Period-Linkage Closure This bounded interruption is closed and control returns to `GROWTH-P0-001` without changing its lane order. @@ -85,7 +96,8 @@ These are reusable foundations, not proof that the full program is complete. | Completed | Marketplace identity matcher replay | Candidate identity only. | | Completed | PromotionGate replay | No production write. | | Completed | Embedding-signature guard replay | Signature readiness only. | -| Completed | Candidate knowledge replay | Internal RAG preview only; canary remains P0. | +| Completed | Candidate knowledge replay | Internal RAG preview only; no DB/model call. | +| Scheduled source ready | Internal RAG candidate canary | Daily bounded pgvector/Ollama canary exists in V10.811 source with same-run identity, rollback/no-write terminal and Telegram acknowledgement; production execution, model pin and RAG shadow activation remain P0. | | Completed | PixelRAG application portfolio | Commerce/RAG/UX/ops/marketing/governance inventory. | | Completed | Ollama-first VLM route readiness and replay worker | Evidence-bound artifact output; no direct price write. | | Completed | Platform probe worker | Shopee/Coupang barriers become structured fallback/backoff receipts. | diff --git a/routes/system_public_routes.py b/routes/system_public_routes.py index 8d80e2f..77e7db0 100644 --- a/routes/system_public_routes.py +++ b/routes/system_public_routes.py @@ -1164,6 +1164,49 @@ def ai_automation_pixelrag_marketplace_candidate_knowledge_replay_api(): )) +@system_public_bp.route( + '/api/ai-automation/internal-rag-candidate-canary', + methods=['GET'], +) +@login_required +def ai_automation_internal_rag_candidate_canary_api(): + """Read-only PixelRAG candidate canary readiness and latest runtime receipt.""" + from services.internal_rag_candidate_canary_service import ( + run_internal_rag_candidate_canary, + ) + + platforms = tuple( + str(item or '').strip() + for item in request.args.getlist('platform') + if str(item or '').strip() + ) + max_age_hours = request.args.get('max_age_hours', 168, type=int) + limit = request.args.get('limit', 1, type=int) + threshold = request.args.get('similarity_threshold', 0.70, type=float) + return jsonify(run_internal_rag_candidate_canary( + platform=platforms, + max_age_hours=max(1, min(max_age_hours or 168, 720)), + limit=max(1, min(limit or 1, 5)), + similarity_threshold=max(0.0, min(threshold or 0.70, 1.0)), + execute=False, + write_receipt=False, + )) + + +@system_public_bp.route('/api/ai-automation/agent-product-integration') +@login_required +def ai_automation_agent_product_integration_api(): + """Source/runtime/product closure truth for all four AI agents.""" + from services.ai_agent_product_integration_service import ( + build_ai_agent_product_integration_readback, + ) + + window_hours = request.args.get('window_hours', 168, type=int) + return jsonify(build_ai_agent_product_integration_readback( + window_hours=max(1, min(window_hours or 168, 24 * 31)), + )) + + @system_public_bp.route('/api/ai-automation/external-mcp-rag-integration') @login_required def ai_automation_external_mcp_rag_integration_api(): diff --git a/run_scheduler.py b/run_scheduler.py index 3bbefb9..979880e 100644 --- a/run_scheduler.py +++ b/run_scheduler.py @@ -9,7 +9,7 @@ run_scheduler.py — momo-scheduler 容器入口點 每 6 小時：pchome_growth_yahoo_backfill、quality_rescore、action_plan_hygiene 每 12 小時：dedup_batch 每 10 分鐘：ppt_auto_generation_catchup（補跑被長任務卡過的定期簡報） - 每 1 天 ：db_backup（03:00）、cleanup_agent_context（03:30）、backup_monitor（04:00）、daily_report（09:00）、roi_monthly_report gate（09:05）、ai_smoke_summary（09:10）、observability_daily_summary（09:30）、pchome_match_backfill（10:30）、pchome_growth_momo_backfill（10:45）、openclaw_meta_analysis（12:00, Phase 4 降頻）、ppt_auto_generation_daily（20:30）、ppt_vision_audit（22:00）、daily_token_report（23:55） + 每 1 天 ：db_backup（03:00）、cleanup_agent_context（03:30）、backup_monitor（04:00）、internal_rag_candidate_canary（04:45）、daily_report（09:00）、roi_monthly_report gate（09:05）、ai_smoke_summary（09:10）、observability_daily_summary（09:30）、pchome_match_backfill（10:30）、pchome_growth_momo_backfill（10:45）、openclaw_meta_analysis（12:00, Phase 4 降頻）、ppt_auto_generation_daily（20:30）、ppt_vision_audit（22:00）、daily_token_report（23:55） 每 1 週 ：weekly_strategy（週一 06:00）、ppt_auto_generation_weekly（週一 20:40） 每 1 月 ：monthly_report（每月1日 07:00）、ppt_auto_generation_monthly（每月1日 20:50） 每 1 季 ：ppt_auto_generation_quarterly（1/4/7/10 月 1 日 21:00） @@ -283,6 +283,121 @@ def _notify_scheduler_failure( logger.error("[%s] event_router notify failed: %s", task_name, router_error) +def run_internal_rag_candidate_canary_task() -> dict[str, object]: + """Run one scheduled PixelRAG-to-pgvector canary with durable acknowledgements.""" + if not _env_flag("INTERNAL_RAG_CANDIDATE_CANARY_SCHEDULED_ENABLED", True): + payload: dict[str, object] = { + "success": True, + "status": "skipped", + "terminal_status": "no_write_terminal", + "reason": "scheduled_canary_disabled_by_policy", + "next_machine_action": "enable_scheduled_canary_after_policy_review", + } + _save_stats("internal_rag_candidate_canary", payload) + return payload + + try: + from services.internal_rag_candidate_canary_service import ( + run_internal_rag_candidate_canary, + ) + from services.telegram_templates import send_telegram_with_result + + payload = run_internal_rag_candidate_canary( + limit=1, + execute=True, + write_receipt=True, + ) + summary = payload.get("summary") or {} + telegram = send_telegram_with_result( + "\n".join([ + "Internal RAG candidate canary", + f"status: {payload.get('status')}", + f"source: {summary.get('source_receipt_count', 0)}", + f"executed: {summary.get('executed_count', 0)}", + f"passed: {summary.get('canary_passed_count', 0)}", + f"next: {payload.get('next_machine_action')}", + f"run_id: {(payload.get('run_identity') or {}).get('run_id')}", + ]), + parse_mode=None, + ) + telegram_ack = "acknowledged" if telegram.get("ok") else "failed" + rag_ack = ( + "rag_canary_receipt_written" + if int(summary.get("executed_count") or 0) > 0 + else "no_write_terminal" + ) + payload["acknowledgements"] = { + "telegram": { + "status": telegram_ack, + "sent": int(telegram.get("sent") or 0), + "failed": int(telegram.get("failed") or 0), + }, + "km": "not_applicable_no_incident", + "rag": rag_ack, + "mcp": "not_applicable_no_mcp_call", + "playbook": "not_applicable_no_incident", + } + closure = payload.setdefault("closure_receipt", {}) + closure["telegram_acknowledgement"] = telegram_ack + closure["learning_write_acknowledgement"] = rag_ack + if payload.get("status") in {"canary_failed", "blocked"} or telegram_ack == "failed": + payload["terminal_status"] = "partial" + elif payload.get("status") == "warning": + payload["terminal_status"] = "no_write_terminal" + elif payload.get("status") == "canary_passed_activation_blocked": + payload["terminal_status"] = "verified_with_activation_blocker" + else: + payload["terminal_status"] = "verified" + + _save_stats("internal_rag_candidate_canary", payload) + logger.info( + "[InternalRAGCanary] status=%s terminal=%s executed=%s passed=%s telegram=%s", + payload.get("status"), + payload.get("terminal_status"), + summary.get("executed_count", 0), + summary.get("canary_passed_count", 0), + telegram_ack, + ) + if payload.get("status") == "canary_failed": + _notify_scheduler_failure( + "run_internal_rag_candidate_canary_task", + RuntimeError("internal RAG candidate canary failed"), + source="Scheduler.InternalRAGCanary", + event_type="internal_rag_candidate_canary_failure", + title="Internal RAG candidate canary failed", + dedup_ttl_sec=86400, + ) + elif telegram_ack == "failed": + _notify_scheduler_failure( + "run_internal_rag_candidate_canary_task", + RuntimeError("internal RAG canary Telegram acknowledgement failed"), + source="Scheduler.InternalRAGCanary", + event_type="internal_rag_candidate_canary_ack_failure", + title="Internal RAG canary acknowledgement failed", + dedup_ttl_sec=86400, + ) + return payload + except Exception as error: + logger.error("[InternalRAGCanary] task failed: %s", error, exc_info=True) + _notify_scheduler_failure( + "run_internal_rag_candidate_canary_task", + error, + source="Scheduler.InternalRAGCanary", + event_type="internal_rag_candidate_canary_task_failure", + title="Internal RAG candidate canary task failed", + dedup_ttl_sec=86400, + ) + payload = { + "success": False, + "status": "failed", + "terminal_status": "partial", + "error": f"{type(error).__name__}: {str(error)[:300]}", + "next_machine_action": "inspect_canary_receipt_and_retry_bounded_run", + } + _save_stats("internal_rag_candidate_canary", payload) + return payload + + def _register_schedules(): schedule.every(30).minutes.do(run_revenue_automation_cycle_background_task) logger.info("📅 每 30 分鐘：revenue_automation_cycle（業績匯入 → 比價回填，非阻塞）") @@ -464,6 +579,9 @@ def _register_schedules(): schedule.every().day.at("04:15").do(run_codebase_modularization_performance_task) logger.info("📅 每日 04:15：codebase_modularization_performance inventory") + schedule.every().day.at("04:45").do(run_internal_rag_candidate_canary_task) + logger.info("📅 每日 04:45：internal_rag_candidate_canary（bounded read-only probe）") + schedule.every().monday.at("06:00").do(run_weekly_strategy_task) logger.info("📅 每週一 06:00：weekly_strategy") diff --git a/scripts/ops/report_ai_agent_product_integration.py b/scripts/ops/report_ai_agent_product_integration.py new file mode 100644 index 0000000..90fe560 --- /dev/null +++ b/scripts/ops/report_ai_agent_product_integration.py @@ -0,0 +1,39 @@ +#!/usr/bin/env python3 +"""Report source/runtime/product integration truth for all four AI agents.""" + +from __future__ import annotations + +import argparse +import contextlib +import json +import sys +from pathlib import Path + + +ROOT = Path(__file__).resolve().parents[2] +if str(ROOT) not in sys.path: + sys.path.insert(0, str(ROOT)) + +from services.ai_agent_product_integration_service import ( # noqa: E402 + build_ai_agent_product_integration_readback, +) + + +def main() -> int: + parser = argparse.ArgumentParser( + description="輸出四個 AI Agent 的 source、runtime 與產品閉環整合實證。" + ) + parser.add_argument("--window-hours", type=int, default=168) + args = parser.parse_args() + # Some legacy DB modules still print startup diagnostics. Keep stdout as a + # strict JSON channel for automation and forward diagnostics to stderr. + with contextlib.redirect_stdout(sys.stderr): + payload = build_ai_agent_product_integration_readback( + window_hours=args.window_hours, + ) + print(json.dumps(payload, ensure_ascii=False, indent=2, sort_keys=True)) + return 0 if payload.get("success") else 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/ops/run_internal_rag_candidate_canary.py b/scripts/ops/run_internal_rag_candidate_canary.py new file mode 100644 index 0000000..c3d8ab3 --- /dev/null +++ b/scripts/ops/run_internal_rag_candidate_canary.py @@ -0,0 +1,51 @@ +#!/usr/bin/env python3 +"""Dry-run or execute the bounded internal RAG candidate canary.""" + +from __future__ import annotations + +import argparse +import contextlib +import json +import sys +from pathlib import Path + + +ROOT = Path(__file__).resolve().parents[2] +if str(ROOT) not in sys.path: + sys.path.insert(0, str(ROOT)) + +from services.internal_rag_candidate_canary_service import ( # noqa: E402 + run_internal_rag_candidate_canary, +) + + +def main() -> int: + parser = argparse.ArgumentParser( + description="執行 PixelRAG candidate 到內部 pgvector RAG 的受控 canary。" + ) + parser.add_argument("--candidate-knowledge-receipt-root") + parser.add_argument("--output-root") + parser.add_argument("--platform", action="append", dest="platforms") + parser.add_argument("--max-age-hours", type=int, default=168) + parser.add_argument("--limit", type=int, default=1) + parser.add_argument("--similarity-threshold", type=float, default=0.70) + parser.add_argument("--execute", action="store_true") + parser.add_argument("--write-receipt", action="store_true") + args = parser.parse_args() + with contextlib.redirect_stdout(sys.stderr): + payload = run_internal_rag_candidate_canary( + candidate_knowledge_receipt_root=args.candidate_knowledge_receipt_root, + output_root=args.output_root, + platform=tuple(args.platforms or ()), + max_age_hours=args.max_age_hours, + limit=args.limit, + similarity_threshold=args.similarity_threshold, + execute=args.execute, + write_receipt=bool(args.execute and args.write_receipt), + ) + print(json.dumps(payload, ensure_ascii=False, indent=2, sort_keys=True)) + return 0 if payload.get("success") else 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/services/ai_agent_product_integration_service.py b/services/ai_agent_product_integration_service.py new file mode 100644 index 0000000..d052898 --- /dev/null +++ b/services/ai_agent_product_integration_service.py @@ -0,0 +1,531 @@ +"""Truthful product-integration readback for the four AI agents. + +Source wiring and runtime telemetry are reported separately so a class import or +configured fallback cannot be mistaken for a working production automation loop. +""" + +from __future__ import annotations + +from datetime import datetime, timedelta, timezone +from pathlib import Path +from typing import Any, Mapping + +from sqlalchemy import text + +from services.external_mcp_rag_integration_service import ( + build_external_mcp_rag_integration_readback, +) +from services.internal_rag_candidate_canary_service import ( + run_internal_rag_candidate_canary, +) + + +POLICY = "runtime_truth_ai_agent_product_integration_v1" +ROOT = Path(__file__).resolve().parent.parent + +AGENT_SPECS: dict[str, dict[str, Any]] = { + "hermes": { + "label": "Hermes", + "role": "競價情報分析", + "source_path": "services/hermes_analyst_service.py", + "source_markers": ["class HermesAnalystService", "def analyze"], + "scheduler_path": "scheduler.py", + "scheduler_markers": ["HermesAnalystService", "run_icaim"], + }, + "nemotron": { + "label": "NemoTron", + "role": "受控行動派發", + "source_path": "services/nemoton_dispatcher_service.py", + "source_markers": ["class Nemotron", "dispatch"], + "scheduler_path": "scheduler.py", + "scheduler_markers": ["NemoTron", "dispatch"], + }, + "openclaw": { + "label": "OpenClaw", + "role": "策略、報表與學習寫回", + "source_path": "services/openclaw_strategist_service.py", + "source_markers": ["daily_report", "ai_insights"], + "scheduler_path": "scheduler.py", + "scheduler_markers": ["OpenClaw", "daily"], + }, + "elephant_alpha": { + "label": "ElephantAlpha", + "role": "跨 Agent 編排與自癒", + "source_path": "services/elephant_alpha_autonomous_engine.py", + "source_markers": ["class ElephantAlphaAutonomousEngine", "action_plans"], + "scheduler_path": "run_scheduler.py", + "scheduler_markers": ["ElephantAlpha", "Autonomous engine"], + }, +} + + +def _agent_key(value: Any) -> str | None: + caller = str(value or "").strip().lower().replace("-", "_") + if "hermes" in caller: + return "hermes" + if "nemotron" in caller or "nemoton" in caller or caller.startswith("nim_"): + return "nemotron" + if "openclaw" in caller or "open_claw" in caller: + return "openclaw" + if "elephant" in caller or caller.startswith("ea_"): + return "elephant_alpha" + return None + + +def _row_dict(row: Any) -> dict[str, Any]: + mapping = getattr(row, "_mapping", None) + return dict(mapping) if mapping is not None else dict(row or {}) + + +def _query_rows(sql: str, params: Mapping[str, Any]) -> tuple[list[dict[str, Any]], str | None]: + session = None + try: + from database.manager import get_session + + session = get_session() + rows = session.execute(text(sql), dict(params)).fetchall() + return [_row_dict(row) for row in rows], None + except (Exception, SystemExit) as exc: + return [], f"{type(exc).__name__}: {str(exc)[:240]}" + finally: + if session is not None: + session.close() + + +def _blank_agent_metrics() -> dict[str, dict[str, Any]]: + return { + key: { + "calls": 0, + "errors": 0, + "rag_hits": 0, + "fallbacks": 0, + "mcp_calls": 0, + "rag_queries": 0, + "rag_query_hits": 0, + "ai_insights": 0, + "action_plans": 0, + "executed_action_plans": 0, + "last_called": None, + } + for key in AGENT_SPECS + } + + +def _collect_runtime_telemetry(since_at: datetime) -> dict[str, Any]: + agents = _blank_agent_metrics() + errors: list[str] = [] + ai_rows, error = _query_rows( + """ + SELECT caller, + COUNT(*) AS calls, + COUNT(*) FILTER ( + WHERE lower(COALESCE(status, '')) IN + ('error', 'failed', 'failure', 'timeout', 'cancelled') + OR error IS NOT NULL + ) AS errors, + COUNT(*) FILTER (WHERE COALESCE(rag_hit, false)) AS rag_hits, + COUNT(*) FILTER (WHERE fallback_to IS NOT NULL) AS fallbacks, + MAX(called_at) AS last_called + FROM ai_calls + WHERE called_at >= :since_at + GROUP BY caller + """, + {"since_at": since_at}, + ) + if error: + errors.append(f"ai_calls:{error}") + unmatched_callers: list[dict[str, Any]] = [] + for row in ai_rows: + key = _agent_key(row.get("caller")) + if key is None: + unmatched_callers.append( + {"caller": row.get("caller"), "calls": int(row.get("calls") or 0)} + ) + continue + metrics = agents[key] + metrics["calls"] += int(row.get("calls") or 0) + metrics["errors"] += int(row.get("errors") or 0) + metrics["rag_hits"] += int(row.get("rag_hits") or 0) + metrics["fallbacks"] += int(row.get("fallbacks") or 0) + last_called = row.get("last_called") + if last_called and ( + not metrics["last_called"] or str(last_called) > str(metrics["last_called"]) + ): + metrics["last_called"] = str(last_called) + + mcp_rows, error = _query_rows( + """ + SELECT caller, COUNT(*) AS calls + FROM mcp_calls + WHERE called_at >= :since_at + GROUP BY caller + """, + {"since_at": since_at}, + ) + if error: + errors.append(f"mcp_calls:{error}") + all_mcp_calls = sum(int(row.get("calls") or 0) for row in mcp_rows) + for row in mcp_rows: + key = _agent_key(row.get("caller")) + if key: + agents[key]["mcp_calls"] += int(row.get("calls") or 0) + + rag_rows, error = _query_rows( + """ + SELECT caller, + COUNT(*) AS queries, + COALESCE(SUM(hit_count), 0) AS hits + FROM rag_query_log + WHERE queried_at >= :since_at + GROUP BY caller + """, + {"since_at": since_at}, + ) + if error: + errors.append(f"rag_query_log:{error}") + all_rag_queries = sum(int(row.get("queries") or 0) for row in rag_rows) + all_rag_hits = sum(int(row.get("hits") or 0) for row in rag_rows) + for row in rag_rows: + key = _agent_key(row.get("caller")) + if key: + agents[key]["rag_queries"] += int(row.get("queries") or 0) + agents[key]["rag_query_hits"] += int(row.get("hits") or 0) + + insight_rows, error = _query_rows( + """ + SELECT created_by, COUNT(*) AS count + FROM ai_insights + WHERE created_at >= :since_at + GROUP BY created_by + """, + {"since_at": since_at.replace(tzinfo=None)}, + ) + if error: + errors.append(f"ai_insights:{error}") + insight_total = 0 + for row in insight_rows: + count = int(row.get("count") or 0) + insight_total += count + key = _agent_key(row.get("created_by")) + if key: + agents[key]["ai_insights"] += count + + plan_rows, error = _query_rows( + """ + SELECT created_by, + COUNT(*) AS count, + COUNT(*) FILTER ( + WHERE status = 'executed' OR executed_at IS NOT NULL + ) AS executed + FROM action_plans + WHERE created_at >= :since_at + GROUP BY created_by + """, + {"since_at": since_at.replace(tzinfo=None)}, + ) + if error: + errors.append(f"action_plans:{error}") + action_plan_total = 0 + executed_action_plan_total = 0 + for row in plan_rows: + count = int(row.get("count") or 0) + executed = int(row.get("executed") or 0) + action_plan_total += count + executed_action_plan_total += executed + key = _agent_key(row.get("created_by")) + if key: + agents[key]["action_plans"] += count + agents[key]["executed_action_plans"] += executed + + outcome_rows, error = _query_rows( + "SELECT COUNT(*) AS count FROM action_outcomes WHERE created_at >= :since_at", + {"since_at": since_at.replace(tzinfo=None)}, + ) + if error: + errors.append(f"action_outcomes:{error}") + action_outcome_total = int(outcome_rows[0].get("count") or 0) if outcome_rows else 0 + + heal_rows, error = _query_rows( + """ + SELECT COUNT(*) AS count, + COUNT(*) FILTER (WHERE result = 'success') AS success + FROM heal_logs + WHERE created_at >= :since_at + """, + {"since_at": since_at.replace(tzinfo=None)}, + ) + if error: + errors.append(f"heal_logs:{error}") + heal_total = int(heal_rows[0].get("count") or 0) if heal_rows else 0 + heal_success = int(heal_rows[0].get("success") or 0) if heal_rows else 0 + + retry_rows, error = _query_rows( + """ + SELECT COUNT(DISTINCT incidents.id) AS count + FROM incidents + JOIN heal_logs ON heal_logs.incident_id = incidents.id + WHERE incidents.created_at >= :since_at + AND incidents.retry_count > 0 + AND incidents.status = 'closed' + AND heal_logs.result = 'success' + """, + {"since_at": since_at.replace(tzinfo=None)}, + ) + if error: + errors.append(f"verified_retry_or_rollback_incidents:{error}") + verified_retry_or_rollback_incidents = ( + int(retry_rows[0].get("count") or 0) if retry_rows else 0 + ) + + queue_rows, error = _query_rows( + "SELECT status, COUNT(*) AS count FROM embedding_retry_queue GROUP BY status", + {}, + ) + if error: + errors.append(f"embedding_retry_queue:{error}") + queue_counts = { + str(row.get("status") or "unknown"): int(row.get("count") or 0) + for row in queue_rows + } + + return { + "agents": agents, + "unmatched_ai_callers": unmatched_callers, + "totals": { + "ai_calls": sum(item["calls"] for item in agents.values()), + "ai_errors": sum(item["errors"] for item in agents.values()), + "mcp_calls": all_mcp_calls, + "agent_mapped_mcp_calls": sum( + item["mcp_calls"] for item in agents.values() + ), + "rag_queries": all_rag_queries, + "agent_mapped_rag_queries": sum( + item["rag_queries"] for item in agents.values() + ), + "rag_query_hits": all_rag_hits, + "ai_insights": insight_total, + "action_plans": action_plan_total, + "executed_action_plans": executed_action_plan_total, + "action_outcomes": action_outcome_total, + "heal_logs": heal_total, + "heal_success": heal_success, + "verified_retry_or_rollback_incidents": ( + verified_retry_or_rollback_incidents + ), + }, + "embedding_retry_queue": queue_counts, + "read_errors": errors, + } + + +def _source_surface(spec: Mapping[str, Any]) -> dict[str, Any]: + source_path = ROOT / str(spec["source_path"]) + scheduler_path = ROOT / str(spec["scheduler_path"]) + try: + source_text = source_path.read_text(encoding="utf-8") + except OSError: + source_text = "" + try: + scheduler_text = scheduler_path.read_text(encoding="utf-8") + except OSError: + scheduler_text = "" + source_markers = { + marker: marker in source_text for marker in list(spec.get("source_markers") or []) + } + scheduler_markers = { + marker: marker in scheduler_text + for marker in list(spec.get("scheduler_markers") or []) + } + return { + "source_path": str(spec["source_path"]), + "source_exists": source_path.exists(), + "source_markers": source_markers, + "source_wired": source_path.exists() and all(source_markers.values()), + "scheduler_path": str(spec["scheduler_path"]), + "scheduler_exists": scheduler_path.exists(), + "scheduler_markers": scheduler_markers, + "scheduler_wired": scheduler_path.exists() and all(scheduler_markers.values()), + } + + +def _agent_readback( + key: str, + metrics: Mapping[str, Any], +) -> dict[str, Any]: + spec = AGENT_SPECS[key] + source = _source_surface(spec) + calls = int(metrics.get("calls") or 0) + errors = int(metrics.get("errors") or 0) + error_rate = round(errors / calls, 4) if calls else None + runtime_active = calls > 0 + runtime_healthy = runtime_active and errors / calls <= 0.2 + source_complete = source["source_wired"] and source["scheduler_wired"] + integration_complete = source_complete and runtime_healthy + if not source_complete: + status = "source_incomplete" + elif not runtime_active: + status = "runtime_inactive" + elif not runtime_healthy: + status = "runtime_degraded" + else: + status = "runtime_active" + return { + "id": key, + "label": spec["label"], + "role": spec["role"], + "status": status, + "source": source, + "runtime": { + **dict(metrics), + "active_in_window": runtime_active, + "error_rate": error_rate, + "healthy_in_window": runtime_healthy, + }, + "integration_complete": integration_complete, + "next_machine_action": ( + "repair_agent_source_or_scheduler_wiring" + if not source_complete + else ( + "execute_bounded_agent_runtime_canary" + if not runtime_active + else ( + "repair_agent_error_path_and_replay" + if not runtime_healthy + else "continue_runtime_closure_verification" + ) + ) + ), + } + + +def build_ai_agent_product_integration_readback( + *, + window_hours: int = 168, +) -> dict[str, Any]: + """Return source, runtime and product-closure truth for all four agents.""" + window = max(1, min(int(window_hours or 168), 24 * 31)) + now = datetime.now(timezone.utc) + since_at = now - timedelta(hours=window) + telemetry = _collect_runtime_telemetry(since_at) + agents = [ + _agent_readback(key, telemetry["agents"].get(key) or {}) + for key in AGENT_SPECS + ] + external = build_external_mcp_rag_integration_readback() + mcp_runtime = dict((external.get("runtime") or {}).get("mcp") or {}) + rag_runtime = dict((external.get("runtime") or {}).get("rag") or {}) + rag_canary = run_internal_rag_candidate_canary(execute=False) + latest_canary = dict(rag_canary.get("latest_execution") or {}) + totals = telemetry["totals"] + active_agents = sum( + 1 for item in agents if item["runtime"]["active_in_window"] + ) + healthy_agents = sum( + 1 for item in agents if item["runtime"]["healthy_in_window"] + ) + source_agents = sum( + 1 + for item in agents + if item["source"]["source_wired"] and item["source"]["scheduler_wired"] + ) + stages = [ + {"stage": "Detect", "passed": totals["ai_calls"] > 0, "evidence": totals["ai_calls"]}, + {"stage": "Normalize", "passed": source_agents == len(agents), "evidence": source_agents}, + {"stage": "Correlate", "passed": active_agents == len(agents), "evidence": active_agents}, + {"stage": "Decide", "passed": totals["action_plans"] > 0, "evidence": totals["action_plans"]}, + {"stage": "Check", "passed": healthy_agents == len(agents), "evidence": healthy_agents}, + {"stage": "Controlled Apply", "passed": totals["executed_action_plans"] > 0, "evidence": totals["executed_action_plans"]}, + {"stage": "Verify", "passed": totals["action_outcomes"] > 0 or totals["heal_success"] > 0, "evidence": totals["action_outcomes"] + totals["heal_success"]}, + { + "stage": "Retry/Rollback", + "passed": totals["verified_retry_or_rollback_incidents"] > 0, + "evidence": totals["verified_retry_or_rollback_incidents"], + }, + {"stage": "Learn/Writeback", "passed": totals["ai_insights"] > 0 and totals["rag_query_hits"] > 0, "evidence": totals["rag_query_hits"]}, + ] + stage_passed = sum(1 for stage in stages if stage["passed"]) + blockers: list[str] = [] + if source_agents != len(agents): + blockers.append("agent_source_or_scheduler_wiring_incomplete") + if active_agents != len(agents): + blockers.append("not_all_agents_active_in_runtime_window") + if healthy_agents != len(agents): + blockers.append("not_all_agents_healthy_in_runtime_window") + if mcp_runtime.get("enabled") is not True: + blockers.append("mcp_router_runtime_disabled") + if rag_runtime.get("enabled") is not True: + blockers.append("rag_runtime_disabled") + if totals["mcp_calls"] == 0: + blockers.append("mcp_runtime_telemetry_empty") + if totals["rag_queries"] == 0: + blockers.append("rag_runtime_telemetry_empty") + if latest_canary.get("canary_passed") is not True: + blockers.append("internal_rag_candidate_canary_not_proven") + if stage_passed != len(stages): + blockers.append("agent_controlled_apply_loop_not_closed") + if telemetry["read_errors"]: + blockers.append("runtime_telemetry_read_degraded") + full_integration = not blockers + status = "fully_integrated" if full_integration else "partially_integrated" + + return { + "success": True, + "policy": POLICY, + "generated_at": now.isoformat(), + "status": status, + "answer_to_owner": ( + f"四個 AI Agent source/排程 wiring={source_agents}/4；正式環境尚未完整整合：" + f"最近 {window} 小時只有 {active_agents}/4 個 Agent 有實際呼叫，" + f"完整閉環 {stage_passed}/{len(stages)} 階段，MCP/RAG runtime 與 canary 必須以實證補齊。" + if not full_integration + else "四個 AI Agent 已有 source、runtime、MCP/RAG 與受控執行閉環實證。" + ), + "window": { + "hours": window, + "since_at": since_at.isoformat(), + }, + "completion": { + "agent_count": len(agents), + "source_wired_agents": source_agents, + "runtime_active_agents": active_agents, + "runtime_healthy_agents": healthy_agents, + "source_percent": round(source_agents / len(agents) * 100, 1), + "runtime_active_percent": round(active_agents / len(agents) * 100, 1), + "closure_stage_passed": stage_passed, + "closure_stage_total": len(stages), + "closure_percent": round(stage_passed / len(stages) * 100, 1), + "full_product_integration": full_integration, + }, + "agents": agents, + "closure_stages": stages, + "runtime_dependencies": { + "mcp": { + "enabled": mcp_runtime.get("enabled") is True, + "telemetry_calls": totals["mcp_calls"], + }, + "rag": { + "enabled": rag_runtime.get("enabled") is True, + "telemetry_queries": totals["rag_queries"], + "telemetry_hits": totals["rag_query_hits"], + "latest_candidate_canary": latest_canary, + }, + "pixelrag": (external.get("runtime") or {}).get("pixelrag") or {}, + }, + "telemetry": telemetry, + "blockers": blockers, + "controlled_apply": { + "database_read": True, + "database_write": False, + "network_call": False, + "model_call": False, + "secret_read": False, + }, + "next_machine_action": ( + "execute_internal_rag_candidate_canary_then_activate_shadow_runtime" + if not full_integration + else "continue_scheduled_agent_product_integration_verification" + ), + } + + +__all__ = ["POLICY", "build_ai_agent_product_integration_readback"] diff --git a/services/ai_automation_smoke_service.py b/services/ai_automation_smoke_service.py index 4f80e8c..11e8bb5 100644 --- a/services/ai_automation_smoke_service.py +++ b/services/ai_automation_smoke_service.py @@ -13313,6 +13313,105 @@ def _external_mcp_rag_integration_check() -> Dict[str, Any]: ) +def _ai_agent_product_integration_check() -> Dict[str, Any]: + """Runtime-truth sentinel for all four product AI agents.""" + try: + from services.ai_agent_product_integration_service import ( + build_ai_agent_product_integration_readback, + ) + + readback = build_ai_agent_product_integration_readback(window_hours=168) + completion = readback.get("completion") or {} + full = completion.get("full_product_integration") is True + active = int(completion.get("runtime_active_agents") or 0) + healthy = int(completion.get("runtime_healthy_agents") or 0) + stage_passed = int(completion.get("closure_stage_passed") or 0) + stage_total = int(completion.get("closure_stage_total") or 0) + blockers = list(readback.get("blockers") or []) + return _check( + "AI Agent product integration truth", + "ok" if full else "warning", + ( + f"AI Agents runtime active={active}/4, healthy={healthy}/4, " + f"closure={stage_passed}/{stage_total}, blockers={len(blockers)}" + ), + { + "policy": readback.get("policy"), + "integration_status": readback.get("status"), + "source_wired_agents": int( + completion.get("source_wired_agents") or 0 + ), + "runtime_active_agents": active, + "runtime_healthy_agents": healthy, + "closure_stage_passed": stage_passed, + "closure_stage_total": stage_total, + "full_product_integration": full, + "blockers": blockers, + "next_machine_action": readback.get("next_machine_action"), + "writes_database": False, + "writes_database_count": 0, + "primary_human_gate_count": 0, + }, + ) + except Exception as exc: + return _check( + "AI Agent product integration truth", + "critical", + f"AI Agent product integration readback failed: {exc}", + { + "writes_database": False, + "writes_database_count": 0, + "primary_human_gate_count": 0, + }, + ) + + +def _internal_rag_candidate_canary_check() -> Dict[str, Any]: + """No-model-call readback for PixelRAG to internal pgvector canary proof.""" + try: + from services.internal_rag_candidate_canary_service import ( + run_internal_rag_candidate_canary, + ) + + readback = run_internal_rag_candidate_canary(execute=False) + summary = readback.get("summary") or {} + latest = readback.get("latest_execution") or {} + proven = latest.get("canary_passed") is True + ready = int(summary.get("ready_count") or 0) + blockers = list(readback.get("activation_blockers") or []) + return _check( + "Internal RAG candidate canary", + "ok" if proven and not blockers else "warning", + ( + f"PixelRAG candidate ready={ready}, historical canary={proven}, " + f"activation blockers={len(blockers)}" + ), + { + "policy": readback.get("policy"), + "status": readback.get("status"), + "ready_count": ready, + "historical_canary_passed": proven, + "latest_execution": latest, + "activation_blockers": blockers, + "next_machine_action": readback.get("next_machine_action"), + "writes_database": False, + "writes_database_count": 0, + "primary_human_gate_count": 0, + }, + ) + except Exception as exc: + return _check( + "Internal RAG candidate canary", + "critical", + f"Internal RAG candidate canary readback failed: {exc}", + { + "writes_database": False, + "writes_database_count": 0, + "primary_human_gate_count": 0, + }, + ) + + def _pixelrag_rag_candidate_replay_check() -> Dict[str, Any]: """Read-only sentinel for PixelRAG receipts becoming internal RAG candidates.""" try: @@ -14252,7 +14351,9 @@ def collect_ai_automation_smoke(*, record_history: bool = True, history_limit: i _sitewide_ui_ux_agent_check(), _sitewide_visual_qa_check(), _security_governance_review_check(), + _ai_agent_product_integration_check(), _external_mcp_rag_integration_check(), + _internal_rag_candidate_canary_check(), _pixelrag_rag_candidate_replay_check(), _pixelrag_source_contract_replay_worker_check(), _pixelrag_marketplace_adapter_preflight_check(), diff --git a/services/internal_rag_candidate_canary_service.py b/services/internal_rag_candidate_canary_service.py new file mode 100644 index 0000000..e359975 --- /dev/null +++ b/services/internal_rag_candidate_canary_service.py @@ -0,0 +1,597 @@ +"""Controlled PixelRAG candidate canary for the internal pgvector RAG plane. + +The canary consumes candidate-knowledge receipts, generates Ollama-first text +embeddings, and verifies retrieval with PostgreSQL's pgvector operator inside a +read-only transaction. It never writes ai_insights or product price tables. +""" + +from __future__ import annotations + +import json +import os +import uuid +from datetime import datetime, timezone +from pathlib import Path +from typing import Any, Mapping + +from services.pixelrag_crawler_integration_service import DEFAULT_ARTIFACT_MAX_AGE_HOURS +from services.pixelrag_marketplace_candidate_knowledge_replay_service import ( + CANDIDATE_KNOWLEDGE_REPLAY_VERSION, + DEFAULT_OUTPUT_ROOT as DEFAULT_CANDIDATE_KNOWLEDGE_RECEIPT_ROOT, +) +from services.rag_service import ( + RAG_EMBED_DIM, + RAG_EMBED_MODEL, + get_embedding_signature, + is_rag_enabled, +) + + +POLICY = "controlled_internal_rag_candidate_canary_v1" +CANARY_VERSION = "internal_rag_candidate_canary_v1" +DEFAULT_LIMIT = 1 +DEFAULT_SIMILARITY_THRESHOLD = float( + os.getenv("INTERNAL_RAG_CANDIDATE_CANARY_THRESHOLD", "0.70") +) +DEFAULT_OUTPUT_ROOT = os.getenv( + "INTERNAL_RAG_CANDIDATE_CANARY_RECEIPT_ROOT", + "/app/data/ai_automation/internal_rag_candidate_canary_receipts" + if Path("/app/data").exists() + else "runtime_artifacts/internal_rag_candidate_canary_receipts", +) + + +def _as_mapping(value: Any) -> Mapping[str, Any]: + return value if isinstance(value, Mapping) else {} + + +def _as_list(value: Any) -> list[Any]: + return list(value) if isinstance(value, list) else [] + + +def _parse_datetime(value: Any) -> datetime | None: + if not value: + return None + try: + parsed = datetime.fromisoformat(str(value).replace("Z", "+00:00")) + except ValueError: + return None + if parsed.tzinfo is None: + parsed = parsed.replace(tzinfo=timezone.utc) + return parsed.astimezone(timezone.utc) + + +def _normalise_platforms( + platform: str | tuple[str, ...] | list[str] | None, +) -> tuple[str, ...]: + if isinstance(platform, str): + value = platform.strip().lower() + return (value,) if value else () + return tuple( + str(item or "").strip().lower() + for item in (platform or ()) + if str(item or "").strip() + ) + + +def _safe_segment(value: Any) -> str: + text = "".join( + char if char.isalnum() or char in "._-" else "-" + for char in str(value or "unknown").strip().lower() + ) + return text.strip("-") or "unknown" + + +def _receipt_candidates( + root: Path, + *, + platforms: tuple[str, ...], + limit: int, +) -> list[Path]: + if not root.exists(): + return [] + candidates: list[Path] = [] + if platforms: + for platform in platforms: + candidates.extend( + (root / platform).glob( + "*/marketplace_candidate_knowledge_replay_receipt.json" + ) + ) + else: + candidates.extend( + root.glob("*/*/marketplace_candidate_knowledge_replay_receipt.json") + ) + return sorted( + candidates, + key=lambda path: path.stat().st_mtime, + reverse=True, + )[:limit] + + +def _latest_execution_receipt(root: Path) -> dict[str, Any]: + if not root.exists(): + return {} + candidates = sorted( + root.glob("*/*/internal_rag_candidate_canary_receipt.json"), + key=lambda path: path.stat().st_mtime, + reverse=True, + ) + if not candidates: + return {} + path = candidates[0] + try: + payload = json.loads(path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError): + return {} + return { + "receipt_path": str(path), + "generated_at": payload.get("generated_at"), + "status": payload.get("status"), + "canary_passed": payload.get("canary_passed") is True, + "platform": payload.get("platform"), + "manifest_id": payload.get("manifest_id"), + "embedding_signature": payload.get("embedding_signature"), + "probe_similarity": payload.get("probe_similarity"), + "transaction_read_only": payload.get("transaction_read_only") is True, + "writes_ai_insights": payload.get("writes_ai_insights") is True, + "writes_price_tables": payload.get("writes_price_tables") is True, + } + + +def _load_receipt(path: Path) -> tuple[dict[str, Any], list[str]]: + try: + payload = json.loads(path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError) as exc: + return {}, [str(exc)[:300]] + return payload, [] + + +def _source_item( + path: Path, + *, + now: datetime, + max_age_hours: int, +) -> dict[str, Any]: + receipt, errors = _load_receipt(path) + knowledge = _as_mapping(receipt.get("candidate_knowledge_replay")) + contracts = [ + item + for item in _as_list(knowledge.get("candidate_knowledge_contracts")) + if isinstance(item, Mapping) + ] + generated_at = _parse_datetime(receipt.get("generated_at")) + if generated_at is None: + try: + generated_at = datetime.fromtimestamp(path.stat().st_mtime, tz=timezone.utc) + except OSError: + generated_at = None + age_hours = ((now - generated_at).total_seconds() / 3600) if generated_at else None + stale = age_hours is None or age_hours > max_age_hours + expected_signature = str( + _as_mapping(knowledge.get("embedding_signature_contract")).get( + "embedding_signature" + ) + or "" + ) + source_checks = { + "receipt_parse_ok": not errors, + "receipt_fresh": not stale, + "candidate_knowledge_version_supported": ( + knowledge.get("candidate_knowledge_replay_version") + == CANDIDATE_KNOWLEDGE_REPLAY_VERSION + ), + "candidate_knowledge_execute_completed": ( + receipt.get("worker_status") + == "executed_marketplace_candidate_knowledge_replay_ready" + ), + "candidate_contracts_present": bool(contracts), + "candidate_contracts_ready": bool(contracts) + and all( + item.get("ready_for_internal_rag_candidate_replay") is True + and item.get("ready_for_ai_insights_write") is False + and item.get("ready_for_price_table_write") is False + for item in contracts + ), + "embedding_signature_matches_runtime": ( + bool(expected_signature) + and expected_signature == get_embedding_signature() + ), + "source_database_write_absent": ( + receipt.get("writes_database") is False + and int(receipt.get("writes_database_count") or 0) == 0 + ), + "source_ai_insights_write_absent": receipt.get("writes_ai_insights") is False, + "source_price_write_absent": receipt.get("writes_price_tables") is False, + } + ready = all(source_checks.values()) + return { + "platform": str(receipt.get("platform") or path.parent.parent.name).lower(), + "manifest_id": str(receipt.get("manifest_id") or path.parent.name), + "source_receipt_path": str(path), + "generated_at": generated_at.isoformat() if generated_at else None, + "age_hours": round(age_hours, 3) if age_hours is not None else None, + "stale": stale, + "expected_embedding_signature": expected_signature, + "candidate_contracts": contracts, + "source_checks": source_checks, + "source_check_count": len(source_checks), + "source_check_pass_count": sum(source_checks.values()), + "ready_for_canary": ready, + "errors": errors, + } + + +def _probe_text(candidate: Mapping[str, Any], *, platform: str, manifest_id: str) -> str: + return " | ".join( + [ + f"platform={platform}", + f"manifest_id={manifest_id}", + f"candidate_id={candidate.get('candidate_id') or 'unknown'}", + "retrieve PixelRAG marketplace evidence for internal RAG verification", + ] + ) + + +def _generate_embedding(text: str) -> list[float]: + from services.ollama_service import ollama_service + + return list( + ollama_service.generate_embedding( + text, + model=RAG_EMBED_MODEL, + allow_111_fallback=False, + ) + or [] + ) + + +def _verify_embedding_consistency() -> dict[str, Any]: + from services.rag_service import verify_embedding_consistency + + return dict(verify_embedding_consistency()) + + +def _run_pgvector_probe( + candidate_embedding: list[float], + probe_embedding: list[float], +) -> dict[str, Any]: + from sqlalchemy import text + + from database.manager import get_session + + session = get_session() + try: + session.execute(text("SET TRANSACTION READ ONLY")) + transaction_read_only = str( + session.execute(text("SHOW transaction_read_only")).scalar() or "" + ).lower() == "on" + row = session.execute( + text( + """ + SELECT + 1.0 - ( + CAST(:candidate_embedding AS vector) + <=> CAST(:candidate_embedding AS vector) + ) AS exact_similarity, + 1.0 - ( + CAST(:candidate_embedding AS vector) + <=> CAST(:probe_embedding AS vector) + ) AS probe_similarity, + to_regclass('public.ai_insights') IS NOT NULL AS ai_insights_exists + """ + ), + { + "candidate_embedding": str(candidate_embedding), + "probe_embedding": str(probe_embedding), + }, + ).mappings().one() + return { + "transaction_read_only": transaction_read_only, + "exact_similarity": round(float(row["exact_similarity"] or 0.0), 6), + "probe_similarity": round(float(row["probe_similarity"] or 0.0), 6), + "ai_insights_table_present": bool(row["ai_insights_exists"]), + "database_write_performed": False, + } + finally: + session.rollback() + session.close() + + +def _execute_item( + item: Mapping[str, Any], + *, + consistency: Mapping[str, Any], + similarity_threshold: float, + run_identity: Mapping[str, str], +) -> dict[str, Any]: + contracts = list(item.get("candidate_contracts") or []) + candidate = contracts[0] if contracts else {} + candidate_text = str(candidate.get("candidate_knowledge_text") or "") + probe_text = _probe_text( + candidate, + platform=str(item.get("platform") or "unknown"), + manifest_id=str(item.get("manifest_id") or "unknown"), + ) + result = dict(item) + result["run_identity"] = dict(run_identity) + result["candidate_contracts"] = [] + result["candidate_id"] = candidate.get("candidate_id") + result["candidate_knowledge_fingerprint"] = candidate.get( + "candidate_knowledge_fingerprint" + ) + result["embedding_signature"] = get_embedding_signature() + result["embedding_model"] = RAG_EMBED_MODEL + result["similarity_threshold"] = similarity_threshold + result["network_call_performed"] = True + result["model_call_performed"] = True + result["writes_database"] = False + result["writes_ai_insights"] = False + result["writes_price_tables"] = False + try: + candidate_embedding = _generate_embedding(candidate_text) + probe_embedding = _generate_embedding(probe_text) + pgvector = _run_pgvector_probe(candidate_embedding, probe_embedding) + reachable = list(consistency.get("reachable") or []) + required_hosts = {"gcp_ollama", "ollama_secondary"} + checks = { + "candidate_embedding_dimension_valid": ( + len(candidate_embedding) == RAG_EMBED_DIM + ), + "probe_embedding_dimension_valid": len(probe_embedding) == RAG_EMBED_DIM, + "cross_host_embedding_consistent": ( + consistency.get("ok") is True + and required_hosts.issubset(set(reachable)) + ), + "pgvector_transaction_read_only": ( + pgvector.get("transaction_read_only") is True + ), + "pgvector_exact_similarity_valid": ( + float(pgvector.get("exact_similarity") or 0.0) >= 0.999 + ), + "pgvector_probe_similarity_passed": ( + float(pgvector.get("probe_similarity") or 0.0) + >= similarity_threshold + ), + "database_write_absent": ( + pgvector.get("database_write_performed") is False + ), + "ai_insights_write_absent": True, + "price_table_write_absent": True, + } + passed = all(checks.values()) + result.update( + { + "status": "canary_passed" if passed else "canary_failed", + "canary_passed": passed, + "candidate_embedding_dimension": len(candidate_embedding), + "probe_embedding_dimension": len(probe_embedding), + "exact_similarity": pgvector.get("exact_similarity"), + "probe_similarity": pgvector.get("probe_similarity"), + "transaction_read_only": pgvector.get("transaction_read_only"), + "pgvector_probe": pgvector, + "required_consistency_hosts": sorted(required_hosts), + "reachable_consistency_hosts": reachable, + "canary_checks": checks, + "canary_check_count": len(checks), + "canary_check_pass_count": sum(checks.values()), + "error": None, + } + ) + except Exception as exc: + result.update( + { + "status": "canary_failed", + "canary_passed": False, + "transaction_read_only": False, + "canary_checks": {}, + "canary_check_count": 0, + "canary_check_pass_count": 0, + "error": f"{type(exc).__name__}: {str(exc)[:300]}", + } + ) + return result + + +def _write_receipt(root: Path, item: Mapping[str, Any]) -> str: + target = ( + root + / _safe_segment(item.get("platform")) + / _safe_segment(item.get("manifest_id")) + / "internal_rag_candidate_canary_receipt.json" + ) + target.parent.mkdir(parents=True, exist_ok=True) + payload = dict(item) + payload["policy"] = POLICY + payload["canary_version"] = CANARY_VERSION + payload["generated_at"] = datetime.now(timezone.utc).isoformat() + payload["artifact_write_performed"] = True + payload["receipt_path"] = str(target) + target.write_text( + json.dumps(payload, ensure_ascii=False, indent=2, sort_keys=True), + encoding="utf-8", + ) + return str(target) + + +def run_internal_rag_candidate_canary( + *, + candidate_knowledge_receipt_root: str | Path | None = None, + output_root: str | Path | None = None, + platform: str | tuple[str, ...] | list[str] | None = None, + max_age_hours: int | None = None, + limit: int | None = None, + similarity_threshold: float | None = None, + execute: bool = False, + write_receipt: bool = False, + trace_id: str | None = None, + run_id: str | None = None, + work_item_id: str = "RAG-P0-001", +) -> dict[str, Any]: + """Dry-run or execute a bounded, read-only pgvector retrieval canary.""" + source_root = Path( + candidate_knowledge_receipt_root + or DEFAULT_CANDIDATE_KNOWLEDGE_RECEIPT_ROOT + ) + receipt_root = Path(output_root or DEFAULT_OUTPUT_ROOT) + platforms = _normalise_platforms(platform) + item_limit = max(1, min(int(limit or DEFAULT_LIMIT), 5)) + max_age = max(1, int(max_age_hours or DEFAULT_ARTIFACT_MAX_AGE_HOURS)) + threshold = max( + 0.0, + min(float(similarity_threshold or DEFAULT_SIMILARITY_THRESHOLD), 1.0), + ) + resolved_run_id = str(run_id or f"rag-canary-{uuid.uuid4()}") + run_identity = { + "trace_id": str(trace_id or f"trace-{resolved_run_id}"), + "run_id": resolved_run_id, + "work_item_id": str(work_item_id or "RAG-P0-001"), + } + now = datetime.now(timezone.utc) + source_items = [ + _source_item(path, now=now, max_age_hours=max_age) + for path in _receipt_candidates( + source_root, + platforms=platforms, + limit=item_limit, + ) + ] + ready_items = [item for item in source_items if item.get("ready_for_canary")] + consistency: dict[str, Any] = {} + executed_items: list[dict[str, Any]] = [] + if execute and ready_items: + consistency = _verify_embedding_consistency() + for item in ready_items: + executed = _execute_item( + item, + consistency=consistency, + similarity_threshold=threshold, + run_identity=run_identity, + ) + if write_receipt: + executed["receipt_path"] = _write_receipt(receipt_root, executed) + executed["artifact_write_performed"] = True + executed_items.append(executed) + + canary_passed_count = sum( + 1 for item in executed_items if item.get("canary_passed") is True + ) + canary_failed_count = len(executed_items) - canary_passed_count + activation_blockers: list[str] = [] + if not is_rag_enabled(): + activation_blockers.append("rag_runtime_disabled") + if RAG_EMBED_MODEL.endswith(":latest"): + activation_blockers.append("rag_embedding_model_floating_tag") + latest_execution = _latest_execution_receipt(receipt_root) + historical_canary_passed = latest_execution.get("canary_passed") is True + + if not source_items: + status = "warning" + next_action = "run_marketplace_candidate_knowledge_replay_execute" + elif not ready_items: + status = "blocked" + next_action = "repair_candidate_knowledge_receipt_guards" + elif not execute: + status = "ready_for_canary" + next_action = "run_internal_rag_candidate_canary_execute" + elif canary_failed_count: + status = "canary_failed" + next_action = "repair_embedding_or_pgvector_canary_failure" + elif activation_blockers: + status = "canary_passed_activation_blocked" + next_action = "pin_embedding_model_then_enable_rag_controlled_canary" + else: + status = "complete" + next_action = "continue_scheduled_internal_rag_canary" + + return { + "success": bool(source_items) and bool(ready_items) and not canary_failed_count, + "run_identity": run_identity, + "policy": POLICY, + "canary_version": CANARY_VERSION, + "generated_at": now.isoformat(), + "status": status, + "execute": bool(execute), + "source_receipt_root": str(source_root), + "output_root": str(receipt_root), + "summary": { + "source_receipt_count": len(source_items), + "ready_count": len(ready_items), + "blocked_count": len(source_items) - len(ready_items), + "executed_count": len(executed_items), + "canary_passed_count": canary_passed_count, + "canary_failed_count": canary_failed_count, + "historical_canary_passed": historical_canary_passed, + }, + "embedding": { + "model": RAG_EMBED_MODEL, + "dimension": RAG_EMBED_DIM, + "signature": get_embedding_signature(), + "immutable_model_reference": not RAG_EMBED_MODEL.endswith(":latest"), + "cross_host_consistency": consistency, + }, + "source_items": source_items, + "executed_items": executed_items, + "latest_execution": latest_execution, + "activation_blockers": activation_blockers, + "source_of_truth_diff": { + "expected_embedding_signature": get_embedding_signature(), + "observed_embedding_signatures": sorted({ + str(item.get("expected_embedding_signature") or "") + for item in source_items + if item.get("expected_embedding_signature") + }), + "rag_runtime_expected_enabled": True, + "rag_runtime_observed_enabled": is_rag_enabled(), + "candidate_receipt_expected_minimum": 1, + "candidate_receipt_observed": len(source_items), + }, + "controlled_apply": { + "risk": "medium", + "bounded_candidate_limit": item_limit, + "network_call": bool(execute and ready_items), + "model_call": bool(execute and ready_items), + "database_transaction_read_only": True, + "database_write": False, + "ai_insights_write": False, + "price_table_write": False, + "artifact_write": bool(execute and write_receipt), + "rollback_terminal": "transaction_rollback_after_read_only_pgvector_probe", + "independent_verifier": "pgvector_read_only_similarity_probe", + }, + "closure_receipt": { + "sensor_source_receipt": bool(source_items), + "normalized_asset_identity": bool(source_items) and all( + item.get("platform") and item.get("manifest_id") + for item in source_items + ), + "source_of_truth_diff_recorded": True, + "ai_candidate_decision_recorded": True, + "risk_policy_decision": "medium_bounded_read_only_canary", + "check_mode_passed": bool(ready_items), + "bounded_execution_performed": bool(executed_items), + "independent_verifier_passed": ( + bool(executed_items) + and canary_passed_count == len(executed_items) + ), + "rollback_or_no_write_terminal": ( + "transaction_rollback_after_read_only_pgvector_probe" + if executed_items + else "no_write_terminal" + ), + "telegram_acknowledgement": "pending_scheduler_dispatch" + if execute + else "not_applicable_dry_run", + "learning_write_acknowledgement": "rag_canary_receipt_written" + if executed_items and write_receipt + else "no_learning_write", + }, + "next_machine_action": next_action, + } + + +__all__ = [ + "CANARY_VERSION", + "POLICY", + "run_internal_rag_candidate_canary", +] diff --git a/tests/test_ai_agent_product_integration_schema_contract.py b/tests/test_ai_agent_product_integration_schema_contract.py new file mode 100644 index 0000000..748e417 --- /dev/null +++ b/tests/test_ai_agent_product_integration_schema_contract.py @@ -0,0 +1,17 @@ +def test_runtime_telemetry_uses_rag_query_log_queried_at(monkeypatch): + from services import ai_agent_product_integration_service as service + + sql_seen = [] + + def fake_query(sql, _params): + sql_seen.append(sql) + return [], None + + monkeypatch.setattr(service, "_query_rows", fake_query) + service._collect_runtime_telemetry( + service.datetime.now(service.timezone.utc) + ) + + rag_sql = next(sql for sql in sql_seen if "FROM rag_query_log" in sql) + assert "queried_at >= :since_at" in rag_sql + assert "created_at >= :since_at" not in rag_sql diff --git a/tests/test_ai_agent_product_integration_service.py b/tests/test_ai_agent_product_integration_service.py new file mode 100644 index 0000000..3458511 --- /dev/null +++ b/tests/test_ai_agent_product_integration_service.py @@ -0,0 +1,159 @@ +import json +import subprocess +import sys + + +def _telemetry(*, all_active): + from services.ai_agent_product_integration_service import AGENT_SPECS + + agents = {} + for index, key in enumerate(AGENT_SPECS): + active = all_active or index < 2 + agents[key] = { + "calls": 5 if active else 0, + "errors": 0, + "rag_hits": 1 if active else 0, + "fallbacks": 0, + "mcp_calls": 1 if all_active else 0, + "rag_queries": 1 if all_active else 0, + "rag_query_hits": 1 if all_active else 0, + "ai_insights": 1 if active else 0, + "action_plans": 1 if active else 0, + "executed_action_plans": 1 if all_active else 0, + "last_called": "2026-07-17T00:00:00+00:00" if active else None, + } + return { + "agents": agents, + "unmatched_ai_callers": [], + "totals": { + "ai_calls": sum(item["calls"] for item in agents.values()), + "ai_errors": 0, + "mcp_calls": 4 if all_active else 0, + "rag_queries": 4 if all_active else 0, + "rag_query_hits": 4 if all_active else 0, + "ai_insights": 4, + "action_plans": 4, + "executed_action_plans": 4 if all_active else 0, + "action_outcomes": 2 if all_active else 0, + "heal_logs": 2 if all_active else 0, + "heal_success": 2 if all_active else 0, + "verified_retry_or_rollback_incidents": 1 if all_active else 0, + }, + "embedding_retry_queue": {"done": 8}, + "read_errors": [], + } + + +def _external_runtime(*, enabled): + return { + "runtime": { + "mcp": {"enabled": enabled}, + "rag": {"enabled": enabled}, + "pixelrag": {"enabled": True, "platform_count": 8}, + } + } + + +def _canary(*, passed): + return {"latest_execution": {"canary_passed": passed}} + + +def test_ai_agent_integration_reports_production_like_partial_truth(monkeypatch): + from services import ai_agent_product_integration_service as service + + monkeypatch.setattr( + service, + "_collect_runtime_telemetry", + lambda _since: _telemetry(all_active=False), + ) + monkeypatch.setattr( + service, + "build_external_mcp_rag_integration_readback", + lambda: _external_runtime(enabled=False), + ) + monkeypatch.setattr( + service, + "run_internal_rag_candidate_canary", + lambda **_kwargs: _canary(passed=False), + ) + payload = service.build_ai_agent_product_integration_readback(window_hours=168) + + assert payload["status"] == "partially_integrated" + assert payload["completion"]["source_wired_agents"] == 4 + assert payload["completion"]["runtime_active_agents"] == 2 + assert payload["completion"]["full_product_integration"] is False + assert "mcp_router_runtime_disabled" in payload["blockers"] + assert "rag_runtime_disabled" in payload["blockers"] + assert "尚未完整整合" in payload["answer_to_owner"] + + +def test_ai_agent_integration_requires_full_runtime_closure(monkeypatch): + from services import ai_agent_product_integration_service as service + + monkeypatch.setattr( + service, + "_collect_runtime_telemetry", + lambda _since: _telemetry(all_active=True), + ) + monkeypatch.setattr( + service, + "build_external_mcp_rag_integration_readback", + lambda: _external_runtime(enabled=True), + ) + monkeypatch.setattr( + service, + "run_internal_rag_candidate_canary", + lambda **_kwargs: _canary(passed=True), + ) + payload = service.build_ai_agent_product_integration_readback() + + assert payload["status"] == "fully_integrated" + assert payload["completion"]["runtime_active_agents"] == 4 + assert payload["completion"]["closure_stage_passed"] == 9 + assert payload["completion"]["full_product_integration"] is True + assert payload["blockers"] == [] + + +def test_ai_agent_integration_route_returns_truth(monkeypatch): + from flask import Flask + from routes import system_public_routes as routes + from services import ai_agent_product_integration_service as service + + monkeypatch.setattr( + service, + "build_ai_agent_product_integration_readback", + lambda **kwargs: { + "success": True, + "policy": service.POLICY, + "status": "partially_integrated", + "window_hours": kwargs["window_hours"], + }, + ) + app = Flask(__name__) + with app.test_request_context( + "/api/ai-automation/agent-product-integration?window_hours=72" + ): + response = routes.ai_automation_agent_product_integration_api.__wrapped__() + payload = response.get_json() + + assert payload["policy"] == "runtime_truth_ai_agent_product_integration_v1" + assert payload["window_hours"] == 72 + + +def test_ai_agent_integration_cli_outputs_json(): + completed = subprocess.run( + [ + sys.executable, + "scripts/ops/report_ai_agent_product_integration.py", + "--window-hours", + "1", + ], + capture_output=True, + check=False, + text=True, + ) + + assert completed.returncode == 0 + payload = json.loads(completed.stdout) + assert payload["policy"] == "runtime_truth_ai_agent_product_integration_v1" + assert payload["controlled_apply"]["database_write"] is False diff --git a/tests/test_ai_automation_smoke_service.py b/tests/test_ai_automation_smoke_service.py index 1dab687..3d86a50 100644 --- a/tests/test_ai_automation_smoke_service.py +++ b/tests/test_ai_automation_smoke_service.py @@ -1309,7 +1309,9 @@ def test_collect_ai_automation_smoke_uses_worst_status(monkeypatch): monkeypatch.setattr(smoke, "_sitewide_ui_ux_agent_check", lambda: smoke._check("sitewide", "ok", "ok")) monkeypatch.setattr(smoke, "_sitewide_visual_qa_check", lambda: smoke._check("visual", "ok", "ok")) monkeypatch.setattr(smoke, "_security_governance_review_check", lambda: smoke._check("security governance", "ok", "ok")) + monkeypatch.setattr(smoke, "_ai_agent_product_integration_check", lambda: smoke._check("agent integration", "ok", "ok")) monkeypatch.setattr(smoke, "_external_mcp_rag_integration_check", lambda: smoke._check("external mcp rag", "ok", "ok")) + monkeypatch.setattr(smoke, "_internal_rag_candidate_canary_check", lambda: smoke._check("internal rag canary", "ok", "ok")) monkeypatch.setattr(smoke, "_pixelrag_rag_candidate_replay_check", lambda: smoke._check("pixelrag replay", "ok", "ok")) monkeypatch.setattr(smoke, "_pixelrag_source_contract_replay_worker_check", lambda: smoke._check("pixelrag source contract", "ok", "ok")) monkeypatch.setattr(smoke, "_pixelrag_marketplace_adapter_preflight_check", lambda: smoke._check("pixelrag marketplace adapter preflight", "ok", "ok")) @@ -1327,7 +1329,7 @@ def test_collect_ai_automation_smoke_uses_worst_status(monkeypatch): result = smoke.collect_ai_automation_smoke(record_history=False) assert result["status"] == "critical" - assert result["summary"] == {"ok": 43, "warning": 1, "critical": 1, "total": 45} + assert result["summary"] == {"ok": 45, "warning": 1, "critical": 1, "total": 47} def test_pchome_controlled_apply_drift_monitor_reports_verified_zero_drift(monkeypatch): @@ -3855,7 +3857,9 @@ def test_collect_ai_automation_smoke_persists_recent_history(tmp_path, monkeypat monkeypatch.setattr(smoke, "_sitewide_ui_ux_agent_check", lambda: smoke._check("sitewide", "ok", "ok")) monkeypatch.setattr(smoke, "_sitewide_visual_qa_check", lambda: smoke._check("visual", "ok", "ok")) monkeypatch.setattr(smoke, "_security_governance_review_check", lambda: smoke._check("security governance", "ok", "ok")) + monkeypatch.setattr(smoke, "_ai_agent_product_integration_check", lambda: smoke._check("agent integration", "ok", "ok")) monkeypatch.setattr(smoke, "_external_mcp_rag_integration_check", lambda: smoke._check("external mcp rag", "ok", "ok")) + monkeypatch.setattr(smoke, "_internal_rag_candidate_canary_check", lambda: smoke._check("internal rag canary", "ok", "ok")) monkeypatch.setattr(smoke, "_pixelrag_rag_candidate_replay_check", lambda: smoke._check("pixelrag replay", "ok", "ok")) monkeypatch.setattr(smoke, "_pixelrag_source_contract_replay_worker_check", lambda: smoke._check("pixelrag source contract", "ok", "ok")) monkeypatch.setattr(smoke, "_pixelrag_marketplace_adapter_preflight_check", lambda: smoke._check("pixelrag marketplace adapter preflight", "ok", "ok")) @@ -6647,11 +6651,21 @@ def test_surface_html_readback_check_is_part_of_ai_smoke(monkeypatch): "ok", "security governance ok", )) + monkeypatch.setattr(smoke, "_ai_agent_product_integration_check", lambda: smoke._check( + "AI Agent product integration truth", + "ok", + "agent integration ok", + )) monkeypatch.setattr(smoke, "_external_mcp_rag_integration_check", lambda: smoke._check( "External MCP/RAG integration readback", "ok", "external mcp rag ok", )) + monkeypatch.setattr(smoke, "_internal_rag_candidate_canary_check", lambda: smoke._check( + "Internal RAG candidate canary", + "ok", + "internal rag canary ok", + )) monkeypatch.setattr(smoke, "_pixelrag_rag_candidate_replay_check", lambda: smoke._check( "PixelRAG RAG candidate replay", "ok", @@ -6732,7 +6746,7 @@ def test_surface_html_readback_check_is_part_of_ai_smoke(monkeypatch): item for item in result["checks"] if item["name"] == "Sitewide visual QA readback" ) - assert result["summary"]["total"] == 45 + assert result["summary"]["total"] == 47 assert surface_check["status"] == "ok" assert surface_check["details"]["checked_surface_count"] == 10 assert sitewide_check["status"] == "ok" diff --git a/tests/test_internal_rag_candidate_canary_service.py b/tests/test_internal_rag_candidate_canary_service.py new file mode 100644 index 0000000..c28069b --- /dev/null +++ b/tests/test_internal_rag_candidate_canary_service.py @@ -0,0 +1,227 @@ +import json +import inspect +import subprocess +import sys +from datetime import datetime, timezone + + +def _write_candidate_receipt(root, *, signature=None, platform="shopee_tw"): + from services.rag_service import get_embedding_signature + + embedding_signature = signature or get_embedding_signature() + manifest_id = "shopee-canary-001" + target = root / platform / manifest_id + target.mkdir(parents=True) + payload = { + "generated_at": datetime.now(timezone.utc).isoformat(), + "worker_status": "executed_marketplace_candidate_knowledge_replay_ready", + "platform": platform, + "manifest_id": manifest_id, + "writes_database": False, + "writes_database_count": 0, + "writes_ai_insights": False, + "writes_price_tables": False, + "candidate_knowledge_replay": { + "candidate_knowledge_replay_version": "pixelrag_marketplace_candidate_knowledge_replay_v1", + "embedding_signature_contract": {"embedding_signature": embedding_signature}, + "candidate_knowledge_contracts": [{ + "candidate_id": f"{platform}:{manifest_id}:0", + "candidate_knowledge_fingerprint": "candidate-fingerprint-001", + "candidate_knowledge_text": ( + f"platform={platform} | manifest_id={manifest_id} | " + "candidate_id=0 | product evidence" + ), + "ready_for_internal_rag_candidate_replay": True, + "ready_for_ai_insights_write": False, + "ready_for_price_table_write": False, + }], + }, + } + path = target / "marketplace_candidate_knowledge_replay_receipt.json" + path.write_text(json.dumps(payload), encoding="utf-8") + return path + + +def test_internal_rag_canary_dry_run_never_calls_model_or_database(tmp_path, monkeypatch): + from services import internal_rag_candidate_canary_service as service + + source_root = tmp_path / "candidate" + _write_candidate_receipt(source_root) + monkeypatch.setattr( + service, + "_generate_embedding", + lambda _text: (_ for _ in ()).throw(AssertionError("model call forbidden")), + ) + monkeypatch.setattr( + service, + "_run_pgvector_probe", + lambda *_args: (_ for _ in ()).throw(AssertionError("DB call forbidden")), + ) + payload = service.run_internal_rag_candidate_canary( + candidate_knowledge_receipt_root=source_root, + ) + + assert payload["status"] == "ready_for_canary" + assert payload["summary"]["ready_count"] == 1 + assert payload["summary"]["executed_count"] == 0 + assert payload["controlled_apply"]["model_call"] is False + assert payload["controlled_apply"]["database_write"] is False + + +def test_internal_rag_canary_executes_read_only_pgvector_probe(tmp_path, monkeypatch): + from services import internal_rag_candidate_canary_service as service + + source_root = tmp_path / "candidate" + output_root = tmp_path / "canary" + _write_candidate_receipt(source_root) + monkeypatch.setattr( + service, "_generate_embedding", lambda _text: [0.01] * service.RAG_EMBED_DIM + ) + monkeypatch.setattr( + service, + "_verify_embedding_consistency", + lambda: { + "ok": True, + "reachable": ["gcp_ollama", "ollama_secondary"], + "max_diff": 0.0, + "errors": [], + }, + ) + monkeypatch.setattr( + service, + "_run_pgvector_probe", + lambda *_args: { + "transaction_read_only": True, + "exact_similarity": 1.0, + "probe_similarity": 0.91, + "ai_insights_table_present": True, + "database_write_performed": False, + }, + ) + payload = service.run_internal_rag_candidate_canary( + candidate_knowledge_receipt_root=source_root, + output_root=output_root, + execute=True, + write_receipt=True, + ) + + assert payload["summary"]["canary_passed_count"] == 1 + assert payload["status"] in {"canary_passed_activation_blocked", "complete"} + item = payload["executed_items"][0] + assert item["canary_passed"] is True + assert item["transaction_read_only"] is True + assert item["writes_database"] is False + assert item["writes_ai_insights"] is False + assert item["writes_price_tables"] is False + assert payload["latest_execution"]["canary_passed"] is True + assert payload["run_identity"]["work_item_id"] == "RAG-P0-001" + assert item["run_identity"] == payload["run_identity"] + + +def test_internal_rag_canary_requires_both_primary_gcp_hosts(tmp_path, monkeypatch): + from services import internal_rag_candidate_canary_service as service + + source_root = tmp_path / "candidate" + _write_candidate_receipt(source_root) + monkeypatch.setattr( + service, "_generate_embedding", lambda _text: [0.01] * service.RAG_EMBED_DIM + ) + monkeypatch.setattr( + service, + "_verify_embedding_consistency", + lambda: { + "ok": True, + "reachable": ["gcp_ollama", "ollama_111"], + "max_diff": 0.0, + "errors": ["ollama_secondary unavailable"], + }, + ) + monkeypatch.setattr( + service, + "_run_pgvector_probe", + lambda *_args: { + "transaction_read_only": True, + "exact_similarity": 1.0, + "probe_similarity": 0.91, + "ai_insights_table_present": True, + "database_write_performed": False, + }, + ) + + payload = service.run_internal_rag_candidate_canary( + candidate_knowledge_receipt_root=source_root, + execute=True, + ) + + assert payload["status"] == "canary_failed" + item = payload["executed_items"][0] + assert item["canary_checks"]["cross_host_embedding_consistent"] is False + assert item["required_consistency_hosts"] == ["gcp_ollama", "ollama_secondary"] + + +def test_internal_rag_canary_blocks_embedding_signature_drift(tmp_path): + from services.internal_rag_candidate_canary_service import run_internal_rag_candidate_canary + + source_root = tmp_path / "candidate" + _write_candidate_receipt(source_root, signature="deadbeef0000") + payload = run_internal_rag_candidate_canary( + candidate_knowledge_receipt_root=source_root, + ) + + assert payload["status"] == "blocked" + assert payload["summary"]["ready_count"] == 0 + assert payload["source_items"][0]["source_checks"][ + "embedding_signature_matches_runtime" + ] is False + + +def test_internal_rag_canary_cli_outputs_machine_readable_dry_run(tmp_path): + source_root = tmp_path / "candidate" + _write_candidate_receipt(source_root) + completed = subprocess.run( + [ + sys.executable, + "scripts/ops/run_internal_rag_candidate_canary.py", + "--candidate-knowledge-receipt-root", + str(source_root), + ], + capture_output=True, + check=False, + text=True, + ) + + assert completed.returncode == 0 + payload = json.loads(completed.stdout) + assert payload["status"] == "ready_for_canary" + assert payload["controlled_apply"]["database_write"] is False + + +def test_internal_rag_canary_route_returns_readback(tmp_path, monkeypatch): + from flask import Flask + from routes import system_public_routes as routes + from services import internal_rag_candidate_canary_service as service + + source_root = tmp_path / "candidate" + _write_candidate_receipt(source_root, platform="coupang_tw") + monkeypatch.setattr( + service, "DEFAULT_CANDIDATE_KNOWLEDGE_RECEIPT_ROOT", str(source_root) + ) + app = Flask(__name__) + with app.test_request_context( + "/api/ai-automation/internal-rag-candidate-canary?platform=coupang_tw&execute=true" + ): + response = routes.ai_automation_internal_rag_candidate_canary_api.__wrapped__() + payload = response.get_json() + + assert payload["policy"] == "controlled_internal_rag_candidate_canary_v1" + assert payload["summary"]["ready_count"] == 1 + assert payload["execute"] is False + + +def test_internal_rag_canary_http_route_is_read_only(): + from routes import system_public_routes as routes + + source = inspect.getsource(routes.ai_automation_internal_rag_candidate_canary_api) + assert "execute=False" in source + assert "write_receipt=False" in source + assert "get_current_user" not in source diff --git a/tests/test_run_scheduler_embed_consistency.py b/tests/test_run_scheduler_embed_consistency.py index 351cc68..68bf2cb 100644 --- a/tests/test_run_scheduler_embed_consistency.py +++ b/tests/test_run_scheduler_embed_consistency.py @@ -215,6 +215,7 @@ def test_v2_cron_blind_spot_list_has_failure_notifications(monkeypatch): "run_ppt_vision_audit", "run_embed_consistency_check", "run_ollama_111_usage_guard_check", + "run_internal_rag_candidate_canary_task", ]: source = inspect.getsource(getattr(run_scheduler, fn_name)) assert "_notify_scheduler_failure(" in source @@ -236,6 +237,50 @@ def test_roi_ai_smoke_and_daily_report_schedules_stay_staggered(): assert "start_revenue_automation_watchdog()" in source assert "schedule.every(6).hours.do(run_action_plan_hygiene_task)" in source assert "schedule.every(15).minutes.do(run_ollama_111_usage_guard_check)" in source + assert 'schedule.every().day.at("04:45").do(run_internal_rag_candidate_canary_task)' in source + + +def test_scheduled_internal_rag_canary_is_bounded_and_acknowledged(monkeypatch): + run_scheduler = _load_run_scheduler(monkeypatch) + import services.internal_rag_candidate_canary_service as canary_service + import services.telegram_templates as telegram_templates + + calls = [] + saved = [] + monkeypatch.setattr( + canary_service, + "run_internal_rag_candidate_canary", + lambda **kwargs: calls.append(kwargs) or { + "success": True, + "status": "canary_passed_activation_blocked", + "run_identity": { + "trace_id": "trace-test", + "run_id": "run-test", + "work_item_id": "RAG-P0-001", + }, + "summary": { + "source_receipt_count": 1, + "executed_count": 1, + "canary_passed_count": 1, + }, + "closure_receipt": {}, + "next_machine_action": "pin_embedding_model_then_enable_rag_controlled_canary", + }, + ) + monkeypatch.setattr( + telegram_templates, + "send_telegram_with_result", + lambda *_args, **_kwargs: {"ok": True, "sent": 1, "failed": 0}, + ) + monkeypatch.setattr(run_scheduler, "_save_stats", lambda name, data: saved.append((name, data))) + + payload = run_scheduler.run_internal_rag_candidate_canary_task() + + assert calls == [{"limit": 1, "execute": True, "write_receipt": True}] + assert payload["terminal_status"] == "verified_with_activation_blocker" + assert payload["acknowledgements"]["telegram"]["status"] == "acknowledged" + assert payload["acknowledgements"]["rag"] == "rag_canary_receipt_written" + assert saved[0][0] == "internal_rag_candidate_canary" def test_pchome_growth_backfill_catchup_uses_runtime_receipt_cooldown(monkeypatch, tmp_path):