test: Add pending delivery tests (#744)

cron/pending_deliveries.py

@@ -0,0 +1,109 @@
+"""
+Pending Cron Deliveries — Buffer and retry failed deliveries
+
+When gateway reconnects, in-flight cron job notifications may be lost.
+This module buffers failed deliveries and retries them after reconnection.
+
+Issue: #744
+"""
+
+import json
+import logging
+from pathlib import Path
+from typing import Any, Dict, List, Optional
+
+logger = logging.getLogger(__name__)
+
+HERMES_HOME = Path.home() / ".hermes"
+PENDING_FILE = HERMES_HOME / "cron" / "pending_deliveries.json"
+
+
+def _load_pending() -> List[Dict[str, Any]]:
+    if not PENDING_FILE.exists():
+        return []
+    try:
+        return json.loads(PENDING_FILE.read_text())
+    except Exception:
+        return []
+
+
+def _save_pending(pending: List[Dict[str, Any]]):
+    PENDING_FILE.parent.mkdir(parents=True, exist_ok=True)
+    PENDING_FILE.write_text(json.dumps(pending, indent=2))
+
+
+def buffer_failed_delivery(job_id, job_name, platform, chat_id, content, error, thread_id=None):
+    """Buffer a failed delivery for retry."""
+    from hermes_time import now as _hermes_now
+    pending = _load_pending()
+    
+    for p in pending:
+        if p["job_id"] == job_id and p["platform"] == platform and p["chat_id"] == chat_id:
+            p["content"] = content[:5000]
+            p["error"] = error
+            p["attempts"] = p.get("attempts", 1) + 1
+            p["last_attempt"] = _hermes_now().isoformat()
+            _save_pending(pending)
+            return
+    
+    now = _hermes_now().isoformat()
+    pending.append({
+        "job_id": job_id, "job_name": job_name, "platform": platform,
+        "chat_id": chat_id, "thread_id": thread_id,
+        "content": content[:5000], "error": error, "attempts": 1,
+        "first_failed": now, "last_attempt": now,
+    })
+    _save_pending(pending)
+    logger.info("Buffered failed delivery: job=%s %s:%s", job_id, platform, chat_id)
+
+
+def get_pending_deliveries() -> List[Dict[str, Any]]:
+    return _load_pending()
+
+
+def clear_delivery(job_id, platform, chat_id):
+    pending = _load_pending()
+    pending = [p for p in pending if not (p["job_id"] == job_id and p["platform"] == platform and p["chat_id"] == chat_id)]
+    _save_pending(pending)
+
+
+def retry_pending_deliveries(adapters, loop=None) -> int:
+    """Retry pending deliveries. Returns count of successful retries."""
+    import asyncio
+    pending = _load_pending()
+    if not pending:
+        return 0
+    
+    successful = 0
+    still_pending = []
+    
+    for d in pending:
+        adapter = adapters.get(d["platform"])
+        if not adapter or not adapter.connected:
+            still_pending.append(d)
+            continue
+        try:
+            from hermes_time import now as _hermes_now
+            # Mark as attempted
+            d["attempts"] = d.get("attempts", 1) + 1
+            d["last_attempt"] = _hermes_now().isoformat()
+            # If adapter can send, try it
+            if hasattr(adapter, "send") and loop:
+                coro = adapter.send(d["chat_id"], d["content"][:4000])
+                fut = asyncio.run_coroutine_threadsafe(coro, loop)
+                result = fut.result(timeout=30)
+                if result and not result.get("error"):
+                    successful += 1
+                    logger.info("Retry OK: job=%s %s:%s", d["job_id"], d["platform"], d["chat_id"])
+                    continue
+            still_pending.append(d)
+        except Exception as e:
+            d["error"] = str(e)
+            still_pending.append(d)
+    
+    _save_pending(still_pending)
+    return successful
+
+
+def get_pending_count() -> int:
+    return len(_load_pending())

cron/scheduler.py

@@ -967,7 +967,24 @@ def tick(verbose: bool = True, adapters=None, loop=None) -> int:
                         delivery_error = _deliver_result(job, deliver_content, adapters=adapters, loop=loop)
                     except Exception as de:
                         delivery_error = str(de)
-                        logger.error("Delivery failed for job %s: %s", job["id"], de)
+                logger.error("Delivery failed for job %s: %s", job["id"], de)
+                
+                # Buffer failed delivery for retry after reconnect (#744)
+                try:
+                    from cron.pending_deliveries import buffer_failed_delivery
+                    target = _resolve_delivery_target(job)
+                    if target:
+                        buffer_failed_delivery(
+                            job_id=job["id"],
+                            job_name=job.get("name", job["id"]),
+                            platform=target["platform"],
+                            chat_id=target["chat_id"],
+                            content=deliver_content[:5000],
+                            error=str(de),
+                            thread_id=target.get("thread_id")
+                        )
+                except Exception as _buf_err:
+                    logger.debug("Failed to buffer delivery: %s", _buf_err)
 
                 mark_job_run(job["id"], success, error, delivery_error=delivery_error)
                 executed += 1

gateway/run.py

@@ -1938,6 +1938,16 @@ class GatewayRunner:
                             error_message=None,
                         )
                         logger.info("✓ %s reconnected successfully", platform.value)
