diff --git a/.opencode/agents/implementation-orchestrator.md b/.opencode/agents/implementation-orchestrator.md index 630d499e5..bfbaa1882 100644 --- a/.opencode/agents/implementation-orchestrator.md +++ b/.opencode/agents/implementation-orchestrator.md @@ -32,7 +32,9 @@ permission: "timeline-updater": allow "final-reporter": allow "automation-tracking-manager": allow - # implementation-worker removed - launched via curl/prompt_async + # CRITICAL: Use async-agent-starter for ALL worker launches + "async-agent-starter": allow + # implementation-worker BLOCKED - must use async-agent-starter forgejo: "*": allow # CRITICAL: Label creation is COMPLETELY FORBIDDEN @@ -89,6 +91,16 @@ fixes and issue implementation — maintaining a sliding window of N active work at all times. You support configurable parallelism, dependency-aware scheduling, escalation model (codex→sonnet→opus), crash recovery, and web-based CI log access. +**CRITICAL: Worker Launch Protocol** +You MUST use the `async-agent-starter` subagent to launch ALL implementation workers. +DO NOT use direct curl commands to the OpenCode API. The async-agent-starter handles: +- Proper session creation with tagged naming for recovery +- Async agent launch with prompt_async +- Session tracking and monitoring setup +- Error handling and retry logic + +Direct access to `implementation-worker` is BLOCKED in your permissions to enforce this. + **Pool supervisor model:** The `product-builder` launches exactly ONE instance of you. You manage N workers internally (N = `CA_MAX_PARALLEL_WORKERS`). Every time a worker completes, you immediately fill the vacant slot from the queue. @@ -969,9 +981,9 @@ def validate_worker_state(): enforce_worker_limits() validate_worker_state() -# ── Helper: launch one worker via prompt_async with robust verification ── +# ── Helper: launch one worker via async-agent-starter ── function dispatch_worker(mode, work_item, ref_summary): - """Dispatch a worker with comprehensive verification and retry logic.""" + """Dispatch a worker using the async-agent-starter subagent.""" if mode == "pr-fix": # PR fix mode @@ -1011,7 +1023,8 @@ Your task: Fix this PR based on work_type: 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"[AUTO-IMP] worker-pr-fix: PR-{pr.number}" + tag = f"AUTO-IMP-PR-{pr.number}" + display_name = f"worker-pr-fix-{pr.number}" work_id = f"PR-{pr.number}" else: # issue-impl mode @@ -1047,99 +1060,72 @@ Your task: Implement this issue fully, create PR, and shepherd it through review 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"[AUTO-IMP] worker-issue-impl: issue-{issue['number']}" + tag = f"AUTO-IMP-ISSUE-{issue['number']}" + display_name = f"worker-issue-impl-{issue['number']}" work_id = f"issue-{issue['number']}" - def verify_worker_started(session_id, retries=1): - """Quick verification - trust the dispatch worked.""" - bash("sleep 2", timeout=5000) # Brief pause for session to register - - # Single quick check - try: - status_response = bash("curl -s ${SERVER}/session/status", timeout=5000) - if session_id in status_response: - print(f"[VERIFY] ✓ Worker verified active (session {session_id[:8]}...)") - return True - except: - pass # Don't block on verification errors - - # Assume it's still starting - don't block dispatch - print(f"[VERIFY] Session {session_id[:8]}... assumed starting (optimistic)") - return True # Optimistic - let monitoring handle failures - - # Main dispatch logic with retry + # Main dispatch logic with retry using async-agent-starter max_attempts = 2 for attempt in range(max_attempts): try: - print(f"[DISPATCH] Attempt {attempt+1}: Launching worker for {work_id}") + print(f"[DISPATCH] Attempt {attempt+1}: Launching worker for {work_id} via async-agent-starter") - # 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 + # Use async-agent-starter subagent to launch the worker + result = task( + description=f"Start async worker for {work_id}", + subagent_type="async-agent-starter", + prompt=f"""Launch an implementation-worker asynchronously with these parameters: - # 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": "implementation-worker", "parts": [{{"type": "text", "text": "{escaped_prompt}"}}]}}'""" +agent_name: implementation-worker +tag: {tag} +display_name: {display_name} +prompt_text: {prompt} +server_url: {SERVER} +restart_existing: false + +Return the session ID if successful.""" + ) - launch_response = bash(launch_cmd, timeout=30000) - - # Give worker more time to initialize (was 3 seconds, now 5) - print(f"[DISPATCH] Waiting for worker initialization...") - # Verify worker is actually running (includes brief delay) - if verify_worker_started(session_id): - return session_id - else: - print(f"[ERROR] Worker verification failed for {work_id}") - print(f"[ERROR] Session {session_id[:8]}... was created but did not become active within timeout") - print(f"[ERROR] Possible causes: worker crashed on startup, invalid agent name, or server issues") - print(f"[ERROR] Cleaning up orphaned session...") - - cleanup_response = bash(f"curl -s -X DELETE ${SERVER}/session/{session_id}", timeout=15000) - if cleanup_response == "true": - print(f"[CLEANUP] Successfully deleted session {session_id[:8]}...") - else: - print(f"[CLEANUP] Warning: Failed to delete session {session_id[:8]}...") - - if attempt < max_attempts - 1: - print(f"[RETRY] Will retry dispatch in 2 seconds (attempt {attempt+2}/{max_attempts})...") - bash("sleep 2", timeout=5000) # Quick retry - continue - else: - print(f"[CRITICAL] Exhausted all {max_attempts} dispatch attempts for {work_id}") - return None + # Parse the result from async-agent-starter + if result and hasattr(result, 'content'): + try: + # The async-agent-starter returns JSON output + import json + result_data = json.loads(result.content) + if result_data.get('status') == 'success': + session_id = result_data.get('session_id') + if session_id: + print(f"[DISPATCH] ✓ Successfully launched worker via async-agent-starter") + print(f"[DISPATCH] Session ID: {session_id}") + print(f"[DISPATCH] Tag: {tag}") + print(f"[DISPATCH] Display name: {display_name}") + return session_id + else: + print(f"[ERROR] No session ID in successful response") + elif result_data.get('status') == 'skipped': + print(f"[WARNING] Worker launch skipped: {result_data.get('reason')}") + print(f"[WARNING] Message: {result_data.get('message')}") + # Session already exists - try to get its ID + # This shouldn't happen with restart_existing=false + return None + else: + print(f"[ERROR] Worker launch failed: {result_data.get('error')}") + print(f"[ERROR] Operation: {result_data.get('operation')}") + except json.JSONDecodeError as e: + print(f"[ERROR] Failed to parse async-agent-starter response: {e}") + print(f"[ERROR] Raw response: {result.content[:500]}...") + else: + print(f"[ERROR] No response from async-agent-starter") + + if attempt < max_attempts - 1: + print(f"[RETRY] Will retry dispatch in 2 seconds (attempt {attempt+2}/{max_attempts})...") + bash("sleep 2", timeout=5000) + continue + else: + print(f"[CRITICAL] Exhausted all {max_attempts} dispatch attempts for {work_id}") + return None + except Exception as e: print(f"[ERROR] Dispatch attempt {attempt+1} failed for {work_id}: {e}") if attempt < max_attempts - 1: @@ -1406,6 +1392,8 @@ Supervisor: Implementation Pool | Agent: implementation-orchestrator""" bash("sleep 2", timeout=5000) # Get session status from server + # NOTE: Could also use async-agent-monitor subagent for health checks + # but direct API calls are more efficient for bulk status checks status_response = bash("curl -s ${SERVER}/session/status", timeout=30000) all_sessions = safe_json_parse(status_response, [])