bypassBackoff?: boolean;

}

export type ActiveDeliveryClaimResult<T> =

| { status: "claimed"; value: T }

| { status: "claimed-by-other-owner" };

const MAX_RETRIES = 5;

/** Backoff delays in milliseconds indexed by retry count (1-based). */

entriesInProgress.delete(entryId);

}

/**

* Claim an entry id against the shared in-memory recovery set so a concurrent

* reconnect/startup drain will skip it while the owning caller is mid-flight.

* Returns `false` if the id is already claimed. Callers must pair a successful

* claim with {@link releaseActiveDelivery} in a `finally`. The claim is

* process-local and intentionally does not survive a crash, so crash-replay

* paths still recover fresh entries whose owning process died.

*/

export function tryClaimActiveDelivery(entryId: string): boolean {

return claimRecoveryEntry(entryId);

}

export async function withActiveDeliveryClaim<T>(

entryId: string,

fn: () => Promise<T>,

): Promise<ActiveDeliveryClaimResult<T>> {

if (!claimRecoveryEntry(entryId)) {

return { status: "claimed-by-other-owner" };

}

export function releaseActiveDelivery(entryId: string): void {

releaseRecoveryEntry(entryId);

try {

return { status: "claimed", value: await fn() };

} finally {

releaseRecoveryEntry(entryId);

}

}

function buildRecoveryDeliverParams(entry: QueuedDelivery, cfg: OpenClawConfig) {

const now = Date.now();

const deliver = opts.deliver;

const matchingEntries = (await loadPendingDeliveries(opts.stateDir))

.map((entry) => ({

entry,

decision: opts.selectEntry(entry, now),

}))

.filter(

(item): item is { entry: QueuedDelivery; decision: PendingDeliveryDrainDecision } =>

item.decision.match,

)

.toSorted((a, b) => a.entry.enqueuedAt - b.entry.enqueuedAt);

.filter((entry) => opts.selectEntry(entry, now).match)

.toSorted((a, b) => a.enqueuedAt - b.enqueuedAt);

if (matchingEntries.length === 0) {

return;

`${opts.logLabel}: ${matchingEntries.length} pending message(s) matched ${opts.drainKey}`,

);

for (const { entry, decision } of matchingEntries) {

for (const entry of matchingEntries) {

if (!claimRecoveryEntry(entry.id)) {

opts.log.info(`${opts.logLabel}: entry ${entry.id} is already being recovered`);

continue;

continue;

}

const currentDecision = opts.selectEntry(currentEntry, Date.now());

if (!currentDecision.match) {

opts.log.info(`${opts.logLabel}: entry ${currentEntry.id} no longer matches, skipping`);

continue;

}

if (currentEntry.retryCount >= MAX_RETRIES) {

try {

await moveToFailed(currentEntry.id, opts.stateDir);

continue;

}

if (!decision.bypassBackoff) {

if (!currentDecision.bypassBackoff) {

const retryEligibility = isEntryEligibleForRecoveryRetry(currentEntry, Date.now());

if (!retryEligibility.eligible) {

opts.log.info(