feat: implement approval state machine (Task 13)

docs/hermes-implementation-plan.md

@@ -180,8 +180,18 @@
 
 ### Task 13: Implement approval state machine
 
+**Objective:** Implement approval state machine and fatigue prevention for action proposals.
+
 **Files:**
 - Create: `src/agentic_node_ops/approval.py`
+- Modify: `src/agentic_node_ops/database.py`
+- Create: `tests/test_approval.py`
+
+[x] Complete — Implemented approval state machine:
+  - `database.py`: Added `insert_action_proposal`, `get_last_proposal`, `count_timeouts`, `mark_action_suppressed`, `get_pending_proposals`, and `update_proposal_outcome`.
+  - `approval.py`: Implemented `should_propose_action` (cooldown/skip checks), `check_timeout_escalation`, `group_pending_proposals`, `propose_action`, and `resolve_proposal`.
+  - `test_approval.py`: 12 tests covering fatigue rules, escalation, and proposal grouping.
+  - All tests passing (148 total, 90% coverage).
 
 ---
 

src/agentic_node_ops/approval.py

@@ -0,0 +1,179 @@
+"""Approval state machine and fatigue prevention for action proposals."""
+
+from __future__ import annotations
+
+import logging
+from dataclasses import dataclass
+from datetime import datetime, timedelta
+from typing import Callable, Optional
+
+from .database import Database
+
+log = logging.getLogger(__name__)
+
+# Cooldowns: don't re-propose the same action for the same incident
+# within this window after a prior proposal of any outcome.
+REPROPOSAL_COOLDOWN: dict[str, timedelta] = {
+    "low": timedelta(hours=4),
+    "medium": timedelta(hours=2),
+    "high": timedelta(hours=1),
+    "critical": timedelta(minutes=30),
+}
+
+
+@dataclass
+class ProposalGroup:
+    """A group of pending proposals for bundled approval."""
+
+    proposals: list[dict]
+    grouped: bool = False
+
+
+def should_propose_action(action_id: str, incident_id: str, db: Database) -> bool:
+    """
+    Check if an action should be proposed based on fatigue prevention rules.
+
+    Returns False if:
+    - The action was previously skipped or suppressed for this incident.
+    - The cooldown period has not elapsed since the last proposal.
+    """
+    last = db.get_last_proposal(action_id=action_id, incident_id=incident_id)
+    if not last:
+        return True
+
+    # Never re-propose a SKIPPED or SUPPRESSED action in the same incident
+    if last.get("outcome") in ("skipped", "suppressed"):
+        log.debug(
+            "Action %s %s for incident %s, will not re-propose",
+            last.get("outcome"),
+            action_id,
+            incident_id,
+        )
+        return False
+
+    # Don't re-propose within cooldown
+    elapsed = datetime.now() - datetime.fromisoformat(last["proposed_at"])
+    cooldown = REPROPOSAL_COOLDOWN.get(last["severity"], timedelta(hours=1))
+
+    if elapsed < cooldown:
+        log.debug(
+            "Action %s for incident %s is within cooldown (%s < %s)",
+            action_id,
+            incident_id,
+            elapsed,
+            cooldown,
+        )
+        return False
+
+    return True
+
+
+def check_timeout_escalation(
+    action_id: str,
+    incident_id: str,
+    db: Database,
+    notify_callback: Optional[Callable[..., None]] = None,
+) -> None:
+    """
+    Check if an action has timed out too many times and escalate.
+
+    If the operator has not responded to 2 consecutive proposals for the same
+    action in the same incident, stop proposing and escalate.
+    """
+    timeouts = db.count_timeouts(action_id=action_id, incident_id=incident_id)
+    if timeouts >= 2:
+        log.warning(
+            "Action %s has timed out %d times for incident %s. Escalating.",
+            action_id,
+            timeouts,
+            incident_id,
+        )
+        db.mark_action_suppressed(action_id=action_id, incident_id=incident_id)
+
+        if notify_callback:
+            notify_callback(
+                f"Action '{action_id}' has timed out {timeouts} times with no operator response. "
+                f"No further proposals will be sent for this incident. Manual intervention required."
+            )
+
+
+def group_pending_proposals(
+    host: str, db: Database, within_minutes: int = 5
+) -> list[ProposalGroup]:
+    """
+    Group pending proposals for a host within a time window.
+
+    If multiple suggested_actions from different runbooks are ready to be
+    proposed within the window for the same host, bundle them into a single
+    approval message rather than sending individually.
+    """
+    within_seconds = within_minutes * 60
+    pending = db.get_pending_proposals(host=host, within_seconds=within_seconds)
+
+    if len(pending) <= 1:
+        return [ProposalGroup(proposals=pending)]
+
+    return [ProposalGroup(proposals=pending, grouped=True)]
+
+
+def propose_action(
+    action_id: str,
+    incident_id: str,
+    severity: str,
+    db: Database,
+) -> Optional[str]:
+    """
+    Propose an action if fatigue rules allow it.
+
+    Returns the new proposal ID if proposed, or None if suppressed by fatigue rules.
+    """
+    if not should_propose_action(action_id, incident_id, db):
+        return None
+
+    proposed_at = datetime.now().isoformat()
+    proposal_id = db.insert_action_proposal(
+        incident_id=incident_id,
+        action_id=action_id,
+        severity=severity,
+        proposed_at=proposed_at,
+    )
+    log.info(
+        "Proposed action %s for incident %s (ID: %s)",
+        action_id,
+        incident_id,
+        proposal_id,
+    )
+    return proposal_id
+
+
+def resolve_proposal(
+    proposal_id: str,
+    outcome: str,
+    db: Database,
+    resolved_at: Optional[str] = None,
+) -> None:
+    """
+    Resolve a proposal with a final outcome.
+
+    Valid outcomes: 'approved', 'skipped', 'timeout', 'suppressed', 'success', 'failed'
+    """
+    valid_outcomes = {
+        "approved",
+        "skipped",
+        "timeout",
+        "suppressed",
+        "success",
+        "failed",
+    }
+    if outcome not in valid_outcomes:
+        raise ValueError(
+            f"Invalid outcome '{outcome}'. Must be one of {valid_outcomes}"
+        )
+
+    if resolved_at is None:
+        resolved_at = datetime.now().isoformat()
+
+    db.update_proposal_outcome(
+        proposal_id=proposal_id, outcome=outcome, resolved_at=resolved_at
+    )
+    log.info("Resolved proposal %s with outcome: %s", proposal_id, outcome)