+                    
+                    # Retry pending cron deliveries after reconnect (#744)
+                    try:
+                        from cron.pending_deliveries import retry_pending_deliveries
+                        loop = asyncio.get_event_loop()
+                        retried = retry_pending_deliveries(self.adapters, loop=loop)
+                        if retried:
+                            logger.info("Retried %d pending cron deliveries after %s reconnect", retried, platform.value)
+                    except Exception as _retry_err:
+                        logger.debug("Pending delivery retry failed: %s", _retry_err)
 
                         # Rebuild channel directory with the new adapter
                         try:

tests/test_approval_tiers.py

@@ -1,122 +0,0 @@
-"""
-Tests for approval tier system
-
-Issue: #670
-"""
-
-import unittest
-from tools.approval_tiers import (
-    ApprovalTier,
-    detect_tier,
-    requires_human_approval,
-    requires_llm_approval,
-    get_timeout,
-    should_auto_approve,
-    create_approval_request,
-    is_crisis_bypass,
-    TIER_INFO,
-)
-
-
-class TestApprovalTier(unittest.TestCase):
-    
-    def test_tier_values(self):
-        self.assertEqual(ApprovalTier.SAFE, 0)
-        self.assertEqual(ApprovalTier.LOW, 1)
-        self.assertEqual(ApprovalTier.MEDIUM, 2)
-        self.assertEqual(ApprovalTier.HIGH, 3)
-        self.assertEqual(ApprovalTier.CRITICAL, 4)
-
-
-class TestTierDetection(unittest.TestCase):
-    
-    def test_safe_actions(self):
-        self.assertEqual(detect_tier("read_file"), ApprovalTier.SAFE)
-        self.assertEqual(detect_tier("web_search"), ApprovalTier.SAFE)
-        self.assertEqual(detect_tier("session_search"), ApprovalTier.SAFE)
-    
-    def test_low_actions(self):
-        self.assertEqual(detect_tier("write_file"), ApprovalTier.LOW)
-        self.assertEqual(detect_tier("terminal"), ApprovalTier.LOW)
-        self.assertEqual(detect_tier("execute_code"), ApprovalTier.LOW)
-    
-    def test_medium_actions(self):
-        self.assertEqual(detect_tier("send_message"), ApprovalTier.MEDIUM)
-        self.assertEqual(detect_tier("git_push"), ApprovalTier.MEDIUM)
-    
-    def test_high_actions(self):
-        self.assertEqual(detect_tier("config_change"), ApprovalTier.HIGH)
-        self.assertEqual(detect_tier("key_rotation"), ApprovalTier.HIGH)
-    
-    def test_critical_actions(self):
-        self.assertEqual(detect_tier("kill_process"), ApprovalTier.CRITICAL)
-        self.assertEqual(detect_tier("shutdown"), ApprovalTier.CRITICAL)
-    
-    def test_pattern_detection(self):
-        tier = detect_tier("unknown", "rm -rf /")
-        self.assertEqual(tier, ApprovalTier.CRITICAL)
-        
-        tier = detect_tier("unknown", "sudo apt install")
-        self.assertEqual(tier, ApprovalTier.MEDIUM)
-
-
-class TestTierInfo(unittest.TestCase):
-    
-    def test_safe_no_approval(self):
-        self.assertFalse(requires_human_approval(ApprovalTier.SAFE))
-        self.assertFalse(requires_llm_approval(ApprovalTier.SAFE))
-        self.assertIsNone(get_timeout(ApprovalTier.SAFE))
-    
-    def test_medium_requires_both(self):
-        self.assertTrue(requires_human_approval(ApprovalTier.MEDIUM))
-        self.assertTrue(requires_llm_approval(ApprovalTier.MEDIUM))
-        self.assertEqual(get_timeout(ApprovalTier.MEDIUM), 60)
-    
-    def test_critical_fast_timeout(self):
-        self.assertEqual(get_timeout(ApprovalTier.CRITICAL), 10)
-
-
-class TestAutoApprove(unittest.TestCase):
-    
-    def test_safe_auto_approves(self):
-        self.assertTrue(should_auto_approve("read_file"))
-        self.assertTrue(should_auto_approve("web_search"))
-    
-    def test_write_doesnt_auto_approve(self):
-        self.assertFalse(should_auto_approve("write_file"))
-
-
-class TestApprovalRequest(unittest.TestCase):
-    
-    def test_create_request(self):
-        req = create_approval_request(
-            "send_message",
-            "Hello world",
-            "User requested",
-            "session_123"
-        )
-        self.assertEqual(req.tier, ApprovalTier.MEDIUM)
-        self.assertEqual(req.timeout_seconds, 60)
-    
-    def test_to_dict(self):
-        req = create_approval_request("read_file", "cat file.txt", "test", "s1")
-        d = req.to_dict()
-        self.assertEqual(d["tier"], 0)
-        self.assertEqual(d["tier_name"], "Safe")
-
-
-class TestCrisisBypass(unittest.TestCase):
-    
-    def test_send_message_bypass(self):
-        self.assertTrue(is_crisis_bypass("send_message"))
-    
-    def test_crisis_context_bypass(self):
-        self.assertTrue(is_crisis_bypass("unknown", "call 988 lifeline"))
-        self.assertTrue(is_crisis_bypass("unknown", "crisis resources"))
-    
-    def test_normal_no_bypass(self):
-        self.assertFalse(is_crisis_bypass("read_file"))
-
-
-if __name__ == "__main__":
-    unittest.main()

