清理 EA 過期行動隊列

services/action_plan_hygiene.py

@@ -0,0 +1,248 @@
+"""Action plan queue hygiene.
+
+Closes stale, non-executable automation suggestions without deleting audit data.
+The service is intentionally conservative: it only handles sources that are
+known to be advisory/noisy when old, and it records the closure in metadata_json.
+"""
+
+import json
+import os
+from datetime import datetime, timedelta
+from typing import Any, Dict, Iterable, List, Optional
+
+from sqlalchemy import text
+
+from database.manager import get_session
+from services.logger_manager import SystemLogger
+
+logger = SystemLogger("ActionPlanHygiene").get_logger()
+
+DEFAULT_STALE_HOURS = int(os.getenv("ACTION_PLAN_HYGIENE_STALE_HOURS", "72"))
+DEFAULT_MAX_UPDATES = int(os.getenv("ACTION_PLAN_HYGIENE_MAX_UPDATES", "200"))
+
+CLOSABLE_STATUSES = frozenset({"pending", "auto_pending", "pending_review"})
+SOURCE_TARGET_STATUS = {
+    "code_review_fix": "auto_disabled",
+    "code_review_pipeline": "auto_disabled",
+    "openclaw_recommendation": "rejected",
+    "openclaw": "rejected",
+}
+
+
+def _row_get(row: Any, key: str) -> Any:
+    if isinstance(row, dict):
+        return row.get(key)
+    if hasattr(row, "_mapping"):
+        return row._mapping.get(key)
+    try:
+        return row[key]
+    except Exception:
+        return getattr(row, key, None)
+
+
+def _coerce_datetime(value: Any) -> Optional[datetime]:
+    if isinstance(value, datetime):
+        return value.replace(tzinfo=None)
+    if isinstance(value, str) and value.strip():
+        try:
+            return datetime.fromisoformat(value.replace("Z", "+00:00")).replace(tzinfo=None)
+        except ValueError:
+            return None
+    return None
+
+
+def _source_for_row(row: Any) -> str:
+    return str(
+        _row_get(row, "action_type")
+        or _row_get(row, "plan_type")
+        or _row_get(row, "created_by")
+        or "unknown"
+    )
+
+
+def _parse_metadata(raw: Any) -> Dict[str, Any]:
+    if isinstance(raw, dict):
+        return dict(raw)
+    if not raw:
+        return {}
+    try:
+        parsed = json.loads(str(raw))
+        return parsed if isinstance(parsed, dict) else {"legacy_metadata": parsed}
+    except Exception:
+        return {"legacy_metadata_raw": str(raw)[:1000]}
+
+
+def build_action_plan_hygiene_preview(
+    rows: Iterable[Any],
+    *,
+    now: Optional[datetime] = None,
+    stale_hours: int = DEFAULT_STALE_HOURS,
+    max_updates: int = DEFAULT_MAX_UPDATES,
+) -> Dict[str, Any]:
+    """Return stale action plan candidates without mutating the database."""
+    now = (now or datetime.now()).replace(tzinfo=None)
+    cutoff = now - timedelta(hours=stale_hours)
+    candidates: List[Dict[str, Any]] = []
+    scanned = 0
+
+    for row in rows:
+        scanned += 1
+        status = str(_row_get(row, "status") or "")
+        source = _source_for_row(row)
+        created_at = _coerce_datetime(_row_get(row, "created_at"))
+        if status not in CLOSABLE_STATUSES:
+            continue
+        if source not in SOURCE_TARGET_STATUS:
+            continue
+        if not created_at or created_at > cutoff:
+            continue
+
+        age_hours = max(0.0, (now - created_at).total_seconds() / 3600.0)
+        candidates.append({
+            "id": int(_row_get(row, "id")),
+            "source": source,
+            "from_status": status,
+            "to_status": SOURCE_TARGET_STATUS[source],
+            "priority": _row_get(row, "priority"),
+            "created_at": created_at.isoformat(sep=" "),
+            "age_hours": round(age_hours, 1),
+            "description": str(_row_get(row, "description") or "")[:160],
+            "reason": f"stale_{source}_older_than_{stale_hours}h",
+        })
+
+    candidates = candidates[:max_updates]
+    by_source: Dict[str, int] = {}
+    by_from_status: Dict[str, int] = {}
+    for item in candidates:
+        by_source[item["source"]] = by_source.get(item["source"], 0) + 1
+        by_from_status[item["from_status"]] = by_from_status.get(item["from_status"], 0) + 1
+
+    return {
+        "scanned_count": scanned,
+        "candidate_count": len(candidates),
+        "stale_hours": stale_hours,
+        "max_updates": max_updates,
+        "by_source": by_source,
+        "by_from_status": by_from_status,
+        "candidates": candidates,
+    }
+
+
+class ActionPlanHygieneService:
+    """Close stale advisory action_plans while preserving audit metadata."""
+
+    def __init__(self, stale_hours: int = DEFAULT_STALE_HOURS, max_updates: int = DEFAULT_MAX_UPDATES):
+        self.stale_hours = stale_hours
+        self.max_updates = max_updates
+
+    def preview(self) -> Dict[str, Any]:
+        session = get_session()
+        try:
+            rows = session.execute(text("""
+                SELECT id, status, priority, created_at, action_type, plan_type,
+                       created_by, description, metadata_json
+                FROM action_plans
+                WHERE status IN ('pending', 'auto_pending', 'pending_review')
+                  AND (
+                    action_type IN ('code_review_fix', 'openclaw_recommendation')
+                    OR created_by IN ('code_review_pipeline', 'openclaw')
+                  )
+                ORDER BY created_at ASC
+            """)).fetchall()
+            return build_action_plan_hygiene_preview(
+                rows,
+                stale_hours=self.stale_hours,
+                max_updates=self.max_updates,
+            )
+        finally:
+            session.close()
+
+    def run(self) -> Dict[str, Any]:
+        session = get_session()
+        now = datetime.now()
+        try:
+            rows = session.execute(text("""
+                SELECT id, status, priority, created_at, action_type, plan_type,
+                       created_by, description, metadata_json
+                FROM action_plans
+                WHERE status IN ('pending', 'auto_pending', 'pending_review')
+                  AND (
+                    action_type IN ('code_review_fix', 'openclaw_recommendation')
+                    OR created_by IN ('code_review_pipeline', 'openclaw')
+                  )
+                ORDER BY created_at ASC
+            """)).fetchall()
+            preview = build_action_plan_hygiene_preview(
+                rows,
+                now=now,
+                stale_hours=self.stale_hours,
+                max_updates=self.max_updates,
+            )
+
+            row_by_id = {int(_row_get(row, "id")): row for row in rows}
+            updated: List[Dict[str, Any]] = []
+            for item in preview["candidates"]:
+                row = row_by_id.get(item["id"])
+                if not row:
+                    continue
+                metadata = _parse_metadata(_row_get(row, "metadata_json"))
+                history = metadata.get("hygiene_history")
+                if not isinstance(history, list):
+                    history = []
+                history.append({
+                    "closed_at": now.isoformat(timespec="seconds"),
+                    "from_status": item["from_status"],
+                    "to_status": item["to_status"],
+                    "reason": item["reason"],
+                    "age_hours": item["age_hours"],
+                })
+                metadata["hygiene_history"] = history[-10:]
+                metadata["hygiene_last_reason"] = item["reason"]
+                metadata["hygiene_last_closed_at"] = now.isoformat(timespec="seconds")
+
+                session.execute(
+                    text("""
+                        UPDATE action_plans
+                        SET status = :status,
+                            metadata_json = :metadata
+                        WHERE id = :id
+                    """),
+                    {
+                        "id": item["id"],
+                        "status": item["to_status"],
+                        "metadata": json.dumps(metadata, ensure_ascii=False),
+                    },
+                )
+                updated.append(item)
+
+            session.commit()
+            result = dict(preview)
+            result["updated_count"] = len(updated)
+            result["updated_ids"] = [item["id"] for item in updated]
+            result["ran_at"] = now.isoformat(timespec="seconds")
+            logger.info(
+                "Action plan hygiene updated=%d by_source=%s",
+                result["updated_count"],
+                result.get("by_source"),
+            )
+            return result
+        except Exception:
+            session.rollback()
+            raise
+        finally:
+            session.close()
+
+
+def run_action_plan_hygiene(
+    *,
+    stale_hours: int = DEFAULT_STALE_HOURS,
+    max_updates: int = DEFAULT_MAX_UPDATES,
+) -> Dict[str, Any]:
+    return ActionPlanHygieneService(stale_hours=stale_hours, max_updates=max_updates).run()
+
+
+__all__ = [
+    "ActionPlanHygieneService",
+    "build_action_plan_hygiene_preview",
+    "run_action_plan_hygiene",
+]

