Fix critical coordination bugs in implementation pool supervisor

- Fix session adoption logic with correct title patterns for both worker-issue-impl and worker-pr-fix sessions - Add PR worker adoption to coordinate orphaned PR fix workers - Enhance worker verification with comprehensive status checking, retry logic, and proper error handling - Add defensive programming with worker count enforcement and state validation to prevent coordination drift - Improve JSON parsing with safe error handling throughout - Add periodic maintenance cycle (every 5 iterations) for worker state validation and limit enforcement These fixes resolve the core issue where the implementation pool supervisor was not properly coordinating 40+ existing workers, causing worker count to exceed the designed limit of 32.
2026-04-05 19:17:35 +00:00
parent 1bd0c7999d
commit 88cfc33ab2
1 changed files with 348 additions and 91 deletions
@@ -349,23 +349,41 @@ pattern with configurable parallelism and **speculative pre-cloning** for
 maximum throughput.

 ```
-max_workers = CA_MAX_PARALLEL_WORKERS (from env, or ask user if unset)
-queue       = []   # prioritized list of unblocked issues to work on
-active_pr_workers = {}     # pr_number -> {session_id, work_type, assigned_at, issue_number}
-active_issue_workers = {}  # issue_number -> session_id
-pr_work_queue = []        # PRs needing work but no worker assigned yet
-completed   = []   # list of {issue_number, branch, pr_number, pr_url, ...}
-completed_issues = set()  # Set of completed issue numbers
-failed      = {}   # issue_number -> consecutive_failure_count
-ref_summary = result from ca-ref-reader
-cycle       = 0
-SERVER      = "http://localhost:4096"
-FORGEJO_USERNAME = <from env or user>
-owner       = "cleveragents"
-repo        = "cleveragents-core"
-targeted_issue_numbers = []  # If user specifies specific issues
+# Initialize core state variables with proper error checking
+try:
+    max_workers = int(bash("echo $CA_MAX_PARALLEL_WORKERS", timeout=5000) or "4")
+    if max_workers <= 0 or max_workers > 64:
+        print(f"[WARNING] Invalid max_workers {max_workers}, using default 32")
+        max_workers = 32
+except:
+    print(f"[WARNING] Could not parse CA_MAX_PARALLEL_WORKERS, using default 32")
+    max_workers = 32
+
+print(f"[CONFIG] Max parallel workers: {max_workers}")
+
+# Core tracking dictionaries
+queue = []                    # prioritized list of unblocked issues to work on
+active_pr_workers = {}        # pr_number -> {session_id, work_type, assigned_at, issue_number}
+active_issue_workers = {}     # issue_number -> session_id
+pr_work_queue = []           # PRs needing work but no worker assigned yet
+completed = []               # list of {issue_number, branch, pr_number, pr_url, ...}
+completed_issues = set()     # Set of completed issue numbers
+failed = {}                  # issue_number -> consecutive_failure_count
+ref_summary = result_from_ca_ref_reader  # Will be set after ref-reader completes
+cycle = 0
+
+# Constants
+SERVER = "http://localhost:4096"
+FORGEJO_USERNAME = forgejo_username  # From startup sequence
+owner = "cleveragents"
+repo = "cleveragents-core"
+targeted_issue_numbers = []         # If user specifies specific issues
 selective_issue_targeting = False

+print(f"[CONFIG] Repository: {owner}/{repo}")
+print(f"[CONFIG] Forgejo username: {FORGEJO_USERNAME}")
+print(f"[CONFIG] OpenCode server: {SERVER}")
+
 # Helper function to extract branch from issue body metadata
 function extract_branch_from_issue_body(body):
    # Look for branch in metadata section
@@ -391,26 +409,168 @@ function now():
 # active tracking instead of launching duplicates.
 # To start fresh, run ca-session-cleanup BEFORE the product-builder.

-EXISTING_WORKERS = bash("curl -s ${SERVER}/session | python3 -c \"
-import sys, json
-for s in json.loads(sys.stdin.read()):
-    title = s.get('title','')
-    if title.startswith('[CA-AUTO] worker-impl:'):
-        # Extract issue number from title
-        issue_num = title.replace('[CA-AUTO] worker-impl: issue-','')
-        print(issue_num + '=' + s['id'])
-\"", timeout=30000)
+def adopt_existing_workers():
+    """Adopt orphaned worker sessions into coordination."""
+    print("[ADOPTION] Checking for existing worker sessions to adopt...")
+    
+    try:
+        # Get all active sessions
+        sessions_response = bash("curl -s ${SERVER}/session", timeout=30000)
+        sessions = json.loads(sessions_response) if sessions_response else []
+        
+        # Get session status for verification
+        status_response = bash("curl -s ${SERVER}/session/status", timeout=30000) 
+        status_data = json.loads(status_response) if status_response else []
+        active_session_ids = {s["id"] for s in status_data if s.get("status") == "active"}
+        
+        adopted_issue_workers = 0
+        adopted_pr_workers = 0
+        
+        for session in sessions:
+            title = session.get('title', '')
+            session_id = session.get('id', '')
+            
+            # Only adopt active sessions
+            if session_id not in active_session_ids:
+                continue
+                
+            # Adopt issue implementation workers
+            if title.startswith('[CA-AUTO] worker-issue-impl: issue-'):
+                try:
+                    # Extract issue number: "[CA-AUTO] worker-issue-impl: issue-123"
+                    issue_num_str = title.replace('[CA-AUTO] worker-issue-impl: issue-', '').strip()
+                    issue_number = int(issue_num_str)
+                    active_issue_workers[issue_number] = session_id
+                    adopted_issue_workers += 1
+                    print(f"[ADOPTION] Adopted issue worker: Issue #{issue_number} -> {session_id[:8]}...")
+                except (ValueError, IndexError) as e:
+                    print(f"[WARNING] Could not parse issue number from title: {title}")
+                    
+            # Adopt PR fix workers  
+            elif title.startswith('[CA-AUTO] worker-pr-fix: PR-'):
+                try:
+                    # Extract PR number: "[CA-AUTO] worker-pr-fix: PR-456"
+                    pr_num_str = title.replace('[CA-AUTO] worker-pr-fix: PR-', '').strip()
+                    pr_number = int(pr_num_str)
+                    
+                    # We need to reconstruct the PR worker info - get PR data from Forgejo
+                    pr_data = forgejo_get_pull_request_by_index(owner, repo, pr_number)
+                    if pr_data:
+                        # Extract issue number from PR body
+                        issue_number = None
+                        if "Closes #" in pr_data.body:
+                            issue_number = int(pr_data.body.split("Closes #")[1].split()[0])
+                        elif "Fixes #" in pr_data.body:
+                            issue_number = int(pr_data.body.split("Fixes #")[1].split()[0])
+                            
+                        active_pr_workers[pr_number] = {
+                            "session_id": session_id,
+                            "work_type": "unknown",  # Will be re-determined in next cycle
+                            "assigned_at": now(),
+                            "issue_number": issue_number
+                        }
+                        adopted_pr_workers += 1
+                        print(f"[ADOPTION] Adopted PR worker: PR #{pr_number} -> {session_id[:8]}...")
+                except (ValueError, IndexError) as e:
+                    print(f"[WARNING] Could not parse PR number from title: {title}")
+        
+        print(f"[ADOPTION] Adopted {adopted_issue_workers} issue workers, {adopted_pr_workers} PR workers")
+        return adopted_issue_workers + adopted_pr_workers
+        
+    except Exception as e:
+        print(f"[ERROR] Session adoption failed: {e}")
+        return 0

-STATUS = bash("curl -s ${SERVER}/session/status", timeout=30000)
-for line in EXISTING_WORKERS:
-    issue_number, session_id = line.split("=")
-    if session_id is active in STATUS:
-        active[int(issue_number)] = session_id  # Adopt into active tracking
-        # Remove from queue if present (already being worked on)
-        queue = [i for i in queue if i.number != int(issue_number)]
+# Call adoption function
+total_adopted = adopt_existing_workers()

-# ── Helper: launch one worker via prompt_async ───────────────────
+# ── Defensive Programming: Worker Count Enforcement ──────────────
+def enforce_worker_limits():
+    """Enforce hard worker count limits and clean up excess workers."""
+    total_workers = len(active_pr_workers) + len(active_issue_workers)
+    
+    if total_workers > max_workers:
+        print(f"[WARNING] Worker count ({total_workers}) exceeds limit ({max_workers})")
+        
+        # Clean up oldest workers first (LIFO cleanup to preserve recent work)
+        excess = total_workers - max_workers
+        
+        # First clean up issue workers (PRs have higher priority)
+        issue_items = list(active_issue_workers.items())
+        for i, (issue_num, session_id) in enumerate(issue_items[-excess:]):
+            print(f"[CLEANUP] Terminating excess issue worker: Issue #{issue_num}")
+            bash(f"curl -s -X DELETE ${SERVER}/session/{session_id}", timeout=15000)
+            del active_issue_workers[issue_num]
+            excess -= 1
+            if excess <= 0:
+                break
+        
+        # If still over limit, clean up PR workers
+        if excess > 0:
+            pr_items = list(active_pr_workers.items())
+            for i, (pr_num, pr_info) in enumerate(pr_items[-excess:]):
+                print(f"[CLEANUP] Terminating excess PR worker: PR #{pr_num}")
+                bash(f"curl -s -X DELETE ${SERVER}/session/{pr_info['session_id']}", timeout=15000)
+                del active_pr_workers[pr_num]
+                excess -= 1
+                if excess <= 0:
+                    break
+
+def validate_worker_state():
+    """Validate that tracked workers actually exist and are active."""
+    print("[VALIDATION] Validating worker state consistency...")
+    
+    try:
+        status_response = bash("curl -s ${SERVER}/session/status", timeout=30000)
+        if not status_response:
+            print("[WARNING] Could not get session status for validation")
+            return
+            
+        status_data = json.loads(status_response)
+        active_session_ids = {s["id"] for s in status_data if s.get("status") == "active"}
+        
+        # Validate issue workers
+        dead_issue_workers = []
+        for issue_num, session_id in active_issue_workers.items():
+            if session_id not in active_session_ids:
+                print(f"[VALIDATION] Issue worker #{issue_num} session {session_id[:8]}... is dead")
+                dead_issue_workers.append(issue_num)
+        
+        # Clean up dead issue workers
+        for issue_num in dead_issue_workers:
+            del active_issue_workers[issue_num]
+            print(f"[CLEANUP] Removed dead issue worker tracking for Issue #{issue_num}")
+            
+        # Validate PR workers  
+        dead_pr_workers = []
+        for pr_num, pr_info in active_pr_workers.items():
+            session_id = pr_info["session_id"]
+            if session_id not in active_session_ids:
+                print(f"[VALIDATION] PR worker #{pr_num} session {session_id[:8]}... is dead")
+                dead_pr_workers.append(pr_num)
+        
+        # Clean up dead PR workers
+        for pr_num in dead_pr_workers:
+            del active_pr_workers[pr_num]
+            print(f"[CLEANUP] Removed dead PR worker tracking for PR #{pr_num}")
+            
+        cleaned_count = len(dead_issue_workers) + len(dead_pr_workers)
+        if cleaned_count > 0:
+            print(f"[VALIDATION] Cleaned up {cleaned_count} dead worker references")
+        else:
+            print(f"[VALIDATION] ✓ All {len(active_issue_workers)} issue + {len(active_pr_workers)} PR workers are active")
+            
+    except Exception as e:
+        print(f"[ERROR] Worker state validation failed: {e}")
+
+# Run initial worker limit enforcement and validation
+enforce_worker_limits()
+validate_worker_state()
+
+# ── Helper: launch one worker via prompt_async with robust verification ──
 function dispatch_worker(mode, work_item, ref_summary):
+    """Dispatch a worker with comprehensive verification and retry logic."""
+    
    if mode == "pr-fix":
        # PR fix mode
        pr = work_item["pr"]
@@ -442,6 +602,7 @@ CRITICAL: You own this PR until it is merged. Monitor and handle all feedback.
 Do not exit until the PR is merged or blocked by human feedback (needs feedback label)."""
        
        title = f"[CA-AUTO] worker-pr-fix: PR-{pr.number}"
+        work_id = f"PR-{pr.number}"
        
    else:  # issue-impl mode
        issue = work_item
@@ -468,25 +629,141 @@ CRITICAL: You own this issue from implementation through PR merge.
 Do not exit until the PR is merged. Monitor and handle all review feedback."""
        
        title = f"[CA-AUTO] worker-issue-impl: issue-{issue['number']}"
-    
-    # Create session
-    SESSION_ID = bash("curl -s -X POST ${SERVER}/session \
-        -H 'Content-Type: application/json' \
-        -d '{\"title\": \"${title}\"}' \
-        | python3 -c \"import sys,json; print(json.loads(sys.stdin.read())['id'])\"",
-        timeout=30000)
+        work_id = f"issue-{issue['number']}"

-    # Launch worker
-    bash("curl -s -X POST ${SERVER}/session/${SESSION_ID}/prompt_async \
-        -H 'Content-Type: application/json' \
-        -d '{\"agent\": \"ca-issue-worker\", \
-             \"parts\": [{\"type\": \"text\", \"text\": \"${prompt}\"}]}'",
-        timeout=30000)
+    def verify_worker_started(session_id, retries=3):
+        """Comprehensively verify worker is running."""
+        for attempt in range(retries):
+            try:
+                # Check session exists and is active
+                session_response = bash(f"curl -s ${SERVER}/session/{session_id}", timeout=15000)
+                
+                if not session_response or "not found" in session_response.lower():
+                    print(f"[VERIFY] Attempt {attempt+1}: Session {session_id[:8]}... not found")
+                    bash("sleep 2", timeout=5000)
+                    continue
+                
+                # Parse JSON safely
+                try:
+                    session_data = json.loads(session_response)
+                except:
+                    print(f"[VERIFY] Attempt {attempt+1}: Invalid JSON response for session {session_id[:8]}...")
+                    bash("sleep 2", timeout=5000)
+                    continue
+                
+                # Check if session is active
+                status_response = bash("curl -s ${SERVER}/session/status", timeout=15000)
+                if status_response:
+                    try:
+                        status_data = json.loads(status_response)
+                        session_active = any(s["id"] == session_id and s.get("status") == "active" 
+                                           for s in status_data)
+                        if session_active:
+                            print(f"[VERIFY] ✓ Worker {work_id} verified active (session {session_id[:8]}...)")
+                            return True
+                        else:
+                            print(f"[VERIFY] Attempt {attempt+1}: Session {session_id[:8]}... not active")
+                    except:
+                        print(f"[VERIFY] Attempt {attempt+1}: Could not parse status response")
+                
+                bash("sleep 3", timeout=5000)
+                
+            except Exception as e:
+                print(f"[VERIFY] Attempt {attempt+1}: Verification error: {e}")
+                bash("sleep 3", timeout=5000)
+        
+        print(f"[ERROR] Worker verification failed after {retries} attempts for {work_id}")
+        return False

-    return SESSION_ID
+    # Main dispatch logic with retry
+    max_attempts = 2
+    for attempt in range(max_attempts):
+        try:
+            print(f"[DISPATCH] Attempt {attempt+1}: Launching worker for {work_id}")
+            
+            # Create session with error checking
+            session_create_cmd = f"""curl -s -X POST ${SERVER}/session \\
+                -H 'Content-Type: application/json' \\
+                -d '{{"title": "{title}"}}' """
+            
+            session_response = bash(session_create_cmd, timeout=30000)
+            
+            if not session_response:
+                print(f"[ERROR] Empty response from session creation for {work_id}")
+                if attempt < max_attempts - 1:
+                    bash("sleep 5", timeout=10000)
+                    continue
+                else:
+                    return None
+            
+            # Parse session ID safely
+            try:
+                session_data = json.loads(session_response)
+                session_id = session_data.get('id')
+                if not session_id:
+                    print(f"[ERROR] No session ID in response for {work_id}")
+                    if attempt < max_attempts - 1:
+                        bash("sleep 5", timeout=10000)
+                        continue
+                    else:
+                        return None
+            except json.JSONDecodeError as e:
+                print(f"[ERROR] Invalid JSON from session creation for {work_id}: {e}")
+                if attempt < max_attempts - 1:
+                    bash("sleep 5", timeout=10000)
+                    continue
+                else:
+                    return None
+
+            # Launch worker with escaped prompt
+            escaped_prompt = prompt.replace('"', '\\"').replace('\n', '\\n')
+            launch_cmd = f"""curl -s -X POST ${SERVER}/session/{session_id}/prompt_async \\
+                -H 'Content-Type: application/json' \\
+                -d '{{"agent": "ca-issue-worker", "parts": [{{"type": "text", "text": "{escaped_prompt}"}}]}}'"""
+            
+            launch_response = bash(launch_cmd, timeout=30000)
+            
+            # Brief wait for worker to initialize
+            bash("sleep 3", timeout=10000)
+            
+            # Verify worker is actually running
+            if verify_worker_started(session_id):
+                return session_id
+            else:
+                print(f"[ERROR] Worker failed to start properly for {work_id}, cleaning up session {session_id[:8]}...")
+                bash(f"curl -s -X DELETE ${SERVER}/session/{session_id}", timeout=15000)
+                
+                if attempt < max_attempts - 1:
+                    bash("sleep 10", timeout=15000)  # Longer wait before retry
+                    continue
+                else:
+                    return None
+                    
+        except Exception as e:
+            print(f"[ERROR] Dispatch attempt {attempt+1} failed for {work_id}: {e}")
+            if attempt < max_attempts - 1:
+                bash("sleep 10", timeout=15000)
+            
+    print(f"[CRITICAL] All dispatch attempts failed for {work_id}")
+    return None

 # ── Main dispatch + monitoring loop ──────────────────────────────

+# Import required modules for JSON handling
+import json
+import datetime
+
+# Helper function to safely parse JSON responses
+def safe_json_parse(response, default=None):
+    """Safely parse JSON response with error handling."""
+    if not response:
+        return default or []
+    try:
+        return json.loads(response)
+    except (json.JSONDecodeError, TypeError) as e:
+        print(f"[JSON ERROR] Failed to parse response: {e}")
+        return default or []
+
 LOOP FOREVER:
    cycle += 1

@@ -520,19 +797,11 @@ LOOP FOREVER:
        # Actually dispatch PR fix worker using the helper function
        session_id = dispatch_worker("pr-fix", pr_work, ref_summary)
        
-        # Verify the worker actually started
-        bash("sleep 2", timeout=5000)  # Brief wait
-        verify_status = bash(f"curl -s ${SERVER}/session/{session_id}", timeout=30000)
-        
-        if "error" in verify_status.lower() or "not found" in verify_status.lower():
-            print(f"[ERROR] Worker dispatch failed for PR #{pr_work['pr'].number}")
-            # Retry dispatch
-            session_id = dispatch_worker("pr-fix", pr_work, ref_summary)
-            bash("sleep 2", timeout=5000)
-            verify_status = bash(f"curl -s ${SERVER}/session/{session_id}", timeout=30000)
-            if "error" in verify_status.lower() or "not found" in verify_status.lower():
-                print(f"[ERROR] Retry failed - skipping PR #{pr_work['pr'].number}")
-                continue
+        if session_id is None:
+            print(f"[CRITICAL] Failed to dispatch worker for PR #{pr_work['pr'].number} - skipping")
+            # Put work back in queue for next cycle
+            pr_work_queue.insert(0, pr_work)
+            break  # Stop trying to dispatch more workers this cycle
        
        active_pr_workers[pr_work["pr"].number] = {
            "session_id": session_id,
@@ -543,7 +812,7 @@ LOOP FOREVER:
        slots_available -= 1
        
        # Log dispatch
-        print(f"[{now()}] Dispatched PR-fix worker for PR #{pr_work['pr'].number} ({pr_work['work_type']}) - verified active")
+        print(f"[{now()}] ✓ Dispatched PR-fix worker for PR #{pr_work['pr'].number} ({pr_work['work_type']}) - session {session_id[:8]}...")
    
    # ── STEP 3: Only dispatch to issues if ALL PRs have workers ──────
    # CRITICAL: Block ALL issue work if ANY PR needs attention
@@ -628,33 +897,22 @@ LOOP FOREVER:
            # Actually dispatch issue implementation worker
            session_id = dispatch_worker("issue-impl", issue_work, ref_summary)
            
-            # Verify the worker actually started
-            bash("sleep 2", timeout=5000)  # Brief wait
-            verify_status = bash(f"curl -s ${SERVER}/session/{session_id}", timeout=30000)
-            
-            if "error" in verify_status.lower() or "not found" in verify_status.lower():
-                print(f"[ERROR] Worker dispatch failed for Issue #{issue.number}")
-                # Retry dispatch
-                session_id = dispatch_worker("issue-impl", issue_work, ref_summary)
-                bash("sleep 2", timeout=5000)
-                verify_status = bash(f"curl -s ${SERVER}/session/{session_id}", timeout=30000)
-                if "error" in verify_status.lower() or "not found" in verify_status.lower():
-                    print(f"[ERROR] Retry failed - skipping Issue #{issue.number}")
-                    continue
+            if session_id is None:
+                print(f"[CRITICAL] Failed to dispatch worker for Issue #{issue.number} - skipping")
+                # Put issue back in queue for next cycle
+                queue.insert(0, issue)
+                break  # Stop trying to dispatch more workers this cycle
            
            active_issue_workers[issue.number] = session_id
            slots_available -= 1
-            print(f"[{now()}] Dispatched issue worker for Issue #{issue.number} - verified active")
+            print(f"[{now()}] ✓ Dispatched issue worker for Issue #{issue.number} - session {session_id[:8]}...")

    # ── Monitor active workers: poll every 10 seconds ────────────
    bash("sleep 10", timeout=30000)
    
    # Get session status from server
-    try:
-        status_response = bash("curl -s ${SERVER}/session/status", timeout=30000)
-        all_sessions = json.loads(status_response) if status_response else []
-    except:
-        all_sessions = []
+    status_response = bash("curl -s ${SERVER}/session/status", timeout=30000)
+    all_sessions = safe_json_parse(status_response, [])
    
    # Monitor PR workers
    for pr_number, pr_info in list(active_pr_workers.items()):
@@ -666,12 +924,9 @@ LOOP FOREVER:
        
        if not session_active:
            # Session completed or died - check final result
-            try:
-                final_response = bash(f"curl -s ${SERVER}/session/{session_id}/messages", timeout=30000)
-                messages = json.loads(final_response) if final_response else []
-                final_msg = messages[-1]["content"] if messages else "ERROR"
-            except:
-                final_msg = "ERROR: Could not retrieve final message"
+            final_response = bash(f"curl -s ${SERVER}/session/{session_id}/messages", timeout=30000)
+            messages = safe_json_parse(final_response, [])
+            final_msg = messages[-1].get("content", "ERROR") if messages else "ERROR: No messages found"
            
            # Parse result
            if "PR merged successfully" in final_msg:
@@ -705,12 +960,9 @@ LOOP FOREVER:
        
        if not session_active:
            # Session completed or died - check final result
-            try:
-                final_response = bash(f"curl -s ${SERVER}/session/{session_id}/messages", timeout=30000)
-                messages = json.loads(final_response) if final_response else []
-                final_msg = messages[-1]["content"] if messages else "ERROR"
-            except:
-                final_msg = "ERROR: Could not retrieve final message"
+            final_response = bash(f"curl -s ${SERVER}/session/{session_id}/messages", timeout=30000)
+            messages = safe_json_parse(final_response, [])
+            final_msg = messages[-1].get("content", "ERROR") if messages else "ERROR: No messages found"
            
            if "PR merged successfully" in final_msg or "PR created successfully" in final_msg:
                # Success - issue implemented and PR created/merged
@@ -751,6 +1003,11 @@ LOOP FOREVER:
            bash(f"curl -s -X DELETE ${SERVER}/session/{session_id}", timeout=15000)
            del active_issue_workers[issue_number]

+    # ── Periodic maintenance every 5 cycles ──────────────────────
+    if cycle % 5 == 0:
+        validate_worker_state()
+        enforce_worker_limits()
+
    # ── Health signal every 10 cycles ─────────────────────────────
    if cycle % 10 == 0:
        # Helper functions to format worker details