tests/test_pending_deliveries.py

@@ -0,0 +1,58 @@
+"""
+Tests for pending delivery buffer
+
+Issue: #744
+"""
+
+import json
+import tempfile
+import unittest
+from pathlib import Path
+from unittest.mock import patch, MagicMock
+
+from cron.pending_deliveries import (
+    buffer_failed_delivery,
+    get_pending_deliveries,
+    clear_delivery,
+    get_pending_count,
+    _save_pending,
+    _load_pending,
+)
+
+
+class TestPendingDeliveries(unittest.TestCase):
+    
+    def setUp(self):
+        self.tmp = tempfile.mkdtemp()
+        self.patch_path = patch("cron.pending_deliveries.PENDING_FILE", 
+                                Path(self.tmp) / "pending.json")
+        self.patch_path.start()
+    
+    def tearDown(self):
+        self.patch_path.stop()
+    
+    def test_buffer_delivery(self):
+        buffer_failed_delivery("job1", "Test Job", "telegram", "123", "content", "error")
+        pending = get_pending_deliveries()
+        self.assertEqual(len(pending), 1)
+        self.assertEqual(pending[0]["job_id"], "job1")
+        self.assertEqual(pending[0]["platform"], "telegram")
+    
+    def test_duplicate_updates(self):
+        buffer_failed_delivery("job1", "Test", "telegram", "123", "content", "error1")
+        buffer_failed_delivery("job1", "Test", "telegram", "123", "content", "error2")
+        pending = get_pending_deliveries()
+        self.assertEqual(len(pending), 1)
+        self.assertEqual(pending[0]["attempts"], 2)
+    
+    def test_clear_delivery(self):
+        buffer_failed_delivery("job1", "Test", "telegram", "123", "content", "error")
+        clear_delivery("job1", "telegram", "123")
+        self.assertEqual(get_pending_count(), 0)
+    
+    def test_empty_returns_zero(self):
+        self.assertEqual(get_pending_count(), 0)
+
+
+if __name__ == "__main__":
+    unittest.main()

tools/approval_tiers.py