services/elephant_alpha_autonomous_engine.py

@@ -52,6 +52,7 @@ RESOURCE_LOAD_THRESHOLD_PCT = float(os.getenv("ELEPHANT_ALPHA_RESOURCE_LOAD_THRE
 RESOURCE_HIGH_PRIORITY_THRESHOLD = int(os.getenv("ELEPHANT_ALPHA_RESOURCE_HIGH_PRIORITY_THRESHOLD", "5"))
 RESOURCE_STALE_THRESHOLD = int(os.getenv("ELEPHANT_ALPHA_RESOURCE_STALE_THRESHOLD", "5"))
 RESOURCE_STALE_HOURS = int(os.getenv("ELEPHANT_ALPHA_RESOURCE_STALE_HOURS", "24"))
+RESOURCE_HYGIENE_ENABLED = os.getenv("ELEPHANT_ALPHA_RESOURCE_HYGIENE_ENABLED", "true").lower() in {"1", "true", "yes", "on"}
 
 # ---- Constants ----
 _ALLOWED_ACTION_TYPES = frozenset({
@@ -870,12 +871,21 @@ class ElephantAlphaAutonomousEngine:
         else:
             metrics = self._classify_resource_pressure(metrics)
 
+        pre_hygiene_metrics = dict(metrics)
+        hygiene_result = None
+        if metrics.get("should_alert") and RESOURCE_HYGIENE_ENABLED:
+            hygiene_result = self._run_action_plan_hygiene()
+            if hygiene_result and int(hygiene_result.get("updated_count") or 0) > 0:
+                metrics = self._collect_resource_pressure_metrics()
+                metrics["pre_hygiene"] = pre_hygiene_metrics
+                metrics["hygiene_result"] = hygiene_result
+
         trigger.conditions = dict(trigger.conditions or {})
         trigger.conditions["_resource_metrics"] = metrics
         trigger.conditions["resource_pressure_level"] = metrics.get("pressure_level")
         trigger.conditions["resource_evidence"] = metrics.get("evidence", [])
 
-        if not metrics.get("should_alert"):
+        if not metrics.get("should_alert") and not hygiene_result:
             self._store_escalation(trigger.trigger_type)
             self._record_resource_optimization_suppressed(trigger, metrics, "below_actionable_threshold")
             return
@@ -897,6 +907,15 @@ class ElephantAlphaAutonomousEngine:
         self._store_escalation(trigger.trigger_type)
         self._circuit_reset()
 
+    def _run_action_plan_hygiene(self) -> Optional[Dict[str, Any]]:
+        try:
+            from services.action_plan_hygiene import run_action_plan_hygiene
+
+            return run_action_plan_hygiene()
+        except Exception as exc:
+            self._log.error("Action plan hygiene failed (non-blocking): %s", exc)
+            return {"updated_count": 0, "error": f"{type(exc).__name__}: {str(exc)[:200]}"}
+
     def _record_resource_pressure_insight(
         self,
         metrics: Dict[str, Any],
@@ -996,7 +1015,9 @@ class ElephantAlphaAutonomousEngine:
             f"pending_review={metrics.get('human_review_count', 0)}，"
             f"stale={metrics.get('stale_count', 0)}，"
             f"system_load={float(metrics.get('system_load_pct') or 0):.1f}%。"
-            f"{load_text}；autonomous_limit={previous_limit}->{new_limit}。"
+            f"{load_text}；"
+            f"auto_closed={int((metrics.get('hygiene_result') or {}).get('updated_count') or 0)}；"
+            f"autonomous_limit={previous_limit}->{new_limit}。"
         )
 
     @staticmethod
@@ -1030,6 +1051,11 @@ class ElephantAlphaAutonomousEngine:
             "normal": "P4 normal",
         }.get(level, level)
         load_pct = float(metrics.get("system_load_pct") or 0.0)
+        pre_hygiene = metrics.get("pre_hygiene") if isinstance(metrics.get("pre_hygiene"), dict) else None
+        hygiene = metrics.get("hygiene_result") if isinstance(metrics.get("hygiene_result"), dict) else None
+        hygiene_count = int((hygiene or {}).get("updated_count") or 0)
+        if hygiene_count > 0 and not metrics.get("should_alert"):
+            level_label = "P4 resolved"
         load_judgement = (
             "主機 CPU 已達高負載門檻。"
             if metrics.get("load_pressure")
@@ -1042,6 +1068,13 @@ class ElephantAlphaAutonomousEngine:
         ]
         if new_limit != previous_limit:
             executed.append(f"已將 ElephantAlpha 自主決策上限由 {previous_limit} 調整為 {new_limit} 次/小時。")
+        if hygiene_count > 0:
+            by_source = hygiene.get("by_source") or {}
+            source_text = "、".join(f"{key} {value}" for key, value in by_source.items()) or f"{hygiene_count} 筆"
+            executed.insert(
+                1,
+                f"已自動關閉過期 action_plans {hygiene_count} 筆（{source_text}）；只改 status/metadata，不刪除資料。",
+            )
         executed.append("未執行外部修復、未啟動 Hermes/NemoTron 價格分析、未宣稱效益預測。")
 
         lines = [
@@ -1051,6 +1084,16 @@ class ElephantAlphaAutonomousEngine:
             f"時間：{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}",
             "",
             "<b>量測指標</b>",
+        ]
+        if pre_hygiene:
+            lines.append(
+                "• 清理前 Action queue："
+                f"{int(pre_hygiene.get('action_queue_size') or 0)}，"
+                f"P1/P2：{int(pre_hygiene.get('high_priority_count') or 0)}，"
+                f"逾時：{int(pre_hygiene.get('stale_count') or 0)}"
+            )
+            lines.append("<b>清理後</b>")
+        lines += [
             f"• Action queue：{int(metrics.get('action_queue_size') or 0)} / {int(metrics.get('queue_threshold') or RESOURCE_QUEUE_THRESHOLD)}",
             f"• P1/P2 待處理：{int(metrics.get('high_priority_count') or 0)} / {int(metrics.get('high_priority_threshold') or RESOURCE_HIGH_PRIORITY_THRESHOLD)}",
             f"• Pending review：{int(metrics.get('human_review_count') or 0)}",