cleveragents/cleveragents-core
4
0
Fork 3
Code Issues 6.0k Pull Requests 115 Actions Packages Projects Releases Wiki Activity

fix(agents): update implementation-orchestrator to use async-agent-starter for worker dispatch
CI / push-validation (push) Successful in 16s
Details
CI / build (push) Successful in 28s
Details
CI / quality (push) Successful in 34s
Details
CI / lint (push) Successful in 34s
Details
CI / helm (push) Successful in 32s
Details
CI / security (push) Successful in 54s
Details
CI / typecheck (push) Successful in 57s
Details
CI / benchmark-regression (push) Has been skipped
Details
CI / e2e_tests (push) Successful in 3m8s
Details
CI / integration_tests (push) Successful in 4m2s
Details
CI / unit_tests (push) Successful in 5m8s
Details
CI / docker (push) Successful in 10s
Details
CI / coverage (push) Successful in 10m21s
Details
CI / status-check (push) Successful in 1s
Details
CI / benchmark-publish (push) Has been cancelled
Details

Browse Source 
- Add async-agent-starter to allowed task permissions
- Replace direct curl commands with async-agent-starter subagent calls
- Block direct access to implementation-worker to enforce async pattern
- Add explicit instructions about worker launch protocol
- Improve error handling for async worker dispatch

This fixes the issue where the orchestrator failed to launch its pool of
parallel workers. Now it properly uses the async-agent-starter subagent
which handles session creation, tagging, and async launch correctly.

Each worker gets a unique tag (AUTO-IMP-PR-{number} or AUTO-IMP-ISSUE-{number})
for monitoring and recovery. The async approach ensures proper session
management and follows the same pattern as other supervisors in the system.
This commit is contained in:
clever-agent
2026-04-09 18:01:55 +00:00
parent 894d57594f
commit 73e9087df1
1 changed files with 76 additions and 88 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.opencode/agents/implementation-orchestrator.md
+76 -88
View File
@@ -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, [])