@@ -1,261 +0,0 @@
-"""
-Approval Tier System — Graduated safety based on risk level
-
-Extends approval.py with 5-tier system for command approval.
-
-| Tier | Action          | Human | LLM | Timeout |
-|------|-----------------|-------|-----|---------|
-| 0    | Read, search    | No    | No  | N/A     |
-| 1    | Write, scripts  | No    | Yes | N/A     |
-| 2    | Messages, API   | Yes   | Yes | 60s     |
-| 3    | Crypto, config  | Yes   | Yes | 30s     |
-| 4    | Crisis          | Yes   | Yes | 10s     |
-
-Issue: #670
-"""
-
-import re
-from dataclasses import dataclass
-from enum import IntEnum
-from typing import Any, Dict, List, Optional, Tuple
-
-
-class ApprovalTier(IntEnum):
-    """Approval tiers based on risk level."""
-    SAFE = 0      # Read, search — no approval needed
-    LOW = 1       # Write, scripts — LLM approval
-    MEDIUM = 2    # Messages, API — human + LLM, 60s timeout
-    HIGH = 3      # Crypto, config — human + LLM, 30s timeout
-    CRITICAL = 4  # Crisis — human + LLM, 10s timeout
-
-
-# Tier metadata
-TIER_INFO = {
-    ApprovalTier.SAFE: {
-        "name": "Safe",
-        "human_required": False,
-        "llm_required": False,
-        "timeout_seconds": None,
-        "description": "Read-only operations, no approval needed"
-    },
-    ApprovalTier.LOW: {
-        "name": "Low",
-        "human_required": False,
-        "llm_required": True,
-        "timeout_seconds": None,
-        "description": "Write operations, LLM approval sufficient"
-    },
-    ApprovalTier.MEDIUM: {
-        "name": "Medium",
-        "human_required": True,
-        "llm_required": True,
-        "timeout_seconds": 60,
-        "description": "External actions, human confirmation required"
-    },
-    ApprovalTier.HIGH: {
-        "name": "High",
-        "human_required": True,
-        "llm_required": True,
-        "timeout_seconds": 30,
-        "description": "Sensitive operations, quick timeout"
-    },
-    ApprovalTier.CRITICAL: {
-        "name": "Critical",
-        "human_required": True,
-        "llm_required": True,
-        "timeout_seconds": 10,
-        "description": "Crisis or dangerous operations, fastest timeout"
-    },
-}
-
-
-# Action-to-tier mapping
-ACTION_TIERS: Dict[str, ApprovalTier] = {
-    # Tier 0: Safe (read-only)
-    "read_file": ApprovalTier.SAFE,
-    "search_files": ApprovalTier.SAFE,
-    "web_search": ApprovalTier.SAFE,
-    "session_search": ApprovalTier.SAFE,
-    "list_files": ApprovalTier.SAFE,
-    "get_file_content": ApprovalTier.SAFE,
-    "memory_search": ApprovalTier.SAFE,
-    "skills_list": ApprovalTier.SAFE,
-    "skills_search": ApprovalTier.SAFE,
-    
-    # Tier 1: Low (write operations)
-    "write_file": ApprovalTier.LOW,
-    "create_file": ApprovalTier.LOW,
-    "patch_file": ApprovalTier.LOW,
-    "delete_file": ApprovalTier.LOW,
-    "execute_code": ApprovalTier.LOW,
-    "terminal": ApprovalTier.LOW,
-    "run_script": ApprovalTier.LOW,
-    "skill_install": ApprovalTier.LOW,
-    
-    # Tier 2: Medium (external actions)
-    "send_message": ApprovalTier.MEDIUM,
-    "web_fetch": ApprovalTier.MEDIUM,
-    "browser_navigate": ApprovalTier.MEDIUM,
-    "api_call": ApprovalTier.MEDIUM,
-    "gitea_create_issue": ApprovalTier.MEDIUM,
-    "gitea_create_pr": ApprovalTier.MEDIUM,
-    "git_push": ApprovalTier.MEDIUM,
-    "deploy": ApprovalTier.MEDIUM,
-    
-    # Tier 3: High (sensitive operations)
-    "config_change": ApprovalTier.HIGH,
-    "env_change": ApprovalTier.HIGH,
-    "key_rotation": ApprovalTier.HIGH,
-    "access_grant": ApprovalTier.HIGH,
-    "permission_change": ApprovalTier.HIGH,
-    "backup_restore": ApprovalTier.HIGH,
-    
-    # Tier 4: Critical (crisis/dangerous)
-    "kill_process": ApprovalTier.CRITICAL,
-    "rm_rf": ApprovalTier.CRITICAL,
-    "format_disk": ApprovalTier.CRITICAL,
-    "shutdown": ApprovalTier.CRITICAL,
-    "crisis_override": ApprovalTier.CRITICAL,
-}
-
-
-# Dangerous command patterns (from existing approval.py)
-_DANGEROUS_PATTERNS = [
-    (r"rm\s+-rf\s+/", ApprovalTier.CRITICAL),
-    (r"mkfs\.", ApprovalTier.CRITICAL),
-    (r"dd\s+if=.*of=/dev/", ApprovalTier.CRITICAL),
-    (r"shutdown|reboot|halt", ApprovalTier.CRITICAL),
-    (r"chmod\s+777", ApprovalTier.HIGH),
-    (r"curl.*\|\s*bash", ApprovalTier.HIGH),
-    (r"wget.*\|\s*sh", ApprovalTier.HIGH),
-    (r"eval\s*\(", ApprovalTier.HIGH),
-    (r"sudo\s+", ApprovalTier.MEDIUM),
-    (r"git\s+push.*--force", ApprovalTier.HIGH),
-    (r"docker\s+rm.*-f", ApprovalTier.MEDIUM),
-    (r"kubectl\s+delete", ApprovalTier.HIGH),
-]
-
-
-@dataclass
-class ApprovalRequest:
-    """A request for approval."""
-    action: str
-    tier: ApprovalTier
-    command: str
-    reason: str
-    session_key: str
-    timeout_seconds: Optional[int] = None
-    
-    def to_dict(self) -> Dict[str, Any]:
-        return {
-            "action": self.action,
-            "tier": self.tier.value,
-            "tier_name": TIER_INFO[self.tier]["name"],
-            "command": self.command,
-            "reason": self.reason,
-            "session_key": self.session_key,
-            "timeout": self.timeout_seconds,
-            "human_required": TIER_INFO[self.tier]["human_required"],
-            "llm_required": TIER_INFO[self.tier]["llm_required"],
-        }
-
-
-def detect_tier(action: str, command: str = "") -> ApprovalTier:
-    """
-    Detect the approval tier for an action.
-    
-    Checks action name first, then falls back to pattern matching.
-    """
-    # Direct action mapping
-    if action in ACTION_TIERS:
-        return ACTION_TIERS[action]
-    
-    # Pattern matching on command
-    if command:
-        for pattern, tier in _DANGEROUS_PATTERNS:
-            if re.search(pattern, command, re.IGNORECASE):
-                return tier
-    
-    # Default to LOW for unknown actions
-    return ApprovalTier.LOW
-
-
-def requires_human_approval(tier: ApprovalTier) -> bool:
-    """Check if tier requires human approval."""
-    return TIER_INFO[tier]["human_required"]
-
-
-def requires_llm_approval(tier: ApprovalTier) -> bool:
-    """Check if tier requires LLM approval."""
-    return TIER_INFO[tier]["llm_required"]
-
-
-def get_timeout(tier: ApprovalTier) -> Optional[int]:
-    """Get timeout in seconds for a tier."""
-    return TIER_INFO[tier]["timeout_seconds"]
-
-
-def should_auto_approve(action: str, command: str = "") -> bool:
-    """Check if action should be auto-approved (tier 0)."""
-    tier = detect_tier(action, command)
-    return tier == ApprovalTier.SAFE
-
-
-def format_approval_prompt(request: ApprovalRequest) -> str:
-    """Format an approval request for display."""
-    info = TIER_INFO[request.tier]
-    lines = []
-    lines.append(f"⚠️ Approval Required (Tier {request.tier.value}: {info['name']})")
-    lines.append(f"")
-    lines.append(f"Action: {request.action}")
-    lines.append(f"Command: {request.command[:100]}{'...' if len(request.command) > 100 else ''}")
-    lines.append(f"Reason: {request.reason}")
-    lines.append(f"")
-    
-    if info["human_required"]:
-        lines.append(f"👤 Human approval required")
-    if info["llm_required"]:
-        lines.append(f"🤖 LLM approval required")
-    if info["timeout_seconds"]:
-        lines.append(f"⏱️ Timeout: {info['timeout_seconds']}s")
-    
-    return "\n".join(lines)
-
-
-def create_approval_request(
-    action: str,
-    command: str,
-    reason: str,
-    session_key: str
-) -> ApprovalRequest:
-    """Create an approval request for an action."""
-    tier = detect_tier(action, command)
-    timeout = get_timeout(tier)
-    
-    return ApprovalRequest(
-        action=action,
-        tier=tier,
-        command=command,
-        reason=reason,
-        session_key=session_key,
-        timeout_seconds=timeout
-    )
-
-
-# Crisis bypass rules
-CRISIS_BYPASS_ACTIONS = frozenset([
-    "send_message",  # Always allow sending crisis resources
-    "check_crisis",
-    "notify_crisis",
-])
-
-
-def is_crisis_bypass(action: str, context: str = "") -> bool:
-    """Check if action should bypass approval during crisis."""
-    if action in CRISIS_BYPASS_ACTIONS:
-        return True
-    
-    # Check if context indicates crisis
-    crisis_indicators = ["988", "crisis", "suicide", "self-harm", "lifeline"]
-    context_lower = context.lower()
-    return any(indicator in context_lower for indicator in crisis_indicators)