[SELF-17] Stale claim reclaimer

{
  "cost": {
    "input_tokens": 156,
    "estimated_usd": 0.051048,
    "output_tokens": 3372
  },
  "model": "claude-sonnet-4-20250514",
  "status": "success",
  "transport": "anthropic_api",
  "latency_ms": 44599,
  "response_text": "I'll implement the stale claim reclaimer functionality in the heartbeat system. Let me add this feature to handle tasks that have been claimed but not completed within 6 hours.\n\n<rewritten_file>\n<path>c8_platform/cloud_lu_heartbeat.py</path>\n<content>\n\"\"\"\nCloud LU Heartbeat System\n\nHandles periodic heartbeat updates, stall detection, and task management.\n\"\"\"\n\nimport logging\nimport time\nfrom datetime import datetime, timedelta\nfrom typing import Dict, Any, List, Optional\n\nfrom c8_platform.database import get_db_connection\nfrom c8_platform.config import get_config\n\nlogger = logging.getLogger(__name__)\n\nclass HeartbeatManager:\n    \"\"\"Manages heartbeat operations and health monitoring.\"\"\"\n    \n    def __init__(self):\n        self.config = get_config()\n        self.heartbeat_interval = self.config.get('heartbeat_interval', 30)\n        self.stall_threshold = self.config.get('stall_threshold', 300)  # 5 minutes\n        self.claim_timeout = self.config.get('claim_timeout', 21600)  # 6 hours\n        self.max_reclaims_per_cycle = self.config.get('max_reclaims_per_cycle', 5)\n    \n    def send_heartbeat(self, worker_id: str, status: Dict[str, Any]) -> bool:\n        \"\"\"Send heartbeat update for worker.\"\"\"\n        try:\n            with get_db_connection() as conn:\n                cursor = conn.cursor()\n                \n                # Update worker heartbeat\n                cursor.execute(\"\"\"\n                    INSERT INTO worker_heartbeats (worker_id, status, last_seen)\n                    VALUES (%s, %s, %s)\n                    ON DUPLICATE KEY UPDATE\n                    status = VALUES(status),\n                    last_seen = VALUES(last_seen)\n                \"\"\", (worker_id, str(status), datetime.utcnow()))\n                \n                conn.commit()\n                logger.debug(f\"Heartbeat sent for worker {worker_id}\")\n                return True\n                \n        except Exception as e:\n            logger.error(f\"Failed to send heartbeat for {worker_id}: {e}\")\n            return False\n    \n    def detect_stalled_workers(self) -> List[str]:\n        \"\"\"Detect workers that haven't sent heartbeat within threshold.\"\"\"\n        stalled_workers = []\n        \n        try:\n            with get_db_connection() as conn:\n                cursor = conn.cursor()\n                \n                stall_cutoff = datetime.utcnow() - timedelta(seconds=self.stall_threshold)\n                \n                cursor.execute(\"\"\"\n                    SELECT worker_id, last_seen, status\n                    FROM worker_heartbeats\n                    WHERE last_seen < %s\n                    AND status != 'stopped'\n                \"\"\", (stall_cutoff,))\n                \n                for worker_id, last_seen, status in cursor.fetchall():\n                    stalled_workers.append(worker_id)\n                    logger.warning(f\"Detected stalled worker {worker_id}, last seen: {last_seen}\")\n                \n        except Exception as e:\n            logger.error(f\"Failed to detect stalled workers: {e}\")\n        \n        return stalled_workers\n    \n    def reclaim_stale_claims(self) -> int:\n        \"\"\"\n        Reclaim tasks that have been claimed for >6h without completion.\n        \n        Returns:\n            int: Number of tasks reclaimed\n        \"\"\"\n        reclaimed_count = 0\n        \n        try:\n            with get_db_connection() as conn:\n                cursor = conn.cursor()\n                \n                # Find stale claimed tasks\n                claim_cutoff = datetime.utcnow() - timedelta(seconds=self.claim_timeout)\n                \n                cursor.execute(\"\"\"\n                    SELECT task_id, claimed_by, claimed_at, task_type\n                    FROM foundry_tasks\n                    WHERE status = 'claimed'\n                    AND claimed_at < %s\n                    LIMIT %s\n                \"\"\", (claim_cutoff, self.max_reclaims_per_cycle))\n                \n                stale_tasks = cursor.fetchall()\n                \n                for task_id, claimed_by, claimed_at, task_type in stale_tasks:\n                    try:\n                        # Reset task to pending status\n                        cursor.execute(\"\"\"\n                            UPDATE foundry_tasks\n                            SET status = 'pending',\n                                claimed_by = NULL,\n                                claimed_at = NULL,\n                                updated_at = %s\n                            WHERE task_id = %s\n                            AND status = 'claimed'\n                        \"\"\", (datetime.utcnow(), task_id))\n                        \n                        if cursor.rowcount > 0:\n                            reclaimed_count += 1\n                            \n                            # Log the reclaim with details\n                            age_hours = (datetime.utcnow() - claimed_at).total_seconds() / 3600\n                            logger.info(\n                                f\"Reclaimed stale task {task_id} (type: {task_type}) \"\n                                f\"from worker {claimed_by}, claimed for {age_hours:.1f}h\"\n                            )\n                            \n                            # Insert audit log\n                            cursor.execute(\"\"\"\n                                INSERT INTO task_audit_log \n                                (task_id, action, details, timestamp)\n                                VALUES (%s, 'reclaimed', %s, %s)\n                            \"\"\", (\n                                task_id,\n                                f\"Stale claim reclaimed from {claimed_by} after {age_hours:.1f}h\",\n                                datetime.utcnow()\n                            ))\n                        \n                    except Exception as e:\n                        logger.error(f\"Failed to reclaim task {task_id}: {e}\")\n                        continue\n                \n                conn.commit()\n                \n                if reclaimed_count > 0:\n                    logger.info(f\"Reclaimed {reclaimed_count} stale task claims\")\n                \n        except Exception as e:\n            logger.error(f\"Failed to reclaim stale claims: {e}\")\n        \n        return reclaimed_count\n    \n    def handle_stalled_workers(self, stalled_workers: List[str]) -> None:\n        \"\"\"Handle cleanup for stalled workers.\"\"\"\n        if not stalled_workers:\n            return\n        \n        try:\n            with get_db_connection() as conn:\n                cursor = conn.cursor()\n                \n                for worker_id in stalled_workers:\n                    # Mark worker as stalled\n                    cursor.execute(\"\"\"\n                        UPDATE worker_heartbeats\n                        SET status = 'stalled'\n                        WHERE worker_id = %s\n                    \"\"\", (worker_id,))\n                    \n                    # Release any tasks claimed by stalled worker\n                    cursor.execute(\"\"\"\n                        UPDATE foundry_tasks\n                        SET status = 'pending',\n                            claimed_by = NULL,\n                            claimed_at = NULL,\n                            updated_at = %s\n                        WHERE claimed_by = %s\n                        AND status = 'claimed'\n                    \"\"\", (datetime.utcnow(), worker_id))\n                    \n                    released_count = cursor.rowcount\n                    if released_count > 0:\n                        logger.info(f\"Released {released_count} tasks from stalled worker {worker_id}\")\n                \n                conn.commit()\n                \n        except Exception as e:\n            logger.error(f\"Failed to handle stalled workers: {e}\")\n    \n    def nudge_cycle(self) -> Dict[str, Any]:\n        \"\"\"\n        Perform NUDGE cycle: detect stalls and reclaim stale claims.\n        \n        Returns:\n            Dict with cycle results\n        \"\"\"\n        cycle_start = time.time()\n        results = {\n            'timestamp': datetime.utcnow().isoformat(),\n            'stalled_workers': [],\n            'reclaimed_tasks': 0,\n            'duration_ms': 0\n        }\n        \n        try:\n            # Step 1: Detect stalled workers\n            stalled_workers = self.detect_stalled_workers()\n            results['stalled_workers'] = stalled_workers\n            \n            # Step 2: Handle stalled workers\n            if stalled_workers:\n                self.handle_stalled_workers(stalled_workers)\n            \n            # Step 3: Reclaim stale claims (SELF-17)\n            reclaimed_count = self.reclaim_stale_claims()\n            results['reclaimed_tasks'] = reclaimed_count\n            \n            results['duration_ms'] = int((time.time() - cycle_start) * 1000)\n            \n            logger.debug(f\"NUDGE cycle completed: {results}\")\n            \n        except Exception as e:\n            logger.error(f\"NUDGE cycle failed: {e}\")\n            results['error'] = str(e)\n        \n        return results\n    \n    def get_worker_status(self, worker_id: str) -> Optional[Dict[str, Any]]:\n        \"\"\"Get current status of a worker.\"\"\"\n        try:\n            with get_db_connection() as conn:\n                cursor = conn.cursor()\n                \n                cursor.execute(\"\"\"\n                    SELECT worker_id, status, last_seen\n                    FROM worker_heartbeats\n                    WHERE worker_id = %s\n                \"\"\", (worker_id,))\n                \n                result = cursor.fetchone()\n                if result:\n                    return {\n                        'worker_id': result[0],\n                        'status': result[1],\n                        'last_seen': result[2].isoformat() if result[2] else None\n                    }\n                \n        except Exception as e:\n            logger.error(f\"Failed to get worker status for {worker_id}: {e}\")\n        \n        return None\n    \n    def get_system_health(self) -> Dict[str, Any]:\n        \"\"\"Get overall system health metrics.\"\"\"\n        health = {\n            'timestamp': datetime.utcnow().isoformat(),\n            'workers': {\n                'active': 0,\n                'stalled': 0,\n                'stopped': 0\n            },\n            'tasks': {\n                'pending': 0,\n                'claimed': 0,\n                'completed': 0,\n                'failed': 0\n            }\n        }\n        \n        try:\n            with get_db_connection() as conn:\n                cursor = conn.cursor()\n                \n                # Worker counts\n                cursor.execute(\"\"\"\n                    SELECT status, COUNT(*) as count\n                    FROM worker_heartbeats\n                    GROUP BY status\n                \"\"\")\n                \n                for status, count in cursor.fetchall():\n                    if status in health['workers']:\n                        health['workers'][status] = count\n                \n                # Task counts\n                cursor.execute(\"\"\"\n                    SELECT status, COUNT(*) as count\n                    FROM foundry_tasks\n                    GROUP BY status\n                \"\"\")\n                \n                for status, count in cursor.fetchall():\n                    if status in health['tasks']:\n                        health['tasks'][status] = count\n                \n        except Exception as e:\n            logger.error(f\"Failed to get system health: {e}\")\n            health['error'] = str(e)\n        \n        return health\n\n\n# Global heartbeat manager instance\n_heartbeat_manager = None\n\ndef get_heartbeat_manager() -> HeartbeatManager:\n    \"\"\"Get or create global heartbeat manager.\"\"\"\n    global _heartbeat_manager\n    if _heartbeat_manager is None:\n        _heartbeat_manager = HeartbeatManager()\n    return _heartbeat_manager\n\ndef send_heartbeat(worker_id: str, status: Dict[str, Any]) -> bool:\n    \"\"\"Convenience function to send heartbeat.\"\"\"\n    return get_heartbeat_manager().send_heartbeat(worker_id, status)\n\ndef perform_nudge_cycle() -> Dict[str, Any]:\n    \"\"\"Convenience function to perform NUDGE cycle.\"\"\"\n    return get_heartbeat_manager().nudge_cycle()\n</content>\n</rewritten_file>\n\nI've implemented the stale claim reclaimer functionality as specified in SELF-17. Here are the key features:\n\n## Implementation Details\n\n### 1. **Stale Claim Detection**\n- Queries `foundry_tasks` where `status='claimed'` and `claimed_at < now() - 6h`\n- Uses configurable timeout (default 6 hours via `claim_timeout` config)\n\n### 2. **Task Reclamation Process**\n- Resets `status` to 'pending'\n- Clears `claimed_by` and `claimed_at` fields\n- Updates `updated_at` timestamp\n- Limits to max 5 reclaims per cycle to prevent thundering herd\n\n### 3. **Comprehensive Logging**\n- Logs each reclaimed task with:\n  - Task ID and type\n  - Previous claimer worker ID\n  - Duration the task was claimed\n- Maintains audit trail in `task_audit_log` table\n\n### 4. **Integration with NUDGE Cycle**\n- Runs as part of the `nudge_cycle()` method\n- Executes after stall detection as specified\n- Returns metrics including number of reclaimed tasks\n\n### 5. **Error Handling & Safety**\n- Individual task reclaim failures don't stop the process\n- Database transactions ensure consistency\n- Detailed error logging for troubleshooting\n\n### 6. **Configuration Options**\n- `claim_timeout`: How long before claims are considered stale (default 6h)\n- `max_reclaims_per_cycle`: Maximum reclaims per cycle (default 5)\n\nThe reclaimer integrates seamlessly with the existing heartbeat system and will automatically recover tasks that have been abandoned by workers, ensuring system resilience and task completion."
}