Skip to content

fix: recover interrupted command dispatches #407

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
steipete merged 8 commits into from
Jul 4, 2026
Merged

fix: recover interrupted command dispatches #407

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
228f9b3
fix: recover interrupted command dispatches
steipete Jul 4, 2026
4bb104c
ci: deduplicate command dispatch receipts
steipete Jul 4, 2026
6203048
fix: choose oldest command receipt
steipete Jul 4, 2026
2e5d648
fix: complete command dispatch recovery
steipete Jul 4, 2026
deaf276
fix: scope synthetic dispatch receipts per attempt
steipete Jul 4, 2026
1c7e05b
fix: verify dispatch execution receipts
steipete Jul 4, 2026
eab3622
fix: ignore receipt-only duplicate successes
Jul 4, 2026
ca335bc
fix: preserve recoverable command dispatch
steipete Jul 4, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
38 changes: 37 additions & 1 deletion .github/workflows/assist.yml
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
name: ClawSweeper assist

run-name: ${{ format('Assist {0}#{1}', github.event.client_payload.target_repo || 'openclaw/openclaw', github.event.client_payload.item_number || '?') }}
run-name: ${{ github.event.client_payload.dispatch_key && format('Assist {0}#{1} [{2}]', github.event.client_payload.target_repo || 'openclaw/openclaw', github.event.client_payload.item_number || '?', github.event.client_payload.dispatch_key) || format('Assist {0}#{1}', github.event.client_payload.target_repo || 'openclaw/openclaw', github.event.client_payload.item_number || '?') }}

on:
repository_dispatch:
Expand Down Expand Up @@ -47,6 +47,7 @@ on:
type: string

permissions:
actions: read
contents: read
issues: write
pull-requests: read
Expand All @@ -61,7 +62,42 @@ concurrency:
cancel-in-progress: false

jobs:
receipt:
name: Deduplicate command dispatch receipt
runs-on: ubuntu-latest
outputs:
proceed: ${{ steps.receipt.outputs.proceed }}
steps:
- uses: actions/checkout@v5
with:
sparse-checkout: scripts/dispatch-receipt-owner.sh
sparse-checkout-cone-mode: false

- id: receipt
env:
DISPATCH_KEY: ${{ github.event.client_payload.dispatch_key || '' }}
ITEM_NUMBER: ${{ github.event.client_payload.item_number || '?' }}
TARGET_REPO: ${{ github.event.client_payload.target_repo || 'openclaw/openclaw' }}
GH_TOKEN: ${{ github.token }}
run: |
set -euo pipefail
if [ -z "$DISPATCH_KEY" ]; then
echo "proceed=true" >> "$GITHUB_OUTPUT"
exit 0
fi
expected_title="Assist ${TARGET_REPO}#${ITEM_NUMBER} [${DISPATCH_KEY}]"
owner="$(bash scripts/dispatch-receipt-owner.sh \
assist.yml "$expected_title" "$GITHUB_RUN_ID" assist)"
if [ "$owner" = "owner" ]; then
echo "proceed=false" >> "$GITHUB_OUTPUT"
echo "An older active or successfully executed command dispatch already exists; skipping duplicate assist."
exit 0
fi
echo "proceed=true" >> "$GITHUB_OUTPUT"

capacity:
needs: receipt
if: ${{ needs.receipt.outputs.proceed == 'true' }}
runs-on: ubuntu-latest
outputs:
proceed: ${{ steps.limit.outputs.proceed }}
Expand Down
49 changes: 48 additions & 1 deletion .github/workflows/repair-cluster-worker.yml
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
name: repair cluster worker

run-name: ${{ contains(inputs.job, '/inbox/issue-') && format('issue implementation {0}', inputs.job) || contains(inputs.job, '/inbox/automerge-') && format('automerge repair {0}', inputs.job) || format('repair cluster {0}', inputs.job) }}
run-name: ${{ inputs.dispatch_key && format('{0} [{1}]', contains(inputs.job, '/inbox/issue-') && format('issue implementation {0}', inputs.job) || contains(inputs.job, '/inbox/automerge-') && format('automerge repair {0}', inputs.job) || format('repair cluster {0}', inputs.job), inputs.dispatch_key) || contains(inputs.job, '/inbox/issue-') && format('issue implementation {0}', inputs.job) || contains(inputs.job, '/inbox/automerge-') && format('automerge repair {0}', inputs.job) || format('repair cluster {0}', inputs.job) }}