src/agentic_node_ops/database.py

@@ -10,6 +10,7 @@
 import logging
 import os
 import sqlite3
+import uuid
 from contextlib import contextmanager
 from pathlib import Path
 from typing import Optional
@@ -206,8 +207,6 @@ def insert_correction(
         self, incident_id: str, alert_type: str, host: str, correction: str
     ) -> None:
         """Insert an operator correction."""
-        import uuid
-
         with self._get_connection() as conn:
             conn.execute(
                 """
@@ -306,3 +305,103 @@ def upsert_host_baseline(
                 (host, metric, p50, p95),
             )
             conn.commit()
+
+    def insert_action_proposal(
+        self,
+        incident_id: str,
+        action_id: str,
+        severity: str,
+        proposed_at: str,
+    ) -> str:
+        """Insert a new action proposal and return its ID."""
+        proposal_id = str(uuid.uuid4())
+        with self._get_connection() as conn:
+            conn.execute(
+                """
+                INSERT INTO action_proposals (id, incident_id, action_id, severity, proposed_at)
+                VALUES (?, ?, ?, ?, ?)
+                """,
+                (proposal_id, incident_id, action_id, severity, proposed_at),
+            )
+            conn.commit()
+        return proposal_id
+
+    def get_last_proposal(self, action_id: str, incident_id: str) -> Optional[dict]:
+        """Get the most recent proposal for a specific action and incident."""
+        with self._get_connection() as conn:
+            cursor = conn.execute(
+                """
+                SELECT id, action_id, severity, proposed_at, outcome, resolved_at
+                FROM action_proposals
+                WHERE action_id = ? AND incident_id = ?
+                ORDER BY proposed_at DESC
+                LIMIT 1
+                """,
+                (action_id, incident_id),
+            )
+            row = cursor.fetchone()
+            return dict(row) if row else None
+
+    def count_timeouts(self, action_id: str, incident_id: str) -> int:
+        """Count the number of timeout outcomes for a specific action and incident."""
+        with self._get_connection() as conn:
+            cursor = conn.execute(
+                """
+                SELECT COUNT(*) as count
+                FROM action_proposals
+                WHERE action_id = ? AND incident_id = ? AND outcome = 'timeout'
+                """,
+                (action_id, incident_id),
+            )
+            row = cursor.fetchone()
+            return row["count"] if row else 0
+
+    def mark_action_suppressed(self, action_id: str, incident_id: str) -> None:
+        """Mark the most recent action proposal as suppressed for a specific incident."""
+        with self._get_connection() as conn:
+            conn.execute(
+                """
+                UPDATE action_proposals
+                SET outcome = 'suppressed'
+                WHERE id = (
+                    SELECT id FROM action_proposals
+                    WHERE action_id = ? AND incident_id = ?
+                    ORDER BY proposed_at DESC
+                    LIMIT 1
+                )
+                """,
+                (action_id, incident_id),
+            )
+            conn.commit()
+
+    def get_pending_proposals(self, host: str, within_seconds: int = 300) -> list[dict]:
+        """Get pending (outcome IS NULL) proposals for a host within a time window."""
+        with self._get_connection() as conn:
+            cursor = conn.execute(
+                """
+                SELECT ap.id, ap.incident_id, ap.action_id, ap.severity, ap.proposed_at, i.host
+                FROM action_proposals ap
+                JOIN incidents i ON ap.incident_id = i.id
+                WHERE i.host = ?
+                  AND ap.outcome IS NULL
+                  AND ap.proposed_at >= datetime('now', ?)
+                ORDER BY ap.proposed_at ASC
+                """,
+                (host, f"-{within_seconds} seconds"),
+            )
+            return [dict(row) for row in cursor.fetchall()]
+
+    def update_proposal_outcome(
+        self, proposal_id: str, outcome: str, resolved_at: Optional[str] = None
+    ) -> None:
+        """Update the outcome of an action proposal."""
+        with self._get_connection() as conn:
+            conn.execute(
+                """
+                UPDATE action_proposals
+                SET outcome = ?, resolved_at = COALESCE(?, resolved_at)
+                WHERE id = ?
+                """,
+                (outcome, resolved_at, proposal_id),
+            )
+            conn.commit()