t18653: fix(pulse): resume reconcile single pass from a cursor after budget abort or stage timeout (GH#34299)

.agents/scripts/pulse-issue-reconcile-cursor.sh

@@ -0,0 +1,142 @@
+#!/usr/bin/env bash
+# SPDX-License-Identifier: MIT
+# SPDX-FileCopyrightText: 2025-2026 Marcus Quinn
+# =============================================================================
+# pulse-issue-reconcile-cursor.sh — resume cursor for the single-pass reconcile
+# =============================================================================
+# GH#34299: reconcile_issues_single_pass used to restart at the first repo and
+# first issue every cycle. When the pass ended by time-budget abort or stage
+# timeout, every issue past the abort point was starved indefinitely (e.g. a
+# parent tracker whose children had all closed never reached the close stage).
+#
+# The cursor records the next issue to process. The next pass rotates the slug
+# list to start at the cursor slug and skips that slug's earlier issues. A pass
+# that reaches the end of the rotated list clears the cursor, so the next pass
+# starts from the top again and every issue is visited across cycles.
+#
+# Cursor file format (one line): slug<TAB>issue_number<TAB>attempts
+#   issue_number empty → resume at the start of slug.
+#   attempts counts resumes at the same issue that never moved past it (a
+#   stage-timeout kill mid-issue). At AIDEVOPS_RECONCILE_SP_CURSOR_MAX_ATTEMPTS the
+#   issue is skipped so a single slow issue cannot pin the cursor.
+#
+# State is held in the caller's locals (Bash 3.2 dynamic scope):
+#   _sp_cursor_file _sp_cursor_slug _sp_cursor_issue _sp_cursor_attempts
+#
+# Sourced by pulse-issue-reconcile.sh. Do NOT invoke directly.
+#
+# Part of aidevops framework: https://aidevops.sh
+
+# Apply strict mode only when executed directly (not when sourced)
+[[ "${BASH_SOURCE[0]}" == "${0}" ]] && set -euo pipefail
+
+# Include guard
+[[ -n "${_PULSE_ISSUE_RECONCILE_CURSOR_LOADED:-}" ]] && return 0
+_PULSE_ISSUE_RECONCILE_CURSOR_LOADED=1
+
+#######################################
+# Load and validate the cursor into the caller's _sp_cursor_* locals. A
+# missing, truncated or malformed cursor resets to "start from the top".
+# Env: AIDEVOPS_RECONCILE_SP_CURSOR_FILE overrides the default state path.
+# Returns: 0 always
+#######################################
+_pir_sp_cursor_load() {
+	_sp_cursor_file="${AIDEVOPS_RECONCILE_SP_CURSOR_FILE:-${HOME}/.aidevops/state/reconcile-single-pass.cursor}"
+	_sp_cursor_slug=""
+	_sp_cursor_issue=""
+	_sp_cursor_attempts=0
+	local cursor_dir="${_sp_cursor_file%/*}"
+	if [[ "$cursor_dir" != "$_sp_cursor_file" && ! -d "$cursor_dir" ]]; then
+		mkdir -p "$cursor_dir" 2>/dev/null || true
+	fi
+	[[ -r "$_sp_cursor_file" ]] || return 0
+	local line="" slug="" issue="" attempts=""
+	IFS= read -r line <"$_sp_cursor_file" || [[ -n "$line" ]] || return 0
+	IFS=$'\t' read -r slug issue attempts <<<"$line"
+	[[ "$slug" =~ ^[A-Za-z0-9._-]+/[A-Za-z0-9._-]+$ ]] || return 0
+	[[ -z "$issue" || "$issue" =~ ^[0-9]+$ ]] || return 0
+	[[ "$attempts" =~ ^[0-9]+$ ]] || attempts=0
+	_sp_cursor_slug="$slug"
+	_sp_cursor_issue="$issue"
+	_sp_cursor_attempts="$attempts"
+	return 0
+}
+
+#######################################
+# Record the next position to process. Builtin printf with a redirect, so
+# per-issue writes do not fork. A failed write is non-fatal: the next pass
+# simply starts from the top, which is the pre-GH#34299 behaviour.
+# Args: $1=slug $2=issue number (may be empty) $3=attempts (default 0)
+# Returns: 0 always
+#######################################
+_pir_sp_cursor_write() {
+	local slug="$1" issue="${2:-}" attempts="${3:-0}"
+	[[ -n "${_sp_cursor_file:-}" ]] || return 0
+	printf '%s\t%s\t%s\n' "$slug" "$issue" "$attempts" >"$_sp_cursor_file" 2>/dev/null || true
+	return 0
+}
+
+#######################################
+# Remove the cursor after a pass that reached the end of the rotated list.
+# Returns: 0 always
+#######################################
+_pir_sp_cursor_clear() {
+	[[ -n "${_sp_cursor_file:-}" ]] || return 0
+	rm -f "$_sp_cursor_file" 2>/dev/null || true
+	return 0
+}
+
+#######################################
+# Rotate a newline-separated slug list so it starts at the cursor slug.
+# Unknown or empty cursor slug → original order.
+# Args: $1=newline-separated slugs
+# Output: rotated slugs, one per line
+# Returns: 0 always
+#######################################
+_pir_sp_rotate_slugs() {
+	local slugs="$1"
+	if [[ -z "${_sp_cursor_slug:-}" ]]; then
+		printf '%s\n' "$slugs"
+		return 0
+	fi
+	printf '%s\n' "$slugs" | awk -v start="$_sp_cursor_slug" '
+		NF { a[++n] = $0; if ($0 == start && !s) s = n }
+		END {
+			if (!s) s = 1
+			for (i = s; i <= n; i++) print a[i]
+			for (i = 1; i < s; i++) print a[i]
+		}'
+	return 0
+}
+
+#######################################
+# Decide where the cursor slug's issue loop should start. Sets the caller's
+# _sp_seek_issue (issue to fast-forward to, or empty) and _sp_skip_seek_issue
+# (1 when that issue exhausted its attempts and must be skipped). Consumes the
+# cursor so a slug is only fast-forwarded once per pass.
+# Args: $1=slug $2=issues_tsv (rows start with "<number>|")
+# Returns: 0 always
+#######################################
+_pir_sp_cursor_seek_for_slug() {
+	local slug="$1" issues_tsv="$2"
+	local max_attempts="${AIDEVOPS_RECONCILE_SP_CURSOR_MAX_ATTEMPTS:-2}"
+	[[ "$max_attempts" =~ ^[1-9][0-9]*$ ]] || max_attempts=2
+	_sp_seek_issue=""
+	_sp_skip_seek_issue=0
+	[[ -n "${_sp_cursor_slug:-}" && "$slug" == "$_sp_cursor_slug" ]] || return 0
+	local cursor_issue="${_sp_cursor_issue:-}"
+	_sp_cursor_slug=""
+	[[ -n "$cursor_issue" ]] || return 0
+	case $'\n'"$issues_tsv" in
+	*$'\n'"${cursor_issue}|"*) ;;
+	*) return 0 ;; # issue no longer open/cached: start the slug from the top
+	esac
+	_sp_seek_issue="$cursor_issue"
+	if [[ "${_sp_cursor_attempts:-0}" -ge "$max_attempts" ]]; then
+		_sp_skip_seek_issue=1
+		echo "[pulse-wrapper] reconcile_issues_single_pass: skipping ${slug}#${cursor_issue} after ${_sp_cursor_attempts} unfinished resume(s) (GH#34299)" >>"${LOGFILE:-/dev/null}"
+	else
+		echo "[pulse-wrapper] reconcile_issues_single_pass: resuming at ${slug}#${cursor_issue} (GH#34299)" >>"${LOGFILE:-/dev/null}"
+	fi
+	return 0
+}

