Rebase Agent
1276 lines
48 KiB
Bash
Executable File
1276 lines
48 KiB
Bash
Executable File
#!/usr/bin/env bash
|
||
#
|
||
# opencode-builder.sh
|
||
#
|
||
# Script-driven orchestrator for the auto-agents system.
|
||
# Queries the OpenCode server fleet, calculates worker pool gaps,
|
||
# fetches work items directly via list_prs_* / list_issues scripts,
|
||
# and launches workers via session_start.ts — no supervisor agents needed.
|
||
|
||
set -uo pipefail
|
||
|
||
# ── Debug mode ─────────────────────────────────────────────────────────────
|
||
if [[ "${BUILDER_DEBUG:-0}" == "1" ]]; then
|
||
set -x
|
||
log "DEBUG MODE ENABLED"
|
||
fi
|
||
|
||
# ── Configuration (override via environment) ──────────────────────────────
|
||
PORT="${OPENCODE_PORT:-4096}"
|
||
HOST="${OPENCODE_HOST:-127.0.0.1}"
|
||
BASE="http://${HOST}:${PORT}"
|
||
HEALTH_TIMEOUT=60
|
||
MAX_IDLE_PER_MINUTE=10
|
||
SCRIPT_TIMEOUT_SEC=1200 # timeout for list_prs / list_issues scripts (Forgejo is slow — allow 20 min)
|
||
SESSION_SCRIPT_TIMEOUT=30 # timeout for session_* scripts
|
||
|
||
# Worker pool sizing
|
||
total_max_workers="${CA_MAX_PARALLEL_WORKERS:-4}"
|
||
# Calculate per-type max workers (same logic as auto-agents)
|
||
merge_max_workers=$(awk "BEGIN {print int((${total_max_workers} + 7) / 8)}")
|
||
imp_max_workers="${total_max_workers}"
|
||
rev_max_workers=$(awk "BEGIN {print int((${total_max_workers} + 1) / 2)}")
|
||
# Grooming runs cheap metadata-only work via Forgejo API. Each task is fast
|
||
# (a few REST calls), so we don't need many parallel slots — ceil(N/4).
|
||
groom_max_workers=$(awk "BEGIN {print int((${total_max_workers} + 3) / 4)}")
|
||
|
||
# Health check intervals (seconds)
|
||
DOOM_CHECK_INTERVAL=900 # 15 minutes
|
||
DOOM_AGE_THRESHOLD=1800 # 30 minutes
|
||
CONTINUE_BURST_WINDOW=60 # seconds
|
||
CONTINUE_BURST_LIMIT=5
|
||
SLEEP_WHEN_IDLE=30 # seconds
|
||
IDLE_THRESHOLD_MS=180000 # 3 minutes — session considered idle if last_active older than this
|
||
|
||
# Script paths
|
||
SCRIPT_DIR="/app/.opencode/skills/auto-agents-system/scripts"
|
||
SESSION_LIST="npx --yes tsx ${SCRIPT_DIR}/session_list.ts"
|
||
SESSION_FIND_PREFIX="npx --yes tsx ${SCRIPT_DIR}/session_find_by_prefix.ts"
|
||
SESSION_START="npx --yes tsx ${SCRIPT_DIR}/session_start.ts"
|
||
SESSION_MESSAGES="npx --yes tsx ${SCRIPT_DIR}/session_messages.ts"
|
||
SESSION_DELETE="npx --yes tsx ${SCRIPT_DIR}/session_delete.ts"
|
||
SESSION_STOP="npx --yes tsx ${SCRIPT_DIR}/session_stop.ts"
|
||
|
||
# ── Internal state ───────────────────────────────────────────────────────
|
||
SERVER_PID=""
|
||
CURL_PID=""
|
||
STOP_LOOP=false
|
||
LAST_SIGINT=0
|
||
OWN_SERVER=false
|
||
N=0
|
||
IDLE_EVENTS_FILE=""
|
||
|
||
# In-memory tracking for health checks
|
||
declare -A QUESTION_COUNT
|
||
declare -A CONTINUE_TIMES
|
||
LAST_DOOM_CHECK=0
|
||
|
||
# ── Helpers ───────────────────────────────────────────────────────────────
|
||
die() { printf "ERROR: %s\n" "$*" >&2; exit 1; }
|
||
log() { printf "[%s] %s\n" "$(date +%H:%M:%S)" "$*"; }
|
||
log_warn() { printf "[%s] WARN: %s\n" "$(date +%H:%M:%S)" "$*" >&2; }
|
||
log_error() { printf "[%s] ERROR: %s\n" "$(date +%H:%M:%S)" "$*" >&2; }
|
||
|
||
is_timeout() {
|
||
local ec="${1:-0}"
|
||
[[ $ec -eq 124 || $ec -eq 137 || $ec -eq 143 ]]
|
||
}
|
||
|
||
# run_timed OUT_VAR TIMEOUT_SEC CMD [ARG...]
|
||
#
|
||
# Runs CMD under timeout(1), capturing stdout into OUT_VAR and exit code into
|
||
# RUN_TIMED_EXITCODE. The command's stderr is captured separately (fd 3) and
|
||
# logged via log_warn so it never pollutes the output that callers parse with jq.
|
||
# A warning is always logged: on timeout (exit 124/137/143), non-zero failure,
|
||
# or zero-with-stderr. Empty output on timeout/failure is intentional — callers
|
||
# that pipe into jq must handle it gracefully.
|
||
#
|
||
# Usage:
|
||
# run_timed result 30 npx --yes tsx script.ts --arg1 val1
|
||
# if is_timeout "${RUN_TIMED_EXITCODE:-0}"; then
|
||
# echo "Timed out after 30s"
|
||
# fi
|
||
run_timed() {
|
||
local out_var="$1"
|
||
local timeout_sec="$2"
|
||
shift 2
|
||
local cmd_str="$*"
|
||
local tmpout
|
||
local tmperr
|
||
tmpout=$(mktemp)
|
||
tmperr=$(mktemp)
|
||
local exitcode=0
|
||
|
||
timeout "$timeout_sec" "$@" > "$tmpout" 2> "$tmperr"
|
||
exitcode=$?
|
||
|
||
if is_timeout $exitcode; then
|
||
log_warn "TIMEOUT after ${timeout_sec}s: ${cmd_str}"
|
||
printf -v "$out_var" '%s' ""
|
||
elif [[ $exitcode -ne 0 ]]; then
|
||
log_warn "FAILED (exit ${exitcode}) after ${timeout_sec}s: ${cmd_str}"
|
||
local err_snippet
|
||
err_snippet=$(head -n 3 "$tmperr" 2>/dev/null | tr '\n' '|' | sed 's/|$//')
|
||
if [[ -n "$err_snippet" ]]; then
|
||
log_warn " stderr: ${err_snippet}"
|
||
fi
|
||
printf -v "$out_var" '%s' ""
|
||
else
|
||
local output
|
||
output=$(cat "$tmpout")
|
||
local err_output
|
||
err_output=$(cat "$tmperr")
|
||
if [[ -n "$err_output" ]]; then
|
||
local err_first_line
|
||
err_first_line=$(echo "$err_output" | head -n 1)
|
||
if [[ "$err_first_line" == ERROR:* || "$err_first_line" == WARNING:* || "$err_first_line" == FAILED:* ]]; then
|
||
log_warn "Command produced stderr: ${cmd_str}"
|
||
local err_snippet
|
||
err_snippet=$(echo "$err_output" | head -n 2 | tr '\n' '|' | sed 's/|$//')
|
||
log_warn " stderr snippet: ${err_snippet}"
|
||
fi
|
||
fi
|
||
printf -v "$out_var" '%s' "$output"
|
||
fi
|
||
|
||
RUN_TIMED_EXITCODE=$exitcode
|
||
rm -f "$tmpout" "$tmperr"
|
||
}
|
||
|
||
check_opencode_running() {
|
||
if curl -sf "${BASE}/global/health" >/dev/null 2>&1; then
|
||
return 0
|
||
else
|
||
return 1
|
||
fi
|
||
}
|
||
|
||
cleanup() {
|
||
if [[ -n "$IDLE_EVENTS_FILE" ]]; then
|
||
rm -f "$IDLE_EVENTS_FILE" 2>/dev/null || true
|
||
fi
|
||
if $OWN_SERVER && [[ -n "$SERVER_PID" ]] && kill -0 "$SERVER_PID" 2>/dev/null; then
|
||
log "Stopping opencode server (PID $SERVER_PID)..."
|
||
kill "$SERVER_PID" 2>/dev/null
|
||
wait "$SERVER_PID" 2>/dev/null || true
|
||
elif ! $OWN_SERVER; then
|
||
log "Leaving existing OpenCode server running (not managed by this script)"
|
||
fi
|
||
}
|
||
|
||
handle_sigint() {
|
||
local now
|
||
now=$(date +%s)
|
||
|
||
if $STOP_LOOP && (( now - LAST_SIGINT <= 2 )); then
|
||
printf "\nForce quit!\n"
|
||
[[ -n "$CURL_PID" ]] && kill "$CURL_PID" 2>/dev/null || true
|
||
cleanup
|
||
exit 130
|
||
fi
|
||
|
||
STOP_LOOP=true
|
||
LAST_SIGINT=$now
|
||
printf "\nWill stop after the current request completes. Ctrl+C again within 2s to force-quit.\n"
|
||
}
|
||
|
||
trap handle_sigint INT
|
||
trap cleanup EXIT
|
||
|
||
# ── Idle-event tracking ───────────────────────────────────────────────────
|
||
init_idle_tracking() {
|
||
IDLE_EVENTS_FILE="/tmp/opencode-builder-idle-events-$$.txt"
|
||
: > "$IDLE_EVENTS_FILE" 2>/dev/null || true
|
||
}
|
||
|
||
record_idle_event() {
|
||
local now
|
||
now=$(date +%s)
|
||
echo "$now" >> "$IDLE_EVENTS_FILE"
|
||
}
|
||
|
||
count_idle_events_last_minute() {
|
||
local now cutoff
|
||
now=$(date +%s)
|
||
cutoff=$((now - 60))
|
||
if [[ -f "$IDLE_EVENTS_FILE" ]]; then
|
||
awk -v c="$cutoff" '$1 > c {count++} END {print count+0}' "$IDLE_EVENTS_FILE" 2>/dev/null
|
||
else
|
||
echo 0
|
||
fi
|
||
}
|
||
|
||
# ── Session lifecycle helpers ─────────────────────────────────────────────
|
||
query_all_sessions() {
|
||
local result
|
||
run_timed result "$SESSION_SCRIPT_TIMEOUT" \
|
||
$SESSION_LIST --server "${BASE}"
|
||
echo "$result"
|
||
}
|
||
|
||
find_sessions_by_prefix() {
|
||
local prefix="$1"
|
||
local status_filter="${2:-all}"
|
||
local result
|
||
run_timed result "$SESSION_SCRIPT_TIMEOUT" \
|
||
$SESSION_FIND_PREFIX --server "${BASE}" --prefix "$prefix" --exclude-supervisor --status "$status_filter"
|
||
echo "$result"
|
||
}
|
||
|
||
get_busy_worker_tags() {
|
||
local prefix="$1"
|
||
local now_ms
|
||
now_ms=$(date +%s000)
|
||
local cutoff_ms=$((now_ms - 120000))
|
||
local result
|
||
result=$(find_sessions_by_prefix "$prefix" "all" | jq -r \
|
||
--arg cutoff "$cutoff_ms" \
|
||
'.[] | select(.last_active >= ($cutoff | tonumber)) | .tag // empty' 2>/dev/null)
|
||
if [[ -z "$result" ]]; then
|
||
log_warn "[get_busy_worker_tags:${prefix}] no busy tags found (or find_sessions_by_prefix timed out)"
|
||
fi
|
||
echo "$result"
|
||
}
|
||
|
||
create_session() {
|
||
local title="$1"
|
||
curl -sf -X POST "${BASE}/session" \
|
||
-H "Content-Type: application/json" \
|
||
-d "{\"title\":\"${title}\"}" 2>/dev/null
|
||
}
|
||
|
||
delete_session() {
|
||
local sid="$1"
|
||
$SESSION_DELETE --server "${BASE}" --session-id "$sid" --force >/dev/null 2>&1 || true
|
||
}
|
||
|
||
stop_session() {
|
||
local sid="$1"
|
||
$SESSION_STOP --server "${BASE}" --session-id "$sid" >/dev/null 2>&1 || true
|
||
}
|
||
|
||
send_message_capture() {
|
||
local sid="$1"
|
||
local body="$2"
|
||
local out_file="$3"
|
||
|
||
# Refuse to send empty bodies — the server logs them as
|
||
# "No user message found in stream. This should never happen."
|
||
if [[ -z "$body" ]]; then
|
||
log_warn "send_message_capture: refusing to send empty body to ${sid}"
|
||
: > "$out_file" 2>/dev/null || true
|
||
return 1
|
||
fi
|
||
|
||
# Use prompt_async — fire-and-forget; server returns 204 immediately.
|
||
# The synchronous /message endpoint would block until the agent finishes
|
||
# its full response (potentially many minutes for a busy worker).
|
||
( trap '' INT
|
||
exec curl -sf -X POST "${BASE}/session/${sid}/prompt_async" \
|
||
-H "Content-Type: application/json" \
|
||
-d "$body" \
|
||
-w "\n"
|
||
) > "$out_file" 2>/dev/null &
|
||
CURL_PID=$!
|
||
|
||
while kill -0 "$CURL_PID" 2>/dev/null; do
|
||
wait "$CURL_PID" 2>/dev/null || true
|
||
done
|
||
CURL_PID=""
|
||
}
|
||
|
||
get_messages_poll() {
|
||
local sid="$1"
|
||
local limit="${2:-10}"
|
||
local timeout_secs="${3:-60}"
|
||
local deadline
|
||
deadline=$(($(date +%s) + timeout_secs))
|
||
|
||
while true; do
|
||
local msgs
|
||
msgs=$($SESSION_MESSAGES --server "${BASE}" --session-id "$sid" --limit "$limit" 2>/dev/null)
|
||
if [[ -n "$msgs" ]]; then
|
||
local has_assistant
|
||
has_assistant=$(echo "$msgs" | jq -r 'map(select(.info.role == "assistant")) | length')
|
||
if [[ "$has_assistant" -gt 0 ]]; then
|
||
echo "$msgs"
|
||
return 0
|
||
fi
|
||
fi
|
||
if [[ $(date +%s) -ge $deadline ]]; then
|
||
echo "[]"
|
||
return 1
|
||
fi
|
||
sleep 2
|
||
done
|
||
}
|
||
|
||
extract_last_assistant_text() {
|
||
local msgs="$1"
|
||
echo "$msgs" | jq -r '[.[] | select(.info.role == "assistant")] | last | if . == null then "" else (.parts // [] | map(select(.type == "text")) | .[-1].text // "") end'
|
||
}
|
||
|
||
extract_json_array() {
|
||
local text="$1"
|
||
echo "$text" | grep -oP '\[\s*[^\]]*\]' | tail -n1
|
||
}
|
||
|
||
# filter_json_array — strips any non-JSON prefix (e.g. progress messages from
|
||
# list scripts that write to stderr which ends up in stdout) and returns clean
|
||
# JSON. Returns "[]" if no JSON array is found. Use this after any command
|
||
# whose stderr might have leaked into stdout.
|
||
filter_json_array() {
|
||
local text="$1"
|
||
local first_char
|
||
first_char=$(echo "$text" | head -c1)
|
||
if [[ "$first_char" == "[" ]]; then
|
||
echo "$text"
|
||
else
|
||
local json
|
||
json=$(echo "$text" | grep -oP '(\[\s*\].*|\[\s*[^\]]+\])' | tail -n1)
|
||
if [[ -z "$json" || "$json" == "[]" ]]; then
|
||
echo "[]"
|
||
else
|
||
echo "$json"
|
||
fi
|
||
fi
|
||
}
|
||
|
||
# ── Question handling ─────────────────────────────────────────────────────
|
||
get_pending_questions() {
|
||
local result
|
||
run_timed result 15 curl -sf "${BASE}/question"
|
||
echo "$result"
|
||
}
|
||
|
||
answer_question() {
|
||
local qid="$1"
|
||
local text="$2"
|
||
curl -sf -X POST "${BASE}/question/${qid}/answer" \
|
||
-H "Content-Type: application/json" \
|
||
-d "{\"body\":\"${text}\"}" >/dev/null 2>&1
|
||
}
|
||
|
||
reject_question() {
|
||
local qid="$1"
|
||
curl -sf -X POST "${BASE}/question/${qid}/reject" >/dev/null 2>&1
|
||
}
|
||
|
||
# ── Forgejo configuration from environment + git remote fallback ─────────────
|
||
resolve_forgejo_config() {
|
||
local url="${FORGEJO_URL:-}"
|
||
local owner="${FORGEJO_OWNER:-}"
|
||
local repo="${FORGEJO_REPO:-}"
|
||
local pat="${FORGEJO_PAT:-}"
|
||
local git_name="${GIT_USER_NAME:-}"
|
||
local git_email="${GIT_USER_EMAIL:-}"
|
||
|
||
if [[ -z "$owner" || -z "$repo" ]]; then
|
||
local remote
|
||
remote=$(git remote get-url origin 2>/dev/null)
|
||
if [[ -n "$remote" ]]; then
|
||
local proto_host path_seg remote_owner remote_repo
|
||
proto_host=$(echo "$remote" | sed -n 's|^\(https\?://[^/]\+\).*|\1|p')
|
||
path_seg=$(echo "$remote" | sed -n 's|.*://[^/]\+/\([^/]\+/[^/]\+\).*|\1|p' | sed 's/\.git$//')
|
||
if [[ -z "$path_seg" ]]; then
|
||
log_warn "resolve_forgejo_config: could not parse owner/repo from git remote '${remote}'"
|
||
else
|
||
remote_owner="${path_seg%%/*}"
|
||
remote_repo="${path_seg#*/}"
|
||
url="${url:-${proto_host}}"
|
||
owner="${owner:-${remote_owner}}"
|
||
repo="${repo:-${remote_repo}}"
|
||
fi
|
||
else
|
||
log_warn "resolve_forgejo_config: FORGEJO_OWNER/REPO not set and 'git remote get-url origin' failed — forgejo config unavailable"
|
||
fi
|
||
fi
|
||
|
||
FORGEJO_URL="$url"
|
||
FORGEJO_OWNER="$owner"
|
||
FORGEJO_REPO="$repo"
|
||
FORGEJO_PAT="$pat"
|
||
GIT_USER_NAME="$git_name"
|
||
GIT_USER_EMAIL="$git_email"
|
||
}
|
||
|
||
# ── Work-group fetchers ────────────────────────────────────────────────────
|
||
fetch_list_script() {
|
||
local script_path="$1"
|
||
local forgejo_url="$2"
|
||
local forgejo_owner="$3"
|
||
local forgejo_repo="$4"
|
||
local forgejo_pat="$5"
|
||
|
||
if [[ -z "$forgejo_url" || -z "$forgejo_pat" ]]; then
|
||
echo "[]"
|
||
return
|
||
fi
|
||
|
||
local result
|
||
result=$(timeout "$SCRIPT_TIMEOUT_SEC" npx --yes tsx "$script_path" \
|
||
--url "$forgejo_url" \
|
||
--pat "$forgejo_pat" \
|
||
--owner "$forgejo_owner" \
|
||
--repo "$forgejo_repo" 2>/dev/null)
|
||
local status=$?
|
||
if [[ $status -eq 139 || $status -eq 137 || $status -eq 143 || $status -eq 124 ]]; then
|
||
log "[fetch_list_script] WARNING: $(basename "$script_path") timed out after ${SCRIPT_TIMEOUT_SEC}s (exit ${status})"
|
||
echo "[]"
|
||
return
|
||
fi
|
||
if [[ $status -ne 0 ]]; then
|
||
log "[fetch_list_script] WARNING: $(basename "$script_path") failed with exit ${status}"
|
||
echo "[]"
|
||
return
|
||
fi
|
||
local first_char="${result%"${result#?}"}"
|
||
if [[ "$first_char" == "[" ]]; then
|
||
echo "$result"
|
||
else
|
||
log "[fetch_list_script] WARNING: $(basename "$script_path") produced non-JSON output (starts with '${first_char}'); treating as empty"
|
||
echo "[]"
|
||
fi
|
||
}
|
||
|
||
# ── Worker prompt builders ─────────────────────────────────────────────────
|
||
# All fields (title, body) have newlines collapsed so the prompt file stays
|
||
# valid as a single-line-ish text file. The full item JSON is embedded compact
|
||
# (jq -c . → one line, no literal newlines) at the end of the prompt.
|
||
# session_start.ts reads this file as plain text and sends it to the agent.
|
||
build_impl_prompt() {
|
||
local item_json="$1"
|
||
local kind="$2"
|
||
local work_num="$3"
|
||
|
||
local work_type; work_type=$([ "$kind" == "ISSUE" ] && echo "issue_impl" || echo "pr_fix")
|
||
local compact_item
|
||
compact_item=$(echo "$item_json" | jq -c .)
|
||
|
||
jq -n \
|
||
--arg wn "$work_num" \
|
||
--arg wt "$work_type" \
|
||
--argjson ij "$compact_item" \
|
||
'{
|
||
issue_number: $wn,
|
||
pr_number: $wn,
|
||
work_type: $wt,
|
||
item_json: $ij
|
||
}'
|
||
}
|
||
|
||
build_groom_prompt() {
|
||
local item_json="$1"
|
||
local kind="$2"
|
||
local work_num="$3"
|
||
|
||
local work_type; work_type=$([ "$kind" == "ISSUE" ] && echo "issue_groom" || echo "pr_groom")
|
||
local compact_item
|
||
compact_item=$(echo "$item_json" | jq -c .)
|
||
|
||
jq -n \
|
||
--arg wn "$work_num" \
|
||
--arg wt "$work_type" \
|
||
--argjson ij "$compact_item" \
|
||
'{
|
||
issue_number: $wn,
|
||
pr_number: $wn,
|
||
work_type: $wt,
|
||
item_json: $ij
|
||
}'
|
||
}
|
||
|
||
build_merge_prompt() {
|
||
local pr_json="$1"
|
||
local pr_num="$2"
|
||
|
||
local compact_item
|
||
compact_item=$(echo "$pr_json" | jq -c .)
|
||
|
||
jq -n \
|
||
--arg pn "$pr_num" \
|
||
--argjson ij "$compact_item" \
|
||
'{
|
||
pr_number: $pn,
|
||
item_json: $ij
|
||
}'
|
||
}
|
||
|
||
build_review_prompt() {
|
||
local pr_json="$1"
|
||
local pr_num="$2"
|
||
|
||
local title head_sha base_sha merge_base is_stale has_conflicts
|
||
title=$(echo "$pr_json" | jq -r '.title' | tr '\n' ' ' | sed 's/[[:space:]]\+/ /g')
|
||
head_sha=$(echo "$pr_json" | jq -r '.head.sha')
|
||
base_sha=$(echo "$pr_json" | jq -r '.base.sha')
|
||
merge_base=$(echo "$pr_json" | jq -r '.merge_base // empty')
|
||
[[ "$merge_base" == "empty" || -z "$merge_base" || "$merge_base" == "null" ]] && merge_base="$base_sha"
|
||
is_stale=$(echo "$pr_json" | jq -r 'if (.stale_state // "not_stale") == "not_stale" then "false" else "true" end')
|
||
has_conflicts=$(echo "$pr_json" | jq -r 'if .mergeable == false then "true" else "false" end')
|
||
|
||
local approvals_count ci_status compact_item
|
||
approvals_count=$(echo "$pr_json" | jq -r '.approvals_count // 0')
|
||
ci_status=$(echo "$pr_json" | jq -r '.ci_status // "unknown"')
|
||
compact_item=$(echo "$pr_json" | jq -c .)
|
||
|
||
cat <<PROMPT
|
||
pr_number: ${pr_num}
|
||
pr_title: ${title}
|
||
head_sha: ${head_sha}
|
||
base_sha: ${base_sha}
|
||
merge_base_sha: ${merge_base}
|
||
is_stale: ${is_stale}
|
||
has_conflicts: ${has_conflicts}
|
||
approvals_count: ${approvals_count}
|
||
ci_status: ${ci_status}
|
||
item_json: ${compact_item}
|
||
|
||
Process the indicated Pull Request or Issue.
|
||
PROMPT
|
||
}
|
||
|
||
# ── Fleet analysis ────────────────────────────────────────────────────────
|
||
get_busy_worker_tags() {
|
||
local prefix="$1"
|
||
find_sessions_by_prefix "$prefix" "busy" | jq -r '.[].tag // empty'
|
||
}
|
||
|
||
extract_work_number() {
|
||
local tag="$1"
|
||
echo "$tag" | sed -n 's/.*-\(PR\|ISSUE\)-\([0-9]\+\)$/\2/p'
|
||
}
|
||
|
||
# ── Duplicate worker guard ────────────────────────────────────────────────
|
||
deduplicate_workers() {
|
||
local prefix="$1"
|
||
local workers_json
|
||
workers_json=$(find_sessions_by_prefix "$prefix" "busy")
|
||
if [[ -z "$workers_json" ]]; then
|
||
log_warn "[deduplicate_workers:${prefix}] find_sessions_by_prefix returned empty (timeout or error) — skipping deduplication"
|
||
return
|
||
fi
|
||
local duplicates
|
||
duplicates=$(echo "$workers_json" | jq -r '
|
||
group_by(.tag)
|
||
| map(select(length > 1))
|
||
| map(sort_by(.last_active) | reverse | .[1:])
|
||
| flatten
|
||
| .[].id
|
||
' 2>/dev/null)
|
||
if [[ -n "$duplicates" ]]; then
|
||
local count
|
||
count=$(echo "$duplicates" | grep -c . 2>/dev/null || echo 0)
|
||
log "[deduplicate_workers:${prefix}] found ${count} duplicate session(s) — deleting older copies"
|
||
while IFS= read -r dup_id; do
|
||
[[ -n "$dup_id" ]] || continue
|
||
log " deleting duplicate session ${dup_id}"
|
||
delete_session "$dup_id"
|
||
done <<< "$duplicates"
|
||
fi
|
||
}
|
||
|
||
# ── Disk cleanup ──────────────────────────────────────────────────────────
|
||
cleanup_orphan_tmp_dirs() {
|
||
local prefix="$1"
|
||
local sessions_json
|
||
sessions_json=$(find_sessions_by_prefix "$prefix" "all")
|
||
if [[ -z "$sessions_json" ]]; then
|
||
log_warn "[cleanup_orphan_tmp_dirs:${prefix}] find_sessions_by_prefix failed (timeout/error) — skipping disk cleanup"
|
||
return
|
||
fi
|
||
local active_tags
|
||
active_tags=$(echo "$sessions_json" | jq -r '.[].tag // empty' 2>/dev/null | sort -u)
|
||
# Note: active_tags can legitimately be empty when there are zero sessions
|
||
# for this prefix — in that case every matching tmp dir is an orphan.
|
||
|
||
local tag_pattern="${prefix}"
|
||
local find_result
|
||
run_timed find_result 10 find /tmp -maxdepth 1 -type d -name "*${tag_pattern}*" -print0 2>/dev/null
|
||
if [[ -z "$find_result" ]]; then
|
||
log "[cleanup_orphan_tmp_dirs:${prefix}] no tmp dirs found matching ${tag_pattern}"
|
||
return
|
||
fi
|
||
|
||
local removed=0
|
||
while IFS= read -r -d '' dir; do
|
||
[[ -d "$dir" ]] || continue
|
||
local dir_id
|
||
dir_id=$(basename "$dir" | grep -oP '\d+' | tail -n1)
|
||
if [[ -z "$dir_id" ]]; then
|
||
continue
|
||
fi
|
||
local is_active=false
|
||
while IFS= read -r atag; do
|
||
[[ -n "$atag" ]] || continue
|
||
if [[ "$atag" == *"${dir_id}"* ]]; then
|
||
is_active=true
|
||
break
|
||
fi
|
||
done <<< "$active_tags"
|
||
if ! $is_active; then
|
||
log "[cleanup_orphan_tmp_dirs:${prefix}] removing orphan ${dir}"
|
||
rm -rf "$dir"
|
||
((removed++))
|
||
fi
|
||
done < <(printf '%s\0' "$find_result")
|
||
|
||
if [[ $removed -eq 0 ]]; then
|
||
log "[cleanup_orphan_tmp_dirs:${prefix}] no orphan tmp dirs found"
|
||
else
|
||
log "[cleanup_orphan_tmp_dirs:${prefix}] removed ${removed} orphan tmp dir(s)"
|
||
fi
|
||
}
|
||
|
||
# run_single_doom_check — Evaluate the health of a single busy session by
|
||
# spinning up a worker-health-evaluator eval session, waiting for it to
|
||
# finish (up to 120s), and printing one of:
|
||
# "unrecoverable" — eval reports unrecoverable=true; caller should delete
|
||
# "ok" — anything else; session is fine
|
||
# "create_failed" — couldn't create the eval session
|
||
# Designed to be run in the background so many checks can proceed in parallel.
|
||
run_single_doom_check() {
|
||
local sid="$1"
|
||
local eval_tag="BUILDER-HEALTH-${sid: -6}"
|
||
local eval_title="[${eval_tag}] worker-health-evaluator"
|
||
local eval_prompt="session_id: ${sid}
|
||
|
||
Evaluate the health of this session."
|
||
local eval_session_json
|
||
eval_session_json=$(create_session "$eval_title")
|
||
local eval_sid
|
||
eval_sid=$(echo "$eval_session_json" | jq -r '.id')
|
||
if [[ -z "$eval_sid" || "$eval_sid" == "null" ]]; then
|
||
echo "create_failed"
|
||
return
|
||
fi
|
||
local eval_tmp="/tmp/builder-prompt-${eval_tag}-$$.txt"
|
||
printf '%s' "$eval_prompt" > "$eval_tmp"
|
||
local eval_body
|
||
eval_body=$(jq -nc --arg a "worker-health-evaluator" --rawfile t "$eval_tmp" '{agent: $a, parts: [{type: "text", text: $t}]}')
|
||
if [[ -z "$eval_body" ]]; then
|
||
# jq failed (likely couldn't read eval_tmp) — bail rather than send empty
|
||
rm -f "$eval_tmp"
|
||
delete_session "$eval_sid"
|
||
echo "ok"
|
||
return
|
||
fi
|
||
curl -sf -X POST "${BASE}/session/${eval_sid}/prompt_async" -H "Content-Type: application/json" -d "$eval_body" >/dev/null 2>&1
|
||
local eval_deadline=$(( $(date +%s) + 120 ))
|
||
while [[ $(date +%s) -lt $eval_deadline ]]; do
|
||
local eval_status
|
||
eval_status=$(curl -sf "${BASE}/session/status" 2>/dev/null | jq -r --arg sid "$eval_sid" '.[$sid].type // "idle"')
|
||
[[ "$eval_status" == "idle" ]] && break
|
||
sleep 2
|
||
done
|
||
local eval_msgs
|
||
eval_msgs=$($SESSION_MESSAGES --server "${BASE}" --session-id "$eval_sid" --limit 10 2>/dev/null)
|
||
local eval_text
|
||
eval_text=$(extract_last_assistant_text "$eval_msgs")
|
||
delete_session "$eval_sid"
|
||
rm -f "$eval_tmp"
|
||
local is_unrecoverable
|
||
is_unrecoverable=$(echo "$eval_text" | jq -r '.unrecoverable // false' 2>/dev/null || echo "false")
|
||
if [[ "$is_unrecoverable" == "true" ]]; then
|
||
echo "unrecoverable"
|
||
else
|
||
echo "ok"
|
||
fi
|
||
}
|
||
|
||
# ── Health checks ─────────────────────────────────────────────────────────
|
||
run_health_cycle() {
|
||
local now
|
||
now=$(date +%s)
|
||
local did_something=false
|
||
|
||
# 1) Question handling
|
||
local questions_json
|
||
questions_json=$(get_pending_questions)
|
||
if [[ -n "$questions_json" && "$questions_json" != "[]" && "$questions_json" != "null" ]]; then
|
||
local qids sids
|
||
qids=$(echo "$questions_json" | jq -r '.[].id')
|
||
sids=$(echo "$questions_json" | jq -r '.[].sessionID')
|
||
while IFS= read -r qid && IFS= read -r sid <&3; do
|
||
[[ -n "$qid" ]] || continue
|
||
[[ -n "$sid" ]] || continue
|
||
local cnt="${QUESTION_COUNT[$sid]:-0}"
|
||
if [[ "$cnt" -eq 0 ]]; then
|
||
log "Question from ${sid} (first time) — answering with anti-question text"
|
||
answer_question "$qid" "you are never under any circumstances to ask questions or pass along questions from subagents, use your best judgement and do not ask questions again"
|
||
QUESTION_COUNT[$sid]=1
|
||
else
|
||
log "Question from ${sid} (repeat) — deleting session"
|
||
delete_session "$sid"
|
||
unset QUESTION_COUNT[$sid]
|
||
fi
|
||
did_something=true
|
||
done <<< "$qids" 3<<< "$sids"
|
||
fi
|
||
|
||
# 2) Idle worker check — collect all idle SIDs across prefixes, then fetch
|
||
# their messages in parallel before processing each one.
|
||
local all_prefixes=("AUTO-IMP" "AUTO-MRG" "AUTO-REV" "AUTO-GROOM")
|
||
local -a idle_sids=()
|
||
for prefix in "${all_prefixes[@]}"; do
|
||
local sessions_json
|
||
sessions_json=$(find_sessions_by_prefix "$prefix" "all")
|
||
if [[ -z "$sessions_json" ]]; then
|
||
log_warn "[health:idle-check:${prefix}] find_sessions_by_prefix returned empty (timeout or error) — skipping idle check for this pool"
|
||
continue
|
||
fi
|
||
local idle_workers
|
||
idle_workers=$(echo "$sessions_json" | jq -r \
|
||
--arg cutoff "$(( (now * 1000) - IDLE_THRESHOLD_MS ))" \
|
||
'.[] | select(.last_active < ($cutoff | tonumber)) | .id // empty')
|
||
while IFS= read -r sid; do
|
||
[[ -n "$sid" ]] || continue
|
||
idle_sids+=("$sid")
|
||
done <<< "$idle_workers"
|
||
done
|
||
|
||
if [[ ${#idle_sids[@]} -gt 0 ]]; then
|
||
log "Idle check: fetching messages for ${#idle_sids[@]} idle worker(s) in parallel"
|
||
local -a msg_pids=()
|
||
local -a msg_files=()
|
||
for sid in "${idle_sids[@]}"; do
|
||
local msg_file="/tmp/idle-msgs-${sid}-$$.json"
|
||
( $SESSION_MESSAGES --server "${BASE}" --session-id "$sid" --limit 10 > "$msg_file" 2>/dev/null ) &
|
||
msg_pids+=($!)
|
||
msg_files+=("$msg_file")
|
||
done
|
||
for pid in "${msg_pids[@]}"; do
|
||
wait "$pid" 2>/dev/null
|
||
done
|
||
|
||
# Sequential processing (CONTINUE_TIMES is shared state)
|
||
for ((i = 0; i < ${#idle_sids[@]}; i++)); do
|
||
local sid="${idle_sids[i]}"
|
||
local msgs
|
||
msgs=$(cat "${msg_files[i]}" 2>/dev/null)
|
||
rm -f "${msg_files[i]}"
|
||
if [[ -z "$msgs" || "$msgs" == "[]" ]]; then
|
||
log "Idle worker ${sid} has no messages — deleting"
|
||
delete_session "$sid"
|
||
did_something=true
|
||
continue
|
||
fi
|
||
local last_text
|
||
last_text=$(extract_last_assistant_text "$msgs")
|
||
if echo "$last_text" | grep -qiE '(done|finished|complete|exiting|all tests pass|committed|merged|review submitted)'; then
|
||
log "Idle worker ${sid} appears finished — deleting"
|
||
delete_session "$sid"
|
||
did_something=true
|
||
continue
|
||
fi
|
||
local now_s
|
||
now_s=$(date +%s)
|
||
local times_str="${CONTINUE_TIMES[$sid]:-}"
|
||
local cutoff=$((now_s - CONTINUE_BURST_WINDOW))
|
||
local new_times_arr=()
|
||
local count=0
|
||
if [[ -n "$times_str" ]]; then
|
||
for ts in $times_str; do
|
||
[[ -n "$ts" ]] || continue
|
||
if (( ts > cutoff )); then
|
||
new_times_arr+=("$ts")
|
||
((count++))
|
||
fi
|
||
done
|
||
fi
|
||
new_times_arr+=("$now_s")
|
||
((count++))
|
||
CONTINUE_TIMES[$sid]=$(printf '%s ' "${new_times_arr[@]}")
|
||
if [[ "$count" -gt $CONTINUE_BURST_LIMIT ]]; then
|
||
log "Idle worker ${sid} received ${count} continues in ${CONTINUE_BURST_WINDOW}s — deleting"
|
||
delete_session "$sid"
|
||
unset CONTINUE_TIMES[$sid]
|
||
did_something=true
|
||
else
|
||
log "Idle worker ${sid} — sending continue"
|
||
local cont_body
|
||
cont_body=$(jq -nc '{agent: "", parts: [{type: "text", text: "continue"}]}')
|
||
send_message_capture "$sid" "$cont_body" "/tmp/continue-${sid}.txt"
|
||
did_something=true
|
||
fi
|
||
done
|
||
fi
|
||
|
||
# 3) Doom-loop check (every 15 minutes) — runs ALL evaluations in parallel
|
||
if [[ $((now - LAST_DOOM_CHECK)) -ge $DOOM_CHECK_INTERVAL ]]; then
|
||
LAST_DOOM_CHECK=$now
|
||
|
||
# Collect all busy sessions across all prefixes.
|
||
local -a doom_sids=()
|
||
for prefix in "${all_prefixes[@]}"; do
|
||
local sessions_json busy_workers
|
||
sessions_json=$(find_sessions_by_prefix "$prefix" "all")
|
||
if [[ -z "$sessions_json" ]]; then
|
||
log_warn "[health:doom-check:${prefix}] find_sessions_by_prefix returned empty (timeout or error) — skipping doom check for this pool"
|
||
continue
|
||
fi
|
||
busy_workers=$(echo "$sessions_json" | jq -r \
|
||
--arg cutoff "$(( (now - DOOM_AGE_THRESHOLD) * 1000 ))" \
|
||
'.[] | select(.last_active >= ($cutoff | tonumber)) | .id // empty')
|
||
while IFS= read -r sid; do
|
||
[[ -n "$sid" ]] || continue
|
||
doom_sids+=("$sid")
|
||
done <<< "$busy_workers"
|
||
done
|
||
|
||
if [[ ${#doom_sids[@]} -gt 0 ]]; then
|
||
log "Doom-loop check: evaluating ${#doom_sids[@]} session(s) in parallel"
|
||
local -a doom_pids=()
|
||
local -a doom_result_files=()
|
||
for sid in "${doom_sids[@]}"; do
|
||
local result_file="/tmp/doom-result-${sid}-$$.txt"
|
||
( run_single_doom_check "$sid" > "$result_file" 2>/dev/null ) &
|
||
doom_pids+=($!)
|
||
doom_result_files+=("$result_file")
|
||
done
|
||
|
||
# Wait for all evaluations and process results
|
||
for ((i = 0; i < ${#doom_pids[@]}; i++)); do
|
||
wait "${doom_pids[i]}" 2>/dev/null
|
||
local result
|
||
result=$(cat "${doom_result_files[i]}" 2>/dev/null)
|
||
if [[ "$result" == "unrecoverable" ]]; then
|
||
log "Session ${doom_sids[i]} is unrecoverable — deleting"
|
||
delete_session "${doom_sids[i]}"
|
||
did_something=true
|
||
fi
|
||
rm -f "${doom_result_files[i]}"
|
||
done
|
||
fi
|
||
fi
|
||
|
||
if $did_something; then
|
||
return 0
|
||
else
|
||
return 1
|
||
fi
|
||
}
|
||
|
||
# ── Preflight checks ─────────────────────────────────────────────────────
|
||
for cmd in curl jq opencode npx; do
|
||
command -v "$cmd" &>/dev/null || die "'$cmd' is required but not found in PATH"
|
||
done
|
||
|
||
# ── Start the opencode server ─────────────────────────────────────────────
|
||
if check_opencode_running; then
|
||
echo "┌─────────────────────────────────────────────────────────────────┐"
|
||
echo "│ ⚠ WARNING: OpenCode server already running on ${HOST}:${PORT}"
|
||
echo "│ 🔗 Connecting to existing server instead of starting new one"
|
||
echo "│ 💡 The script will NOT manage this server's lifecycle"
|
||
echo "└─────────────────────────────────────────────────────────────────┘"
|
||
echo ""
|
||
log "Using existing OpenCode server at ${HOST}:${PORT}"
|
||
OWN_SERVER=false
|
||
else
|
||
log "Starting opencode server on ${HOST}:${PORT}..."
|
||
opencode serve --port "$PORT" --hostname "$HOST" &
|
||
SERVER_PID=$!
|
||
OWN_SERVER=true
|
||
|
||
log "Waiting for server to become healthy (up to ${HEALTH_TIMEOUT}s)..."
|
||
healthy=false
|
||
for (( i = 1; i <= HEALTH_TIMEOUT; i++ )); do
|
||
if curl -sf "${BASE}/global/health" >/dev/null 2>&1; then
|
||
healthy=true
|
||
break
|
||
fi
|
||
kill -0 "$SERVER_PID" 2>/dev/null || die "Server process exited unexpectedly"
|
||
sleep 1
|
||
done
|
||
$healthy || die "Server did not become healthy within ${HEALTH_TIMEOUT}s"
|
||
log "Server is ready."
|
||
fi
|
||
|
||
# ── Main orchestration loop ───────────────────────────────────────────────
|
||
main_loop() {
|
||
log "Starting orchestration loop."
|
||
log "Worker pools: merge=${merge_max_workers} imp=${imp_max_workers} rev=${rev_max_workers} groom=${groom_max_workers}"
|
||
|
||
while ! $STOP_LOOP; do
|
||
N=$((N + 1))
|
||
|
||
idle_count=$(count_idle_events_last_minute)
|
||
if (( idle_count > MAX_IDLE_PER_MINUTE )); then
|
||
log "── iteration #${N} ── TOO MANY IDLE EVENTS (${idle_count}) — skipping cycle"
|
||
sleep 5
|
||
continue
|
||
fi
|
||
|
||
log "── iteration #${N} ──"
|
||
local did_anything=false
|
||
|
||
local all_sessions_json
|
||
all_sessions_json=$(query_all_sessions)
|
||
if [[ -z "$all_sessions_json" ]]; then
|
||
if is_timeout "${RUN_TIMED_EXITCODE:-0}"; then
|
||
log_warn "query_all_sessions timed out — OpenCode server may be unresponsive; skipping this cycle"
|
||
sleep "$SLEEP_WHEN_IDLE"
|
||
continue
|
||
fi
|
||
log "No sessions found."
|
||
fi
|
||
|
||
# ── Step 2: Per-type analysis & direct worker launch ──────────────────
|
||
log "[step2] resolving Forgejo config..."
|
||
resolve_forgejo_config
|
||
log "[step2] Forgejo config: url=${FORGEJO_URL} owner=${FORGEJO_OWNER} repo=${FORGEJO_REPO} pat_set=$([[ -n "$FORGEJO_PAT" ]] && echo yes || echo no)"
|
||
|
||
if [[ -z "$FORGEJO_PAT" ]]; then
|
||
log "FORGEJO_PAT not set — skipping worker launch"
|
||
else
|
||
declare -a IMP_WORK_GROUPS=(
|
||
"CI Failing PRs|${SCRIPT_DIR}/list_prs_ci_failing.ts|PR"
|
||
"Addressed Changes CI Passing|${SCRIPT_DIR}/list_prs_addressed_changes_ci_passing.ts|PR"
|
||
"Addressed Changes CI Failing|${SCRIPT_DIR}/list_prs_addressed_changes_ci_failing.ts|PR"
|
||
"Changes Requested|${SCRIPT_DIR}/list_prs_changes_requested.ts|PR"
|
||
"CI Missing|${SCRIPT_DIR}/list_prs_missing_ci_checks.ts|PR"
|
||
"No Active Review CI Passing|${SCRIPT_DIR}/list_prs_no_active_review_ci_passing.ts|PR"
|
||
"No Active Review CI Failing|${SCRIPT_DIR}/list_prs_no_active_review_ci_failing.ts|PR"
|
||
"Needs Review Not Stale|${SCRIPT_DIR}/list_prs_needs_review_not_stale.ts|PR"
|
||
"Needs Review Stale Clean|${SCRIPT_DIR}/list_prs_needs_review_stale_clean.ts|PR"
|
||
"Needs Review Stale Conflicts|${SCRIPT_DIR}/list_prs_needs_review_stale_conflicts.ts|PR"
|
||
"Open Issues|${SCRIPT_DIR}/list_issues.ts|ISSUE"
|
||
)
|
||
declare -a MRG_WORK_GROUPS=(
|
||
"Ready to Merge|${SCRIPT_DIR}/list_prs_ready_to_merge.ts|PR"
|
||
"Stale No Conflicts|${SCRIPT_DIR}/list_prs_stale_clean.ts|PR"
|
||
"Stale Has Conflicts|${SCRIPT_DIR}/list_prs_stale_conflicts.ts|PR"
|
||
)
|
||
declare -a REV_WORK_GROUPS=(
|
||
"No Active Review CI Passing|${SCRIPT_DIR}/list_prs_no_active_review_ci_passing.ts|PR"
|
||
"No Active Review CI Failing|${SCRIPT_DIR}/list_prs_no_active_review_ci_failing.ts|PR"
|
||
"Needs Review Not Stale|${SCRIPT_DIR}/list_prs_needs_review_not_stale.ts|PR"
|
||
"Needs Review Stale Clean|${SCRIPT_DIR}/list_prs_needs_review_stale_clean.ts|PR"
|
||
"Needs Review Stale Conflicts|${SCRIPT_DIR}/list_prs_needs_review_stale_conflicts.ts|PR"
|
||
"Addressed Changes CI Passing|${SCRIPT_DIR}/list_prs_addressed_changes_ci_passing.ts|PR"
|
||
"Addressed Changes CI Failing|${SCRIPT_DIR}/list_prs_addressed_changes_ci_failing.ts|PR"
|
||
"Changes Requested|${SCRIPT_DIR}/list_prs_changes_requested.ts|PR"
|
||
"CI Failing|${SCRIPT_DIR}/list_prs_ci_failing.ts|PR"
|
||
)
|
||
# GROOM mirrors IMP: same priority order (PRs first, issues second), same
|
||
# work groups. Re-uses IMP's cached results within the same iteration
|
||
# via the shared cache_dir, so grooming pre-fetch is effectively free.
|
||
declare -a GROOM_WORK_GROUPS=(
|
||
"CI Failing PRs|${SCRIPT_DIR}/list_prs_ci_failing.ts|PR"
|
||
"Addressed Changes CI Passing|${SCRIPT_DIR}/list_prs_addressed_changes_ci_passing.ts|PR"
|
||
"Addressed Changes CI Failing|${SCRIPT_DIR}/list_prs_addressed_changes_ci_failing.ts|PR"
|
||
"Changes Requested|${SCRIPT_DIR}/list_prs_changes_requested.ts|PR"
|
||
"CI Missing|${SCRIPT_DIR}/list_prs_missing_ci_checks.ts|PR"
|
||
"No Active Review CI Passing|${SCRIPT_DIR}/list_prs_no_active_review_ci_passing.ts|PR"
|
||
"No Active Review CI Failing|${SCRIPT_DIR}/list_prs_no_active_review_ci_failing.ts|PR"
|
||
"Needs Review Not Stale|${SCRIPT_DIR}/list_prs_needs_review_not_stale.ts|PR"
|
||
"Needs Review Stale Clean|${SCRIPT_DIR}/list_prs_needs_review_stale_clean.ts|PR"
|
||
"Needs Review Stale Conflicts|${SCRIPT_DIR}/list_prs_needs_review_stale_conflicts.ts|PR"
|
||
"Open Issues|${SCRIPT_DIR}/list_issues.ts|ISSUE"
|
||
)
|
||
|
||
declare -a pools=(
|
||
"AUTO-IMP|implementation-worker|${imp_max_workers}|IMP_WORK_GROUPS"
|
||
"AUTO-MRG|pr-merge-worker|${merge_max_workers}|MRG_WORK_GROUPS"
|
||
"AUTO-REV|pr-review-worker|${rev_max_workers}|REV_WORK_GROUPS"
|
||
"AUTO-GROOM|grooming-worker|${groom_max_workers}|GROOM_WORK_GROUPS"
|
||
)
|
||
|
||
# Shared cache dir for this iteration — pools reuse each other's results
|
||
# (e.g. REV's scripts are all a subset of IMP's, so REV gets them for free)
|
||
local cache_dir="/tmp/opencode-builder-cache-$$/iter-${N}"
|
||
mkdir -p "$cache_dir"
|
||
|
||
for pool_spec in "${pools[@]}"; do
|
||
IFS='|' read -r prefix agent max wg_array_name <<< "$pool_spec"
|
||
|
||
log "[pool:${prefix}] starting pool processing (agent=${agent}, max=${max})"
|
||
log "[pool:${prefix}] deduplicate_workers..."
|
||
deduplicate_workers "$prefix"
|
||
log "[pool:${prefix}] cleanup_orphan_tmp_dirs..."
|
||
cleanup_orphan_tmp_dirs "$prefix"
|
||
|
||
local exclude_numbers=","
|
||
log "[pool:${prefix}] find_sessions_by_prefix '${prefix}' 'all'..."
|
||
local all_sessions_json
|
||
all_sessions_json=$(find_sessions_by_prefix "$prefix" "all")
|
||
if [[ -z "$all_sessions_json" ]]; then
|
||
if is_timeout "${RUN_TIMED_EXITCODE:-0}"; then
|
||
log_warn "[pool:${prefix}] find_sessions_by_prefix timed out after ${SESSION_SCRIPT_TIMEOUT}s — skipping pool"
|
||
continue
|
||
fi
|
||
log_warn "[pool:${prefix}] find_sessions_by_prefix returned empty — proceeding with 0 known sessions"
|
||
all_sessions_json="[]"
|
||
fi
|
||
local total_count
|
||
total_count=$(echo "$all_sessions_json" | jq 'length' 2>/dev/null || echo 0)
|
||
log "[pool:${prefix}] total sessions found: ${total_count}"
|
||
|
||
log "[pool:${prefix}] get_busy_worker_tags..."
|
||
local busy_tags
|
||
busy_tags=$(get_busy_worker_tags "$prefix")
|
||
local busy_count=0
|
||
while IFS= read -r tag; do
|
||
[[ -n "$tag" ]] || continue
|
||
local num
|
||
num=$(echo "$tag" | sed -n 's/.*-\(PR\|ISSUE\)-\([0-9]\+\)$/\2/p')
|
||
if [[ -n "$num" ]]; then
|
||
exclude_numbers="${exclude_numbers}${num},"
|
||
((busy_count++))
|
||
fi
|
||
done <<< "$busy_tags"
|
||
log "[pool:${prefix}] busy_count=${busy_count} (from ${total_count} sessions)"
|
||
|
||
local available_slots=$((max - busy_count))
|
||
if [[ "$available_slots" -le 0 ]]; then
|
||
log "Pool ${prefix}: ${busy_count}/${max} active — full, skipping"
|
||
continue
|
||
fi
|
||
|
||
log "Pool ${prefix}: ${busy_count}/${max} active (${total_count} total sessions, ${available_slots} slots free)"
|
||
did_anything=true
|
||
|
||
local wg_array_ref="${wg_array_name}[@]"
|
||
local work_groups=("${!wg_array_ref}")
|
||
|
||
# ── PRE-FETCH: launch ALL list scripts in parallel, cache results ────
|
||
# Each list_prs_*.ts script fetches ALL open PRs from Forgejo (~90s) and
|
||
# applies a filter. Running them sequentially = 11×90s = 990s.
|
||
# Running them in parallel = ~90s total (one round of Forgejo API calls).
|
||
log "[pool:${prefix}] pre-fetching ${#work_groups[@]} work-group scripts in parallel..."
|
||
local -A script_cache
|
||
cache_pids=()
|
||
cache_keys=()
|
||
|
||
for type_spec in "${work_groups[@]}"; do
|
||
IFS='|' read -r wg_label wg_scripts wg_kind <<< "$type_spec"
|
||
local scripts_arr
|
||
IFS=',' read -ra scripts_arr <<< "$wg_scripts"
|
||
for script_path in "${scripts_arr[@]}"; do
|
||
[[ -n "$script_path" ]] || continue
|
||
local cache_key="${script_path##*/}" # just filename
|
||
cache_key="${cache_key%.ts}"
|
||
local out_file="${cache_dir}/${cache_key}.json"
|
||
# Skip if a prior pool in this iteration already fetched this script
|
||
# and produced a valid JSON array.
|
||
if [[ -s "$out_file" ]] && [[ "$(head -c1 "$out_file" 2>/dev/null)" == "[" ]]; then
|
||
log "[pool:${prefix}] $(basename "$script_path"): reusing cached result from this iteration"
|
||
continue
|
||
fi
|
||
cache_keys+=("$cache_key")
|
||
log "[pool:${prefix}] launching $(basename "$script_path") (cache key: ${cache_key})..."
|
||
timeout "$SCRIPT_TIMEOUT_SEC" npx --yes tsx "$script_path" \
|
||
--url "$FORGEJO_URL" \
|
||
--pat "$FORGEJO_PAT" \
|
||
--owner "$FORGEJO_OWNER" \
|
||
--repo "$FORGEJO_REPO" \
|
||
> "$out_file" 2> "${out_file}.err" &
|
||
cache_pids+=($!)
|
||
done
|
||
done
|
||
|
||
log "[pool:${prefix}] waiting for ${#cache_pids[@]} pre-fetch jobs to complete..."
|
||
local cache_idx=0
|
||
local timed_out=0
|
||
local failed=0
|
||
for pid in "${cache_pids[@]}"; do
|
||
wait "$pid" 2>/dev/null
|
||
local ec=$?
|
||
local ck="${cache_keys[$cache_idx]}"
|
||
local out_file="${cache_dir}/${ck}.json"
|
||
if is_timeout $ec; then
|
||
log_warn "[pool:${prefix}] ${ck}: TIMEOUT after ${SCRIPT_TIMEOUT_SEC}s — treating as empty"
|
||
echo "[]" > "$out_file"
|
||
((timed_out++))
|
||
elif [[ $ec -ne 0 ]]; then
|
||
log_warn "[pool:${prefix}] ${ck}: FAILED (exit ${ec}) — treating as empty"
|
||
local snippet
|
||
snippet=$(head -n 3 "${out_file}.err" 2>/dev/null | tr '\n' '|' | sed 's/|$//')
|
||
if [[ -n "$snippet" ]]; then
|
||
log_warn " stderr: ${snippet}"
|
||
fi
|
||
echo "[]" > "$out_file"
|
||
((failed++))
|
||
elif [[ -s "$out_file" ]]; then
|
||
local first_char
|
||
first_char=$(head -c1 "$out_file" 2>/dev/null)
|
||
if [[ "$first_char" == "[" ]]; then
|
||
log "[pool:${prefix}] ${ck}: ready ($(wc -l < "$out_file" | tr -d ' ') lines)"
|
||
else
|
||
log_warn "[pool:${prefix}] ${ck}: produced non-JSON output (starts with '${first_char}') — treating as empty"
|
||
echo "[]" > "$out_file"
|
||
((failed++))
|
||
fi
|
||
else
|
||
log_warn "[pool:${prefix}] ${ck}: empty output with no error — treating as empty"
|
||
echo "[]" > "$out_file"
|
||
fi
|
||
((cache_idx++))
|
||
done
|
||
if (( timed_out > 0 || failed > 0 )); then
|
||
log_warn "[pool:${prefix}] pre-fetch summary: ${timed_out} timed out, ${failed} failed (of ${#cache_pids[@]} total)"
|
||
else
|
||
log "[pool:${prefix}] pre-fetch complete: all ${#cache_pids[@]} scripts succeeded"
|
||
fi
|
||
|
||
# ── Worker selection: iterate cached work-groups ─────────────────────
|
||
local launched=0
|
||
for type_spec in "${work_groups[@]}"; do
|
||
if (( launched >= available_slots )); then
|
||
break
|
||
fi
|
||
|
||
IFS='|' read -r wg_label wg_scripts wg_kind <<< "$type_spec"
|
||
log "[pool:${prefix}] work-group '${wg_label}': selecting workers..."
|
||
|
||
local scripts_arr
|
||
IFS=',' read -ra scripts_arr <<< "$wg_scripts"
|
||
for script_path in "${scripts_arr[@]}"; do
|
||
if (( launched >= available_slots )); then
|
||
break 2
|
||
fi
|
||
[[ -n "$script_path" ]] || continue
|
||
|
||
local cache_key="${script_path##*/}"
|
||
cache_key="${cache_key%.ts}"
|
||
local out_file="${cache_dir}/${cache_key}.json"
|
||
local json
|
||
json=$(cat "$out_file" 2>/dev/null || echo "[]")
|
||
|
||
local raw_count
|
||
raw_count=$(echo "$json" | jq 'length' 2>/dev/null || echo 0)
|
||
log "[pool:${prefix}] script $(basename "$script_path"): ${raw_count} items in cache"
|
||
|
||
if [[ "$raw_count" -eq 0 || "$raw_count" == "null" || "$raw_count" == "" ]]; then
|
||
continue
|
||
fi
|
||
|
||
local -a items_arr=()
|
||
local -a item_work_nums=()
|
||
local -a item_kinds=()
|
||
while IFS= read -r item; do
|
||
[[ -n "$item" ]] || continue
|
||
|
||
local work_num
|
||
work_num=$(echo "$item" | jq -r 'if type == "object" then .number // empty else empty end')
|
||
if [[ -z "$work_num" || ! "$work_num" =~ ^[0-9]+$ ]]; then
|
||
continue
|
||
fi
|
||
if echo ",${exclude_numbers}," | grep -q ",${work_num},"; then
|
||
continue
|
||
fi
|
||
|
||
local kind="$wg_kind"
|
||
if [[ "$kind" == "AUTO" ]]; then
|
||
if echo "$item" | jq -e '.head // empty' >/dev/null 2>&1; then
|
||
kind="PR"
|
||
else
|
||
kind="ISSUE"
|
||
fi
|
||
fi
|
||
|
||
items_arr+=("$item")
|
||
item_work_nums+=("$work_num")
|
||
item_kinds+=("$kind")
|
||
done < <(jq -c '.[]' "$out_file" 2>/dev/null)
|
||
|
||
if [[ ${#items_arr[@]} -eq 0 ]]; then
|
||
continue
|
||
fi
|
||
|
||
local batch_count
|
||
if [[ ${#items_arr[@]} -lt $((available_slots - launched)) ]]; then
|
||
batch_count=${#items_arr[@]}
|
||
else
|
||
batch_count=$((available_slots - launched))
|
||
fi
|
||
|
||
log "[pool:${prefix}] '${wg_label}': launching ${batch_count} workers in parallel (${#items_arr[@]} candidates)"
|
||
|
||
local -a pids=()
|
||
local -a tmp_files=()
|
||
local -a work_nums_arr=()
|
||
|
||
for ((i = 0; i < batch_count; i++)); do
|
||
local item="${items_arr[i]}"
|
||
local work_num="${item_work_nums[i]}"
|
||
local kind="${item_kinds[i]}"
|
||
|
||
local worker_tag="${prefix}-${kind}-${work_num}"
|
||
local tmp_prompt
|
||
tmp_prompt=$(mktemp "/tmp/ocb-${worker_tag}-${$}.XXXXXX.txt") || continue
|
||
|
||
local prompt_text=""
|
||
case "$agent" in
|
||
*implementation*) prompt_text=$(build_impl_prompt "$item" "$kind" "$work_num") ;;
|
||
*merge*) prompt_text=$(build_merge_prompt "$item" "$work_num") ;;
|
||
*review*) prompt_text=$(build_review_prompt "$item" "$work_num") ;;
|
||
*grooming*) prompt_text=$(build_groom_prompt "$item" "$kind" "$work_num") ;;
|
||
esac
|
||
|
||
# Skip if the build function returned an empty prompt — sending
|
||
# an empty body to prompt_async results in "no user message
|
||
# found in stream" errors on the server.
|
||
if [[ -z "$prompt_text" ]]; then
|
||
log_warn "[pool:${prefix}] built empty prompt for ${kind} ${work_num} (${agent}) — skipping launch"
|
||
rm -f "$tmp_prompt"
|
||
continue
|
||
fi
|
||
|
||
printf '%s' "$prompt_text" > "$tmp_prompt"
|
||
|
||
# Belt-and-braces: confirm the file actually has content on disk
|
||
# before launching the session.
|
||
if [[ ! -s "$tmp_prompt" ]]; then
|
||
log_warn "[pool:${prefix}] prompt file empty after write for ${kind} ${work_num} (${agent}) — skipping launch"
|
||
rm -f "$tmp_prompt"
|
||
continue
|
||
fi
|
||
|
||
$SESSION_START --server "${BASE}" --tag "$worker_tag" --agent "$agent" --prompt-file "$tmp_prompt" --restart \
|
||
>/dev/null 2>&1 &
|
||
pids+=($!)
|
||
tmp_files+=("$tmp_prompt")
|
||
work_nums_arr+=("$work_num")
|
||
done
|
||
|
||
for ((i = 0; i < ${#pids[@]}; i++)); do
|
||
if wait "${pids[i]}" 2>/dev/null; then
|
||
exclude_numbers="${exclude_numbers}${work_nums_arr[i]},"
|
||
((launched++))
|
||
did_anything=true
|
||
log "[${prefix}] launched ${work_nums_arr[i]} (${agent})"
|
||
else
|
||
log "[${prefix}] FAILED ${work_nums_arr[i]}"
|
||
fi
|
||
rm -f "${tmp_files[i]}"
|
||
done
|
||
done
|
||
done
|
||
|
||
log "Pool ${prefix}: done — ${launched} workers launched this iteration"
|
||
done
|
||
|
||
# Clean up the iteration's shared cache dir
|
||
rm -rf "$cache_dir"
|
||
fi
|
||
|
||
# ── Step 3: Health & maintenance cycle ────────────────────────────────
|
||
if run_health_cycle; then
|
||
did_anything=true
|
||
fi
|
||
|
||
# ── Step 4: Idle sleep if nothing happened ────────────────────────────
|
||
if ! $did_anything; then
|
||
log "Nothing to do — sleeping ${SLEEP_WHEN_IDLE}s"
|
||
record_idle_event
|
||
sleep "$SLEEP_WHEN_IDLE"
|
||
fi
|
||
done
|
||
|
||
log "Loop stopped. Shutting down."
|
||
exit 0
|
||
}
|
||
|
||
main_loop
|