job manager

Author: human058382928

agent-tools-ts

@@ -253,7 +253,9 @@ async def _worker(self, worker_name: str) -> None:
                 # Get next job from priority queue
                 execution = await self.priority_queue.get_next_job()
                 if not execution:
-                    await asyncio.sleep(0.1)  # Brief pause if no jobs
+                    await asyncio.sleep(
+                        0.5
+                    )  # Increased pause if no jobs to reduce CPU usage
                     continue
 
                 # Check if we can acquire a slot for this job type

app/services/infrastructure/job_management/executor.py

@@ -4,6 +4,7 @@
 from dataclasses import dataclass
 from datetime import datetime
 from typing import Any, Dict, List, Optional
+from app.backend.factory import backend
 
 from apscheduler.schedulers.asyncio import AsyncIOScheduler
 
@@ -131,12 +132,36 @@ async def _execute_job_via_executor(self, job_type: str) -> None:
             # Convert job_type string to JobType enum
             job_type_enum = JobType.get_or_create(job_type)
 
+            logger.debug(f"🔍 Checking if {job_type} has work to do...")
+
             # Get job metadata to check if it should run
             metadata = JobRegistry.get_metadata(job_type_enum)
             if not metadata:
                 logger.error(f"No metadata found for job type: {job_type}")
                 return
 
+            # For jobs that process messages, check if there are pending messages first
+            # This prevents unnecessary job executions when there's no work
+            if job_type in ["tweet", "discord", "stx_transfer"]:
+                from app.backend.models import QueueMessageFilter
+
+                pending_messages = backend.list_queue_messages(
+                    filters=QueueMessageFilter(
+                        type=QueueMessageType.get_or_create(job_type),
+                        is_processed=False,
+                    )
+                )
+
+                if not pending_messages:
+                    logger.debug(
+                        f"⏭️ Skipping {job_type} execution - no pending messages"
+                    )
+                    return
+
+                logger.debug(
+                    f"📝 Found {len(pending_messages)} pending {job_type} messages"
+                )
+
             # Create a synthetic queue message for scheduled execution
             # This allows the job to go through the proper executor pipeline with concurrency control
             synthetic_message = QueueMessage(
@@ -160,8 +185,8 @@ async def _execute_job_via_executor(self, job_type: str) -> None:
                 synthetic_message, metadata.priority
             )
 
-            logger.debug(
-                f"Enqueued scheduled job {job_type} with ID {job_id} (priority: {metadata.priority})"
+            logger.info(
+                f"✅ Queued {job_type} job {job_id} for execution (priority: {metadata.priority})"
             )
 
         except Exception as e:

app/services/infrastructure/job_management/job_manager.py

@@ -43,12 +43,12 @@ class TweetProcessingResult(RunnerResult):
     job_type="tweet",
     name="Tweet Processor",
     description="Processes and sends tweets for DAOs with automatic retry and error handling",
-    interval_seconds=5,
-    priority=JobPriority.HIGH,
+    interval_seconds=30,  # Reduced frequency from 5s to 30s
+    priority=JobPriority.NORMAL,  # Changed from HIGH to NORMAL to not dominate queue
     max_retries=3,
     retry_delay_seconds=60,
     timeout_seconds=300,
-    max_concurrent=1,
+    max_concurrent=2,  # Increased from 1 to 2 to allow parallel processing
     requires_twitter=True,
     batch_size=5,
     enable_dead_letter_queue=True,
@@ -200,7 +200,7 @@ async def _validate_prerequisites(self, context: JobContext) -> bool:
     async def _validate_task_specific(self, context: JobContext) -> bool:
         """Validate task-specific conditions."""
         if not self._pending_messages:
-            logger.debug("No pending tweet messages found")
+            logger.debug("No pending tweet messages found - skipping execution")
             return False
 
         # Validate each message before processing