.agents/scripts/pulse-issue-reconcile.sh

@@ -243,6 +243,9 @@ source "${_PIR_SCRIPT_DIR}/pulse-issue-reconcile-parent.sh"
 # shellcheck source=./pulse-issue-reconcile-close.sh
 # shellcheck disable=SC1091  # sub-library resolved at runtime via $SCRIPT_DIR
 source "${_PIR_SCRIPT_DIR}/pulse-issue-reconcile-close.sh"
+# shellcheck source=./pulse-issue-reconcile-cursor.sh
+# shellcheck disable=SC1091  # sub-library resolved at runtime via $SCRIPT_DIR
+source "${_PIR_SCRIPT_DIR}/pulse-issue-reconcile-cursor.sh"
 
 #######################################
 # (Phase 12 helper) Assign runner to orphaned active issues.
@@ -1443,6 +1446,9 @@ _pir_reconcile_slug() {
 		] | join("|")
 	') || issues_tsv=""
 	[[ -n "$issues_tsv" ]] || return 0
+	# GH#34299: fast-forward the cursor slug to the issue a prior pass stopped at.
+	local _sp_seek_issue="" _sp_skip_seek_issue=0
+	_pir_sp_cursor_seek_for_slug "$slug" "$issues_tsv"
 
 	# t2985: one merged-PR prefetch per repo; empty lookup defers stage 3 closes.
 	local oimp_lookup=""
