Skip to content

Claim command dispatches before execution #380

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

Closed
ag-linden wants to merge 1 commit into from
Closed

Claim command dispatches before execution #380

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
6 changes: 4 additions & 2 deletions src/repair/comment-router-utils.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -104,7 +104,7 @@ function checkTimestamp(check: LooseRecord) {

export function shouldSuppressProcessedCommentVersion(entry: LooseRecord) {
const status = String(entry.status ?? "").toLowerCase();
if (!["executed", "skipped"].includes(status)) return false;
if (!["claimed", "executed", "skipped"].includes(status)) return false;
const intent = String(entry.intent ?? "");
if (
status === "skipped" &&
Expand Down Expand Up @@ -244,7 +244,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 Down
36 changes: 34 additions & 2 deletions src/repair/comment-router.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -294,8 +294,6 @@ const report: LooseRecord = {
if (execute) {
await measureAsync("execute_commands", async () => {
assertMutationActorIsClawsweeperBot();
for (const command of commands) convergePrecreatedCommandAckComments(command);
for (const command of commands) acknowledgeSkippedMaintainerCommand(command);
const capacityRequests = workerCapacityRequests(actionable);
if (capacityRequests.length > 0) {
const capacities = capacityRequests.map((request) =>
Expand All @@ -304,6 +302,12 @@ if (execute) {
report.live_worker_capacity_before_dispatch =
capacities.length === 1 ? capacities[0] : capacities;
}
report.ledger_claimed = measure("claim_dispatch_commands", () =>
claimDispatchCommands(actionable),
);
if (report.ledger_claimed) writeLedger(ledgerPath(), ledger);
Comment on lines +305 to +308

@chatgpt-codex-connector chatgpt-codex-connector Bot Jun 29, 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 Avoid suppressing commands on failed dispatch attempts

This writes a terminal claimed ledger entry before any executeCommand dispatch has completed. In repair-comment-router.yml the ledger publish step runs with if: always(), and claimed entries are included in shouldSuppressProcessedCommentVersion, so if a dispatch command throws after this write (for example gh workflow run/repository_dispatch fails or the process crashes before executeCommand reaches the dispatch), the failed claim is published and future router runs skip the same comment version instead of retrying it. That can permanently drop re-review/repair/assist requests without ever starting the worker.

Useful? React with 👍 / 👎.

for (const command of commands) convergePrecreatedCommandAckComments(command);
for (const command of commands) acknowledgeSkippedMaintainerCommand(command);
for (const command of actionable) executeCommand(command);
});
report.ledger_changed = measure("append_ledger", () => appendLedger(ledger, commands));
Expand Down Expand Up @@ -335,6 +339,34 @@ async function measureAsync<T>(name: string, fn: () => Promise<T>): Promise<T> {
}
}

function claimDispatchCommands(commands: LooseRecord[]) {
const claims = commands.filter(commandNeedsDurableDispatchClaim).map((command) => ({
...command,
status: "claimed",
actions: Array.isArray(command.actions)
? command.actions.map((action: JsonValue) =>
actionNeedsDurableDispatchClaim(action) ? { ...action, status: "claimed" } : action,
)
: command.actions,
}));
return appendLedger(ledger, claims);
}

function commandNeedsDurableDispatchClaim(command: LooseRecord) {
return (
String(command.status ?? "") === "ready" &&
(commandHasAction(command, "dispatch_clawsweeper") ||
commandHasAction(command, "dispatch_repair") ||
commandHasAction(command, "dispatch_assist"))
);
}

function actionNeedsDurableDispatchClaim(action: JsonValue) {
return ["dispatch_clawsweeper", "dispatch_repair", "dispatch_assist"].includes(
String(action?.action ?? ""),
);
}

function assertMutationActorIsClawsweeperBot() {
try {
const viewer = ghJson<LooseRecord>(["api", "user"]);
Expand Down
23 changes: 23 additions & 0 deletions test/repair/comment-router-core.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1378,6 +1378,29 @@ test("router classifies fresh human-review pauses before label sweeps", () => {
assert.match(source, /\.filter\(isReadyHumanReviewPause\)/);
});

test("comment router durably claims dispatch commands before external execution", () => {
const source = readFileSync("src/repair/comment-router.ts", "utf8");
const executeBlock = source.slice(
source.indexOf('await measureAsync("execute_commands"'),
source.indexOf('report.ledger_changed = measure("append_ledger"'),
);
const claimIndex = executeBlock.indexOf("claimDispatchCommands(actionable)");
const ackIndex = executeBlock.indexOf("convergePrecreatedCommandAckComments(command)");
const executeIndex = executeBlock.indexOf("executeCommand(command)");
const claimFunction = source.slice(
source.indexOf("function claimDispatchCommands"),
source.indexOf("function assertMutationActorIsClawsweeperBot"),
);

assert.ok(claimIndex >= 0);
assert.ok(ackIndex > claimIndex);
assert.ok(executeIndex > claimIndex);
assert.match(claimFunction, /status:\s*"claimed"/);
assert.match(claimFunction, /commandHasAction\(command,\s*"dispatch_clawsweeper"\)/);
assert.match(claimFunction, /commandHasAction\(command,\s*"dispatch_repair"\)/);
assert.match(claimFunction, /commandHasAction\(command,\s*"dispatch_assist"\)/);
});

test("trusted autoclose markers are live close gated before close execution", () => {
const source = readFileSync("src/repair/comment-router.ts", "utf8");
const autocloseClassifier = source.slice(
Expand Down
79 changes: 79 additions & 0 deletions test/repair/comment-router-utils.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -130,6 +130,85 @@ test("appendLedger records waiting commands without making them terminal", () =>
assert.equal(shouldSuppressProcessedCommentVersion(ledger.commands[0]), false);
});

test("appendLedger records claimed dispatch commands as terminal idempotency claims", () => {
const ledger = { updated_at: null, commands: [] };

assert.equal(
appendLedger(ledger, [
{
idempotency_key: "claim-before-dispatch",
comment_id: "125",
comment_version_key: "125:2026-04-29T03:01:00Z",
comment_updated_at: "2026-04-29T03:01:00Z",
status: "claimed",
intent: "clawsweeper_re_review",
issue_number: 74499,
repo: "openclaw/openclaw",
actions: [{ action: "dispatch_clawsweeper", status: "claimed" }],
},
]),
true,
);

assert.equal(ledger.commands.length, 1);
assert.equal(ledger.commands[0].status, "claimed");
assert.equal(shouldSuppressProcessedCommentVersion(ledger.commands[0]), true);
assert.deepEqual(ledger.commands[0].actions, [
{
action: "dispatch_clawsweeper",
status: "claimed",
label: null,
job_path: null,
},
]);
});

test("appendLedger upgrades claimed dispatch commands after execution", () => {
const ledger = { updated_at: null, commands: [] };

appendLedger(ledger, [
{
idempotency_key: "claim-before-dispatch",
comment_id: "125",
comment_version_key: "125:2026-04-29T03:01:00Z",
comment_updated_at: "2026-04-29T03:01:00Z",
status: "claimed",
intent: "clawsweeper_re_review",
issue_number: 74499,
repo: "openclaw/openclaw",
actions: [{ action: "dispatch_clawsweeper", status: "claimed" }],
},
]);

assert.equal(
appendLedger(ledger, [
{
idempotency_key: "claim-before-dispatch",
comment_id: "125",
comment_version_key: "125:2026-04-29T03:01:00Z",
comment_updated_at: "2026-04-29T03:01:00Z",
status: "executed",
intent: "clawsweeper_re_review",
issue_number: 74499,
repo: "openclaw/openclaw",
actions: [{ action: "dispatch_clawsweeper", status: "executed" }],
},
]),
true,
);

assert.equal(ledger.commands.length, 1);
assert.equal(ledger.commands[0].status, "executed");
assert.deepEqual(ledger.commands[0].actions, [
{
action: "dispatch_clawsweeper",
status: "executed",
label: null,
job_path: null,
},
]);
});

test("appendLedger ignores no-op skipped command versions", () => {
const ledger = { updated_at: null, commands: [] };

Expand Down