on:
workflow_dispatch:
Expand All @@ -9,6 +9,11 @@ on:
description: "Job markdown path, for example jobs/openclaw/inbox/cluster-001.md"
required: true
type: string
dispatch_key:
description: "Optional command-router idempotency key"
required: false
default: ""
type: string
mode:
description: "Worker mode"
required: true
Expand Down Expand Up @@ -67,8 +72,50 @@ concurrency:
cancel-in-progress: false

jobs:
receipt:
name: Deduplicate command dispatch receipt
runs-on: ubuntu-latest
permissions:
actions: read
contents: read
outputs:
proceed: ${{ steps.receipt.outputs.proceed }}
steps:
- uses: actions/checkout@v5
with:
sparse-checkout: scripts/dispatch-receipt-owner.sh
sparse-checkout-cone-mode: false

- id: receipt
env:
DISPATCH_KEY: ${{ inputs.dispatch_key }}
JOB_PATH: ${{ inputs.job }}
GH_TOKEN: ${{ github.token }}
run: |
set -euo pipefail
if [ -z "$DISPATCH_KEY" ]; then
echo "proceed=true" >> "$GITHUB_OUTPUT"
exit 0
fi
case "$JOB_PATH" in
*/inbox/issue-*) title="issue implementation $JOB_PATH" ;;
*/inbox/automerge-*) title="automerge repair $JOB_PATH" ;;
*) title="repair cluster $JOB_PATH" ;;
esac
expected_title="${title} [${DISPATCH_KEY}]"
owner="$(bash scripts/dispatch-receipt-owner.sh \
repair-cluster-worker.yml "$expected_title" "$GITHUB_RUN_ID" "Plan and review cluster")"
if [ "$owner" = "owner" ]; then
echo "proceed=false" >> "$GITHUB_OUTPUT"
echo "An older active or successfully executed command dispatch already exists; skipping duplicate repair."
exit 0
fi
echo "proceed=true" >> "$GITHUB_OUTPUT"