@@ -1489,13 +1495,25 @@ _pir_reconcile_issue_rows() {
 		issue_author_association issue_author_login_b64 issue_author_type issue_author_is_bot; do
 		[[ "$issue_num" =~ ^[0-9]+$ ]] || continue
 
+		# GH#34299: skip issues a prior pass already covered on this slug.
+		local _sp_attempts=0
+		if [[ -n "${_sp_seek_issue:-}" ]]; then
+			[[ "$issue_num" == "$_sp_seek_issue" ]] || continue
+			_sp_seek_issue=""
+			[[ "${_sp_skip_seek_issue:-0}" -eq 1 ]] && continue
+			_sp_attempts=$((${_sp_cursor_attempts:-0} + 1))
+		fi
+
 		# t2984: per-issue budget gate (cheap — uses Bash builtin SECONDS)
 		if [[ "$_t2984_budget" -gt 0 ]]; then
 			if [[ $((SECONDS - _t2984_start_ts)) -ge "$_t2984_budget" ]]; then
 				_t2984_aborted=1
+				_pir_sp_cursor_write "$slug" "$issue_num" 0
 				return 0
 			fi
 		fi
+		# Record the in-flight issue so a stage-timeout kill resumes here.
+		_pir_sp_cursor_write "$slug" "$issue_num" "$_sp_attempts"
 
 		local issue_title="" issue_body="" issue_author_login=""
 		if [[ -n "$issue_title_b64" ]]; then
@@ -1666,19 +1684,28 @@ reconcile_issues_single_pass() {
 			"$cpt_max_repair_repo_scans" "$cpt_max_repair_candidates"
 		cpt_total_reopened="$_PIR_RECENT_PARENT_CYCLE_REOPENED"
 	fi
+	# GH#34299: resume where the previous aborted or killed pass stopped.
+	local _sp_cursor_file="" _sp_cursor_slug="" _sp_cursor_issue="" _sp_cursor_attempts=0
+	_pir_sp_cursor_load
+	local _sp_slugs=""
+	_sp_slugs=$(jq -r '.initialized_repos[] | select(.maintenance != false and .pulse == true and (.local_only // false) == false and .slug != "") | .slug // ""' "$repos_json" || true)
 	local slug=""
 	while IFS= read -r slug; do
 		[[ -n "$slug" ]] || continue
 		local _slug_start=$SECONDS
 		if [[ "$_t2984_budget" -gt 0 ]]; then
 			if [[ $((SECONDS - _t2984_start_ts)) -ge "$_t2984_budget" ]]; then
 				_t2984_aborted=1
+				_pir_sp_cursor_write "$slug" "" 0
 				break
 			fi
 		fi
+		# The resume slug keeps its issue position until its rows are reached.
+		[[ "$slug" == "${_sp_cursor_slug:-}" ]] || _pir_sp_cursor_write "$slug" "" 0
 		_pir_reconcile_slug "$slug" "$_slug_start"
 		[[ "$_t2984_aborted" -eq 1 ]] && break
-	done < <(jq -r '.initialized_repos[] | select(.maintenance != false and .pulse == true and (.local_only // false) == false and .slug != "") | .slug // ""' "$repos_json" || true)
+	done < <(_pir_sp_rotate_slugs "$_sp_slugs")
+	[[ "$_t2984_aborted" -eq 1 ]] || _pir_sp_cursor_clear
 	_pir_finish_reconcile_cycle
 	return 0
 }

.agents/scripts/tests/test-reconcile-budget-isolation.sh

@@ -61,6 +61,8 @@ trap 'rm -rf "$TEST_TMPDIR"' EXIT
 LOGFILE="${TEST_TMPDIR}/pulse.log"
 : >"$LOGFILE"
 export LOGFILE
+# GH#34299: keep the single-pass resume cursor out of the real state dir.
+export AIDEVOPS_RECONCILE_SP_CURSOR_FILE="${TEST_TMPDIR}/reconcile-single-pass.cursor"
 
 # Fake repos.json — one pulse-enabled repo
 REPOS_JSON="${TEST_TMPDIR}/repos.json"

.agents/scripts/tests/test-reconcile-single-pass-cursor.sh

@@ -0,0 +1,145 @@
+#!/usr/bin/env bash
+# SPDX-License-Identifier: MIT
+# SPDX-FileCopyrightText: 2025-2026 Marcus Quinn
+# test-reconcile-single-pass-cursor.sh — GH#34299 regression test.
+#
+# reconcile_issues_single_pass used to restart at the first repo and issue on
+# every cycle, so a budget abort or stage-timeout kill starved every issue past
+# the abort point. This drives the real orchestrator with stubbed per-issue
+# stages that advance Bash's SECONDS to simulate cost, and asserts:
+#   1. a budget abort records the next unprocessed position;
+#   2. the next pass resumes there and rotates repos, so coverage completes;
+#   3. a pass that reaches the end of the rotated list clears the cursor;
+#   4. an issue resumed repeatedly without finishing is skipped;
+#   5. a malformed cursor falls back to the original order.
+#
+# Usage: bash .agents/scripts/tests/test-reconcile-single-pass-cursor.sh
+
+# shellcheck disable=SC2155
+readonly TEST_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+readonly RECONCILE_SRC="${TEST_DIR}/../pulse-issue-reconcile.sh"
+
+pass=0
+fail=0
+_pass() { echo "PASS: $1"; pass=$((pass + 1)); return 0; }
+_fail() { echo "FAIL: $1"; fail=$((fail + 1)); return 0; }
+
+TEST_TMPDIR=$(mktemp -d "${TMPDIR:-/tmp}/test-reconcile-sp-cursor.XXXXXX")
+trap 'rm -rf "$TEST_TMPDIR"' EXIT
+
+export LOGFILE="${TEST_TMPDIR}/pulse.log"
+: >"$LOGFILE"
+export REPOS_JSON="${TEST_TMPDIR}/repos.json"
+cat >"$REPOS_JSON" <<'JSON'
+{"initialized_repos": [
+  {"slug": "test/a", "pulse": true, "local_only": false},
+  {"slug": "test/b", "pulse": true, "local_only": false}
+]}
+JSON
+export AIDEVOPS_RECONCILE_SP_CURSOR_FILE="${TEST_TMPDIR}/state/reconcile-single-pass.cursor"
+CURSOR="$AIDEVOPS_RECONCILE_SP_CURSOR_FILE"
+VISITS="${TEST_TMPDIR}/visits"
+INFLIGHT="${TEST_TMPDIR}/inflight"
+
+set_issue_status() { return 0; }
+# shellcheck disable=SC1090
+source "$RECONCILE_SRC"
+
+# Stubs (defined after sourcing so they shadow the module's functions).
+# shellcheck disable=SC2317
+_repair_pending_planning_publications() { return 0; }
+# shellcheck disable=SC2317
+_repair_recently_closed_parents_cycle() { _PIR_RECENT_PARENT_CYCLE_REOPENED=0; return 0; }
+# shellcheck disable=SC2317
+_pir_reconcile_objective_ledger() { return 0; }
+# shellcheck disable=SC2317
+_build_oimp_lookup_for_slug() { return 0; }
+# shellcheck disable=SC2317
+_read_cache_issues_for_slug() {
+	local slug="$1"
+	case "$slug" in
+	test/a) printf '%s' '[{"number":1,"title":"a1","labels":[],"body":""},{"number":2,"title":"a2","labels":[],"body":""},{"number":3,"title":"a3","labels":[],"body":""}]' ;;
+	test/b) printf '%s' '[{"number":4,"title":"b4","labels":[],"body":""},{"number":5,"title":"b5","labels":[],"body":""}]' ;;
+	*) return 1 ;;
+	esac
+	return 0
+}
+# Each issue "costs" 100 seconds of budget. slug/issue_num are the caller's
+# loop locals (Bash dynamic scope), as in the real stage function.
+# shellcheck disable=SC2317,SC2154
+_pir_reconcile_issue_stages() {
+	printf '%s#%s\n' "$slug" "$issue_num" >>"$VISITS"
+	# Snapshot the in-flight cursor, i.e. what a stage-timeout kill would leave.
+	tr '\t' ',' <"$CURSOR" >"$INFLIGHT" 2>/dev/null || true
+	SECONDS=$((SECONDS + 100))
+	return 0
+}
+
+_run_pass() {
+	: >"$VISITS"
+	RECONCILE_TIME_BUDGET_SECS="$1" reconcile_issues_single_pass >/dev/null 2>&1
+	tr '\n' ' ' <"$VISITS" | sed 's/ $//'
+	return 0
+}
+
+_cursor() {
+	if [[ -f "$CURSOR" ]]; then
+		tr '\t' ',' <"$CURSOR" | tr -d '\n'
+	else
+		printf 'none'
+	fi
+	return 0
+}
+
+_expect() {
+	local name="$1" expected="$2" actual="$3"
+	if [[ "$actual" == "$expected" ]]; then
+		_pass "$name"
+	else
+		_fail "${name}: expected '${expected}', got '${actual}'"
+	fi
+	return 0
+}
+
+rm -f "$CURSOR"
+_expect "pass 1 stops at budget" "test/a#1 test/a#2 test/a#3" "$(_run_pass 250)"
+_expect "pass 1 records next repo" "test/b,,0" "$(_cursor)"
+_expect "pass 2 resumes and rotates" "test/b#4 test/b#5 test/a#1" "$(_run_pass 250)"
+_expect "pass 2 records next issue" "test/a,2,0" "$(_cursor)"
+_expect "pass 3 resumes mid-repo" "test/a#2 test/a#3 test/b#4" "$(_run_pass 250)"
+_expect "pass 3 records next issue" "test/b,5,0" "$(_cursor)"
+
+_expect "unbudgeted pass resumes then wraps" "test/b#5 test/a#1 test/a#2 test/a#3" "$(_run_pass 0)"
+_expect "completed pass clears cursor" "none" "$(_cursor)"
+_expect "cleared cursor restarts at top" "test/a#1 test/a#2 test/a#3 test/b#4 test/b#5" "$(_run_pass 0)"
+
+# A stage-timeout kill leaves the in-flight issue as the cursor.
+mkdir -p "${CURSOR%/*}"
+printf 'test/a\t2\t1\n' >"$CURSOR"
+# The resume repo's earlier issues are left for the next (top-down) pass.
+_expect "killed issue is retried" "test/a#2 test/a#3 test/b#4 test/b#5" "$(_run_pass 0)"
+printf 'test/a\t2\t2\n' >"$CURSOR"
+_expect "repeatedly unfinished issue is skipped" "test/a#3 test/b#4 test/b#5" "$(_run_pass 0)"
+if grep -q 'skipping test/a#2' "$LOGFILE"; then
+	_pass "skip is logged"
+else
+	_fail "skip is logged"
+fi
+
+# A resumed in-flight issue is recorded with an incremented attempt count.
+printf 'test/a\t2\t0\n' >"$CURSOR"
+_expect "resumed issue runs first" "test/a#2" "$(_run_pass 50)"
+_expect "resumed in-flight issue increments attempts" "test/a,2,1" "$(tr -d '\n' <"$INFLIGHT")"
+_expect "budget abort records next issue with fresh attempts" "test/a,3,0" "$(_cursor)"
+
+printf 'garbage\n' >"$CURSOR"
+_expect "malformed cursor uses original order" "test/a#1 test/a#2 test/a#3 test/b#4 test/b#5" "$(_run_pass 0)"
+printf 'test/gone\t9\t0\n' >"$CURSOR"
+_expect "unknown repo cursor uses original order" "test/a#1 test/a#2 test/a#3 test/b#4 test/b#5" "$(_run_pass 0)"
+printf 'test/b\t99\t0\n' >"$CURSOR"
+_expect "closed cursor issue restarts its repo" "test/b#4 test/b#5 test/a#1 test/a#2 test/a#3" "$(_run_pass 0)"
+
+echo ""
+echo "Results: ${pass} passed, ${fail} failed"
+[[ "$fail" -eq 0 ]] || exit 1
+exit 0