cluster:
name: Plan and review cluster
needs: receipt
if: ${{ needs.receipt.outputs.proceed == 'true' }}
runs-on: ${{ inputs.runner }}
timeout-minutes: 90
outputs:
Expand Down
13 changes: 11 additions & 2 deletions .github/workflows/sweep.yml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,8 +16,12 @@ run-name: >-
(github.event_name == 'repository_dispatch' &&
github.event.action == 'clawsweeper_target_sweep') &&
format('Review target repo {0}', github.event.client_payload.target_repo || 'openclaw/openclaw') ||
(github.event_name == 'repository_dispatch' && github.event.client_payload.dispatch_key != '') &&
format('Review event item {0}#{1} [{2}]', github.event.client_payload.target_repo || 'openclaw/openclaw', github.event.client_payload.item_number || '?', github.event.client_payload.dispatch_key) ||
github.event_name == 'repository_dispatch' &&
format('Review event item {0}#{1}', github.event.client_payload.target_repo || 'openclaw/openclaw', github.event.client_payload.item_number || '?') ||
(github.event_name == 'workflow_dispatch' && startsWith(github.event.inputs.item_numbers, 'router-')) &&
format('Review event item {0}#{1} [{2}]', github.event.inputs.target_repo || 'openclaw/openclaw', github.event.inputs.item_number || '?', github.event.inputs.item_numbers) ||
((github.event_name == 'workflow_dispatch' &&
github.event.inputs.apply_existing == 'true' &&
github.event.inputs.apply_sync_comments_only == 'true') ||
Expand Down Expand Up @@ -229,9 +233,12 @@ jobs:
const payload = JSON.parse(process.env.CLIENT_PAYLOAD || "{}");
const itemKind = payload.item_kind === "pull_request" ? "pull_request" : "issue";
const sourceEvent = payload.source_event === "pull_request" ? "pull_request" : "issues";
const dispatchKey = String(payload.dispatch_key || "").trim();
process.stdout.write(
JSON.stringify({
delivery_id: `legacy:${process.env.GITHUB_RUN_ID}:${process.env.GITHUB_RUN_ATTEMPT}`,
delivery_id: dispatchKey
? `router:${dispatchKey}`
: `legacy:${process.env.GITHUB_RUN_ID}:${process.env.GITHUB_RUN_ATTEMPT}`,
decision: {
targetRepo: payload.target_repo || "openclaw/openclaw",
targetBranch: payload.target_branch || "main",
Expand Down Expand Up @@ -1157,7 +1164,9 @@ jobs:
BATCH_SIZE: ${{ steps.mode.outputs.batch_size }}
HOT_INTAKE: ${{ steps.mode.outputs.hot_intake }}
ITEM_NUMBER: ${{ github.event.inputs.item_number || '' }}
ITEM_NUMBERS: ${{ github.event.inputs.item_numbers || github.event.client_payload.item_number || '' }}
# A router fallback carries its receipt key in item_numbers because GitHub caps this
# workflow at 25 inputs; item_number remains the exact review target in that case.
ITEM_NUMBERS: ${{ (github.event.inputs.item_number != '' && startsWith(github.event.inputs.item_numbers, 'router-') && github.event.inputs.item_number) || github.event.inputs.item_numbers || github.event.client_payload.item_number || '' }}
MAX_PAGES: ${{ steps.mode.outputs.max_pages }}
MIN_ACTIVE_SHARDS: ${{ steps.mode.outputs.min_active_shards }}
MIN_BACKFILL_REVIEW_AGE_MINUTES: ${{ steps.mode.outputs.min_backfill_review_age_minutes }}
Expand Down
48 changes: 48 additions & 0 deletions scripts/dispatch-receipt-owner.sh
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,48 @@
#!/usr/bin/env bash
set -euo pipefail

workflow="${1:?workflow file is required}"
expected_title="${2:?expected run title is required}"
current_run_id="${3:?current run id is required}"
required_job_name="${4:?required worker job name is required}"

runs_json="$(gh api --method GET \
"repos/${GITHUB_REPOSITORY:?GITHUB_REPOSITORY is required}/actions/workflows/${workflow}/runs?per_page=100")"

active_owner_id="$(
jq -r --arg title "$expected_title" --arg current "$current_run_id" '
first(
.workflow_runs[]
| select(.display_title == $title and .id < ($current | tonumber))
| select(.status == "queued" or .status == "in_progress" or .status == "waiting" or .status == "pending" or .status == "requested")
| .id
) // empty
' <<<"$runs_json"
)"
if [ -n "$active_owner_id" ]; then
printf 'owner\n'
exit 0
fi

successful_run_ids="$(
jq -r --arg title "$expected_title" --arg current "$current_run_id" '
.workflow_runs[]
| select(.display_title == $title and .id < ($current | tonumber) and .conclusion == "success")
| .id
' <<<"$runs_json"
)"
while IFS= read -r run_id; do
if [ -z "$run_id" ]; then
continue
fi
jobs_json="$(gh api --method GET \
"repos/${GITHUB_REPOSITORY}/actions/runs/${run_id}/jobs?per_page=100")"
if jq -e --arg required "$required_job_name" \
'any(.jobs[]; .name == $required and .conclusion == "success")' \
<<<"$jobs_json" >/dev/null; then
printf 'owner\n'
exit 0
fi
done <<<"$successful_run_ids"

printf 'none\n'
88 changes: 85 additions & 3 deletions src/repair/comment-router-utils.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,33 @@ const DEFAULT_IGNORED_CHECKS = [
];
const TRANSIENT_CANCELLED_CHECKS = new Set(["real behavior proof"]);

export function dispatchClaimLookupKeys(entry: LooseRecord) {
const keys: string[] = [];
const commentId = String(entry.comment_id ?? "").trim();
const commentUpdatedAt = String(entry.comment_updated_at ?? "").trim();
if (commentId && commentUpdatedAt) keys.push(`comment:${commentId}:${commentUpdatedAt}`);
const idempotencyKey = String(entry.idempotency_key ?? "").trim();
if (idempotencyKey) keys.push(`idempotency:${idempotencyKey}`);
return keys;
}

export function dispatchReceiptKeyMaterial(entry: LooseRecord, claim: LooseRecord | null) {
const idempotencyKey = String(entry.idempotency_key ?? entry.comment_version_key ?? "unknown");
if (entry.automation_source !== "repair_loop_label_sweep") return idempotencyKey;
const attempt = String(
claim?.processed_at ?? entry.processed_at ?? entry.comment_updated_at ?? "unknown-attempt",
);
return `${idempotencyKey}:${attempt}`;
}

export function hasSuccessfulDispatchExecutionJob(jobs: LooseRecord[], requiredJobName: string) {
return jobs.some(
(job) =>
String(job.name ?? "") === requiredJobName &&
String(job.conclusion ?? "").toLowerCase() === "success",
);
}

export function summarizeChecks(checks: LooseRecord[]) {
const ignored = ignoredCheckNames();
const latestChecks = latestCheckRuns(checks);
Expand Down Expand Up @@ -115,6 +142,49 @@ export function shouldSuppressProcessedCommentVersion(entry: LooseRecord) {
return true;
}

export function dispatchClaimDecision({
claim,
runs,
expectedTitle,
nowMs = Date.now(),
graceMs = 300_000,
}: {
claim: LooseRecord | null;
runs: LooseRecord[];
expectedTitle: string;
nowMs?: number;
graceMs?: number;
}) {
if (!claim) return { action: "dispatch", run: null };
const normalizedGraceMs = Number.isFinite(graceMs) ? Math.max(0, graceMs) : 300_000;
const claimedAtMs = Date.parse(String(claim.processed_at ?? ""));
const matchingRuns = runs.filter((run) => {
if (String(run.display_title ?? run.displayTitle ?? "") !== expectedTitle) return false;
const createdAtMs = Date.parse(String(run.created_at ?? run.createdAt ?? ""));
return (
Number.isFinite(claimedAtMs) &&
Number.isFinite(createdAtMs) &&
createdAtMs >= claimedAtMs - 5_000
);
});
const successfulRun = matchingRuns.find(
(run) =>
String(run.conclusion ?? "").toLowerCase() === "success" &&
run.dispatch_execution_verified !== false,
);
if (successfulRun) return { action: "recover", run: successfulRun };
Comment on lines +170 to +175

@chatgpt-codex-connector chatgpt-codex-connector Bot Jul 4, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 Badge Ignore receipt-only successes when recovering dispatches

With the new receipt jobs in assist.yml and repair-cluster-worker.yml, a duplicate run that sees an older active run writes proceed=false and exits successfully, so that workflow run can have conclusion success even though the assist/cluster worker never ran. Because this branch recovers on any matching successful run, if the older run later fails, a subsequent router retry will recover the skipped duplicate and mark the dispatch executed instead of redispatching.

Useful? React with 👍 / 👎.

const activeRun = matchingRuns.find((run) =>
["queued", "in_progress", "waiting", "pending", "requested"].includes(
String(run.status ?? "").toLowerCase(),
),
);
if (activeRun) return { action: "wait", run: null };
if (Number.isFinite(claimedAtMs) && nowMs - claimedAtMs >= normalizedGraceMs) {
return { action: "dispatch", run: null };
}
return { action: "wait", run: null };
}

export function sortCommentsForRouting(comments: LooseRecord[]) {
return [...comments].sort((left: LooseRecord, right: LooseRecord) => {
const leftTime = commentRoutingTime(left);
Expand Down Expand Up @@ -244,7 +314,9 @@ export function readLedger(file: JsonValue) {

export function appendLedger(current: LooseRecord, entries: LooseRecord[]) {
const compact = entries
.filter((entry: JsonValue) => ["executed", "skipped", "waiting"].includes(entry.status))
.filter((entry: JsonValue) =>
["claimed", "executed", "skipped", "waiting"].includes(entry.status),
)
.filter((entry: JsonValue) => !isNoopSkip(entry))
.map((entry: JsonValue) => {
const actions = compactLedgerActions(entry.actions);
Expand All @@ -271,7 +343,7 @@ export function appendLedger(current: LooseRecord, entries: LooseRecord[]) {
expected_head_sha: entry.expected_head_sha ?? null,
finding_id: entry.finding_id ?? null,
status: entry.status,
processed_at: new Date().toISOString(),
processed_at: entry.processed_at ?? new Date().toISOString(),
target: entry.target
? {
kind: entry.target.kind,
Expand Down Expand Up @@ -313,10 +385,20 @@ function isNoopSkip(entry: LooseRecord) {
}

function stableLedgerEntry(entry: LooseRecord) {
return JSON.stringify({ ...entry, processed_at: null });
return JSON.stringify({
...entry,
processed_at: entry.status === "claimed" ? entry.processed_at : null,
});
}

function ledgerEntryKey(entry: LooseRecord) {
if (
!entry.comment_version_key &&
entry.automation_source === "repair_loop_label_sweep" &&
entry.idempotency_key
) {
return `idempotency:${entry.idempotency_key}`;
Comment on lines +396 to +400

@chatgpt-codex-connector chatgpt-codex-connector Bot Jul 4, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P1 Badge Preserve synthetic sweep repair history

For repair-loop label sweeps, comment_version_key is always null and idempotency_key is stable for the same repo/intent/PR, so this key collapses every executed sweep for that PR into a single ledger entry. autoRepairBlockReason() relies on the ledger entries with executed dispatch_repair actions to enforce maxAutoRepairsPerPr/maxAutoRepairsPerHead; after the second scheduled repair the older dispatch history is overwritten, allowing the sweep lane to undercount prior repairs and exceed the intended safety caps. Collapse only the claimed→executed record for the same attempt, or include attempt-specific material for executed sweep entries.

Useful? React with 👍 / 👎.

}
return (
entry.comment_version_key ??
`${entry.comment_id ?? "unknown"}:${entry.comment_updated_at ?? "unknown"}`
Expand Down
Loading
Loading