// @ts-nocheck import { appendFile, mkdir, readFile, readdir, rename, rm, stat, writeFile } from 'fs/promises'; import { appendFileSync, existsSync, mkdirSync } from 'fs'; import { dirname, join, resolve } from 'path'; import { fileURLToPath } from 'node:url'; import { isDeepStrictEqual } from 'node:util'; import { safeString } from './utils.js'; import { getDefaultBridge, isRuntimeTransportError, isStrictDispatchRecord, preflightRuntimeBridge, resolveBridgeStateDir, RuntimeBridgeError } from '../../runtime/bridge.js'; import { hasRuntimeAuthority, hasRuntimeDispatchIdentity, recordRuntimeFinalization, reconcileRuntimeFinalizations } from '../../team/state/dispatch-recovery.js'; import { readDispatchRequests } from '../../team/state.js'; import { appendTeamDeliveryLog } from '../../team/delivery-log.js'; import { hasUniqueScopedMailboxMatch, mailboxMessageContentMatches, mailboxMessageIdentity, mailboxRecordBelongsToTeam, normalizeBridgeMailboxMessage, readLegacyMailboxMessageIds } from '../../team/state/mailbox.js'; const __filename = fileURLToPath(import.meta.url); const __dirname = dirname(__filename); import { runProcess } from './process-runner.js'; import { resolvePaneTarget, resolveSessionToPane } from './tmux-injection.js'; import { EXACT_PANE_UNAVAILABLE_REASON, evaluatePaneInjectionReadiness, normalizeExactPaneId, sendPaneInput, verifyExactPaneLive, verifyExactPaneOwnerLive, } from './team-tmux-guard.js'; import { buildCapturePaneArgv, normalizeTmuxCapture, paneHasActiveTask, paneLooksReady, } from '../tmux-hook-engine.js'; /** * Route dispatch state transitions through the Rust runtime binary. * Non-fatal: if the binary is missing or fails, the legacy JSON fallback lane * remains available when the caller is already operating outside the bridge- * owned path. * Disable entirely with OMX_RUNTIME_BRIDGE=0. */ function runtimeExec(command, stateDir, team) { if (process.env.OMX_RUNTIME_BRIDGE === '0') return; try { getDefaultBridge(stateDir).execCommand(command); } catch (error) { const isSchemaFailure = error instanceof RuntimeBridgeError && error.context.command === 'schema'; const isTransportFailure = isRuntimeTransportError(error); const isFinalization = command.command === 'MarkNotified' || command.command === 'MarkDelivered' || command.command === 'MarkFailed'; if (isFinalization && !isSchemaFailure && !isTransportFailure && hasRuntimeDispatchIdentity(stateDir, command.request_id)) throw error; if (isFinalization && (isSchemaFailure || isTransportFailure)) { // Compatibility JSON alone is overwritten by runtime event replay. recordRuntimeFinalization(stateDir, team, command); } if (!isFinalization && (command.command === 'MarkMailboxNotified' || !isTransportFailure)) throw error; recordBridgeFallback({ stateDir, team, operation: 'runtimeExec', fallbackTarget: 'js_state_mutation', command: safeString(command?.command).trim() || 'unknown', requestId: safeString(command?.request_id).trim() || undefined, messageId: safeString(command?.message_id).trim() || undefined, reason: bridgeErrorReason(error), }); // non-fatal: JS path is the fallback } } function bridgeErrorReason(error) { const err = error || {}; const stderr = safeString(err.stderr).trim(); if (stderr) return stderr.slice(0, 500); const message = safeString(err.message).trim(); if (message) return message.slice(0, 500); return 'unknown_bridge_error'; } function bridgeFallbackLogPath(stateDir) { return join(dirname(stateDir), 'logs', `team-dispatch-${new Date().toISOString().slice(0, 10)}.jsonl`); } function recordBridgeFallback({ stateDir, team, operation, fallbackTarget, command, requestId, messageId, reason, }) { const event = { timestamp: new Date().toISOString(), type: 'bridge_fallback', source: 'notify-hook.team-dispatch', fallback: true, counter: 'team_dispatch_bridge_fallback_total', bridge_operation: operation, fallback_target: fallbackTarget, team: safeString(team).trim() || null, command: safeString(command).trim() || null, request_id: safeString(requestId).trim() || null, message_id: safeString(messageId).trim() || null, reason: safeString(reason).trim() || 'unknown_bridge_fallback', }; try { const logPath = bridgeFallbackLogPath(stateDir); mkdirSync(dirname(logPath), { recursive: true }); appendFileSync(logPath, `${JSON.stringify(event)}\n`); } catch { // best effort observability only } try { appendTeamDeliveryLog(join(dirname(stateDir), 'logs'), { event: 'bridge_fallback', ...event, }).catch(() => {}); } catch { // best effort observability only } try { process.emitWarning(`[omx] team-dispatch bridge fallback: ${event.bridge_operation} -> ${event.fallback_target}: ${event.reason}`, { code: 'OMX_TEAM_DISPATCH_BRIDGE_FALLBACK', }); } catch { // best effort observability only } } function readJson(path, fallback) { return readFile(path, 'utf8') .then((raw) => JSON.parse(raw)) .catch(() => fallback); } async function readBridgeDispatchRequests(stateDir, teamName, cwd, authoritativeRecords) { if (!authoritativeRecords && existsSync(join(stateDir, 'dispatch.json'))) { const compat = getDefaultBridge(stateDir).readCompatFile('dispatch.json'); if (!compat || !Array.isArray(compat.records)) { recordBridgeFallback({ stateDir, team: teamName, operation: 'readBridgeDispatchRequests', fallbackTarget: 'legacy_dispatch_requests', reason: compat ? 'invalid_dispatch_compat_schema' : 'parse_failed:invalid_dispatch_compat_json', }); } } return await readDispatchRequests(teamName, resolve(cwd), authoritativeRecords, stateDir); } async function writeJsonAtomic(path, value) { await mkdir(dirname(path), { recursive: true }); const tmp = `${path}.tmp-${process.pid}-${Date.now()}-${Math.random().toString(16).slice(2)}`; await writeFile(tmp, JSON.stringify(value, null, 2)); await rename(tmp, path); } // Keep stale-timeout semantics aligned with src/team/state.ts LOCK_STALE_MS. const DISPATCH_LOCK_STALE_MS = 5 * 60 * 1000; const DISPATCH_REQUEST_LEASE_STALE_MS = 30 * 1000; const DEFAULT_ISSUE_DISPATCH_COOLDOWN_MS = 15 * 60 * 1000; const ISSUE_DISPATCH_COOLDOWN_ENV = 'OMX_TEAM_DISPATCH_ISSUE_COOLDOWN_MS'; const DEFAULT_DISPATCH_TRIGGER_COOLDOWN_MS = 30 * 1000; const DISPATCH_TRIGGER_COOLDOWN_ENV = 'OMX_TEAM_DISPATCH_TRIGGER_COOLDOWN_MS'; const LEADER_PANE_MISSING_DEFERRED_REASON = 'leader_pane_missing_deferred'; const LEADER_NOTIFICATION_DEFERRED_TYPE = 'leader_notification_deferred'; async function emitOperationalHookEvent(cwd, eventName, context) { try { const { buildNativeHookEvent } = await import('../../hooks/extensibility/events.js'); const { dispatchHookEvent } = await import('../../hooks/extensibility/dispatcher.js'); const event = buildNativeHookEvent(eventName, { normalized_event: eventName, scope: 'team-dispatch', ...context, }); await dispatchHookEvent(event, { cwd }); } catch { // best effort only } } function mapDispatchFailureToCanonicalEvent(reason) { const normalized = safeString(reason).toLowerCase(); if (!normalized) return 'run.blocked_on_system'; if (normalized.includes('missing') || normalized.includes('deferred') || normalized.includes('rejected')) { return 'run.blocked_on_user'; } return 'run.blocked_on_system'; } function resolveIssueDispatchCooldownMs(env = process.env) { const raw = safeString(env[ISSUE_DISPATCH_COOLDOWN_ENV]).trim(); if (raw === '') return DEFAULT_ISSUE_DISPATCH_COOLDOWN_MS; const parsed = Number.parseInt(raw, 10); if (!Number.isFinite(parsed) || parsed < 0) return DEFAULT_ISSUE_DISPATCH_COOLDOWN_MS; return parsed; } function resolveDispatchTriggerCooldownMs(env = process.env) { const raw = safeString(env[DISPATCH_TRIGGER_COOLDOWN_ENV]).trim(); if (raw === '') return DEFAULT_DISPATCH_TRIGGER_COOLDOWN_MS; const parsed = Number.parseInt(raw, 10); if (!Number.isFinite(parsed) || parsed < 0) return DEFAULT_DISPATCH_TRIGGER_COOLDOWN_MS; return parsed; } function extractIssueKey(triggerMessage) { const match = safeString(triggerMessage).match(/\b([A-Z][A-Z0-9]+-\d+)\b/i); return match?.[1]?.toUpperCase() || null; } function issueCooldownStatePath(teamDirPath) { return join(teamDirPath, 'dispatch', 'issue-cooldown.json'); } function triggerCooldownStatePath(teamDirPath) { return join(teamDirPath, 'dispatch', 'trigger-cooldown.json'); } async function readIssueCooldownState(teamDirPath) { const fallback = { by_issue: {} }; const parsed = await readJson(issueCooldownStatePath(teamDirPath), fallback); if (!parsed || typeof parsed !== 'object' || typeof parsed.by_issue !== 'object' || parsed.by_issue === null) { return fallback; } return parsed; } async function readTriggerCooldownState(teamDirPath) { const fallback = { by_trigger: {} }; const parsed = await readJson(triggerCooldownStatePath(teamDirPath), fallback); if (!parsed || typeof parsed !== 'object' || typeof parsed.by_trigger !== 'object' || parsed.by_trigger === null) { return fallback; } return parsed; } function normalizeTriggerKey(value) { return safeString(value).replace(/\s+/g, ' ').trim(); } function parseTriggerCooldownEntry(entry) { if (typeof entry === 'number') { return { at: entry, lastRequestId: '' }; } if (!entry || typeof entry !== 'object') { return { at: NaN, lastRequestId: '' }; } return { at: Number(entry.at), lastRequestId: safeString(entry.last_request_id).trim(), }; } function reserveDispatchCooldowns({ issueCooldownMs, triggerCooldownMs, issueCooldownByIssue, triggerCooldownByKey, issueKey, triggerKey, requestId, reservedAt = Date.now(), }) { let mutated = false; if (issueKey && issueCooldownMs > 0) { issueCooldownByIssue[issueKey] = reservedAt; mutated = true; } if (triggerKey && triggerCooldownMs > 0) { triggerCooldownByKey[triggerKey] = { at: reservedAt, last_request_id: safeString(requestId).trim(), }; mutated = true; } return mutated; } async function withLockDirectory(lockDir, timeoutError, fn) { const ownerPath = join(lockDir, 'owner'); const ownerToken = `${process.pid}.${Date.now()}.${Math.random().toString(16).slice(2)}`; const deadline = Date.now() + 5_000; await mkdir(dirname(lockDir), { recursive: true }); while (true) { try { await mkdir(lockDir, { recursive: false }); try { await writeFile(ownerPath, ownerToken, 'utf8'); } catch (error) { await rm(lockDir, { recursive: true, force: true }); throw error; } break; } catch (error) { if (error?.code !== 'EEXIST') throw error; try { const info = await stat(lockDir); if (Date.now() - info.mtimeMs > DISPATCH_LOCK_STALE_MS) { await rm(lockDir, { recursive: true, force: true }); continue; } } catch { // best effort } if (Date.now() > deadline) throw new Error(timeoutError); await new Promise((resolveDelay) => setTimeout(resolveDelay, 25)); } } try { return await fn(); } finally { try { const currentOwner = await readFile(ownerPath, 'utf8'); if (currentOwner.trim() === ownerToken) { await rm(lockDir, { recursive: true, force: true }); } } catch { // best effort } } } async function withDispatchLock(teamDirPath, fn) { return await withLockDirectory( join(teamDirPath, 'dispatch', '.lock'), `Timed out acquiring dispatch lock for ${teamDirPath}`, fn, ); } async function withMailboxLock(teamDirPath, workerName, fn) { return await withLockDirectory( join(teamDirPath, 'mailbox', `.lock-${workerName}`), `Timed out acquiring mailbox lock for ${teamDirPath}/${workerName}`, fn, ); } function dispatchRequestLeaseDir(teamDirPath, requestId) { return join(teamDirPath, 'dispatch', `.processing-${safeString(requestId).trim()}`); } async function tryAcquireDispatchRequestLease(teamDirPath, requestId) { const lockDir = dispatchRequestLeaseDir(teamDirPath, requestId); const ownerPath = join(lockDir, 'owner'); const ownerToken = `${process.pid}.${Date.now()}.${Math.random().toString(16).slice(2)}`; await mkdir(dirname(lockDir), { recursive: true }); while (true) { try { await mkdir(lockDir, { recursive: false }); await writeFile(ownerPath, ownerToken, 'utf8'); return { lockDir, ownerPath, ownerToken, requestId }; } catch (error) { if (error?.code !== 'EEXIST') throw error; try { const info = await stat(lockDir); if (Date.now() - info.mtimeMs > DISPATCH_REQUEST_LEASE_STALE_MS) { await rm(lockDir, { recursive: true, force: true }); continue; } } catch { // best effort } return null; } } } async function releaseDispatchRequestLease(lease) { if (!lease?.lockDir || !lease?.ownerPath || !lease?.ownerToken) return; try { const currentOwner = await readFile(lease.ownerPath, 'utf8'); if (currentOwner.trim() === lease.ownerToken) { await rm(lease.lockDir, { recursive: true, force: true }); } } catch { // best effort } } function resolveLeaderPaneId(config) { return safeString(config?.leader_pane_id).trim(); } function explicitPaneIdentity(value) { const rawPaneId = safeString(value).trim(); return { provided: rawPaneId !== '', paneId: normalizeExactPaneId(rawPaneId), rawPaneId }; } function positivePanePid(value) { return Number.isInteger(value) && Number(value) > 0 ? Number(value) : undefined; } function resolveAddressedWorker(request, config) { const workers = Array.isArray(config?.workers) ? config.workers : []; if (Number.isFinite(request?.worker_index)) { const worker = workers.find((candidate) => Number(candidate?.index) === Number(request.worker_index)); if (worker) return worker; } const recipient = safeString(request?.to_worker).trim(); return workers.find((candidate) => safeString(candidate?.name).trim() === recipient) || null; } function resolveConfiguredPaneIdentity(request, config) { if (request.to_worker === 'leader-fixed') { return { source: 'leader_pane_id', expectedPanePid: positivePanePid(config?.leader_pane_pid), expectedPaneOwnerId: safeString(config?.tmux_pane_owner_id).trim(), ...explicitPaneIdentity(config?.leader_pane_id) }; } const worker = resolveAddressedWorker(request, config); const expectedPanePid = positivePanePid(worker?.pid); return { source: 'worker_pane_id', expectedPanePid, expectedPaneOwnerId: safeString(config?.tmux_pane_owner_id).trim(), ...explicitPaneIdentity(worker?.pane_id) }; } function resolveDispatchTarget(request, config) { const requestPane = explicitPaneIdentity(request?.pane_id); const configuredPane = resolveConfiguredPaneIdentity(request, config); const hudPaneId = explicitPaneIdentity(config?.hud_pane_id).paneId; if (requestPane.provided) { if (!requestPane.paneId) { return { failure: 'invalid_explicit_pane', paneId: requestPane.paneId, paneSource: 'request_pane_id', rawPaneId: requestPane.rawPaneId }; } if (requestPane.paneId === hudPaneId) { return { failure: 'hud_pane_target', paneId: requestPane.paneId, paneSource: 'request_pane_id' }; } if (configuredPane.provided) { if (!configuredPane.paneId) { return { failure: 'invalid_explicit_pane', paneId: configuredPane.paneId, paneSource: configuredPane.source, rawPaneId: configuredPane.rawPaneId }; } if (requestPane.paneId !== configuredPane.paneId) { return { failure: 'exact_pane_mismatch', paneId: requestPane.paneId, paneSource: 'request_pane_id', configuredPaneId: configuredPane.paneId, }; } } if (!positivePanePid(configuredPane.expectedPanePid)) { return { failure: 'missing_exact_pane_pid', paneId: requestPane.paneId, paneSource: configuredPane.source }; } if (!configuredPane.expectedPaneOwnerId) { return { failure: 'missing_expected_pane_owner', paneId: requestPane.paneId, paneSource: configuredPane.source }; } return { target: { type: 'pane', value: requestPane.paneId }, exactPaneId: requestPane.paneId, expectedPanePid: configuredPane.expectedPanePid, expectedPaneOwnerId: configuredPane.expectedPaneOwnerId, expectedHudPaneId: hudPaneId, source: 'request_pane_id', reason: 'explicit_request_pane_id' }; } if (configuredPane.provided) { if (!configuredPane.paneId) { return { failure: 'invalid_explicit_pane', paneId: configuredPane.paneId, paneSource: configuredPane.source, rawPaneId: configuredPane.rawPaneId }; } if (configuredPane.paneId === hudPaneId) { return { failure: 'hud_pane_target', paneId: configuredPane.paneId, paneSource: configuredPane.source }; } if (!positivePanePid(configuredPane.expectedPanePid)) { return { failure: 'missing_exact_pane_pid', paneId: configuredPane.paneId, paneSource: configuredPane.source }; } if (!configuredPane.expectedPaneOwnerId) { return { failure: 'missing_expected_pane_owner', paneId: configuredPane.paneId, paneSource: configuredPane.source }; } return { target: { type: 'pane', value: configuredPane.paneId }, exactPaneId: configuredPane.paneId, expectedPanePid: configuredPane.expectedPanePid, expectedPaneOwnerId: configuredPane.expectedPaneOwnerId, expectedHudPaneId: hudPaneId, source: configuredPane.source, reason: configuredPane.source }; } return { target: null }; } async function appendLeaderNotificationDeferredEvent({ stateDir, teamName, request, reason, nowIso, tmuxSession = '', leaderPaneId = '', sourceType = 'team_dispatch', }) { const eventsDir = join(stateDir, 'team', teamName, 'events'); const eventsPath = join(eventsDir, 'events.ndjson'); const event = { event_id: `leader-deferred-${Date.now()}-${Math.random().toString(16).slice(2, 8)}`, team: teamName, type: LEADER_NOTIFICATION_DEFERRED_TYPE, worker: request.to_worker, to_worker: request.to_worker, reason, created_at: nowIso, request_id: request.request_id, ...(request.message_id ? { message_id: request.message_id } : {}), tmux_session: tmuxSession || null, leader_pane_id: leaderPaneId || null, tmux_injection_attempted: false, source_type: sourceType, }; await mkdir(eventsDir, { recursive: true }).catch(() => {}); await appendFile(eventsPath, JSON.stringify(event) + '\n').catch(() => {}); } async function finalizeClaimedDispatchRequest({ claim, result, teamName, teamDirPath, config, cwd, stateDir, logsDir, issueCooldownMs, triggerCooldownMs, }) { const requestsPath = join(teamDirPath, 'dispatch', 'requests.json'); const issueKey = extractIssueKey(claim.request.trigger_message); const triggerKey = normalizeTriggerKey(claim.request.trigger_message); let summary = { processed: 0, skipped: 0, failed: 0 }; await withDispatchLock(teamDirPath, async () => { const records = getDefaultBridge(stateDir).readCompatFile('dispatch.json')?.records; // A live lease may finalize its preflight snapshot when injection damages the projection. const requests = await readBridgeDispatchRequests(stateDir, teamName, cwd, Array.isArray(records) && records.every(isStrictDispatchRecord) ? records : claim.authoritativeRecords); if (!Array.isArray(requests)) return; const index = requests.findIndex((entry) => safeString(entry?.request_id).trim() === safeString(claim.request.request_id).trim()); if (index < 0) return; const request = requests[index]; if (!request || typeof request !== 'object' || shouldSkipRequest(request) || request.status !== 'pending') return; const issueCooldownState = await readIssueCooldownState(teamDirPath); const triggerCooldownState = await readTriggerCooldownState(teamDirPath); const issueCooldownByIssue = issueCooldownState.by_issue || {}; const triggerCooldownByKey = triggerCooldownState.by_trigger || {}; const nowIso = new Date().toISOString(); let mutated = false; mutated = reserveDispatchCooldowns({ issueCooldownMs, triggerCooldownMs, issueCooldownByIssue, triggerCooldownByKey, issueKey, triggerKey, requestId: request.request_id, }) || mutated; request.attempt_count = Number.isFinite(request.attempt_count) ? Math.max(0, request.attempt_count + 1) : 1; request.updated_at = nowIso; if (result.ok) { const MAX_UNCONFIRMED_ATTEMPTS = 3; if (result.reason === 'tmux_send_keys_unconfirmed' && request.attempt_count < MAX_UNCONFIRMED_ATTEMPTS) { request.last_reason = result.reason; summary.skipped += 1; mutated = true; await appendDispatchLog(logsDir, { type: 'dispatch_unconfirmed_retry', team: teamName, request_id: request.request_id, worker: request.to_worker, attempt: request.attempt_count, reason: result.reason, ...buildDispatchAttemptEvidence(result), }); await appendDeliveryTelemetry(logsDir, { event: 'dispatch_result', team: teamName, request_id: request.request_id, message_id: request.message_id || null, to_worker: request.to_worker, transport: 'send-keys', result: 'retry', reason: result.reason, }); await emitOperationalHookEvent(cwd, 'retry-needed', { team: teamName, worker: request.to_worker, request_id: request.request_id, attempt: request.attempt_count, command: request.trigger_message, reason: result.reason, status: 'retry-needed', }); } else if (result.reason === 'tmux_send_keys_unconfirmed') { request.status = 'failed'; request.failed_at = nowIso; request.last_reason = 'unconfirmed_after_max_retries'; runtimeExec({ command: 'MarkFailed', request_id: request.request_id, reason: 'unconfirmed_after_max_retries' }, stateDir, teamName); summary.processed += 1; summary.failed += 1; mutated = true; await appendDispatchLog(logsDir, { type: 'dispatch_failed', team: teamName, request_id: request.request_id, worker: request.to_worker, message_id: request.message_id || null, reason: request.last_reason, ...buildDispatchAttemptEvidence(result), }); await appendDeliveryTelemetry(logsDir, { event: 'dispatch_result', team: teamName, request_id: request.request_id, message_id: request.message_id || null, to_worker: request.to_worker, transport: 'send-keys', result: 'failed', reason: request.last_reason, }); await emitOperationalHookEvent(cwd, 'failed', { team: teamName, worker: request.to_worker, request_id: request.request_id, message_id: request.message_id || null, command: request.trigger_message, reason: request.last_reason, error_summary: request.last_reason, status: 'failed', }); } else { if (result.delivery_confirmed === true && process.env.OMX_RUNTIME_BRIDGE !== '0' && hasRuntimeDispatchIdentity(stateDir, request.request_id)) { // Persist a no-reinject receipt before runtime finalization can reject semantically. recordRuntimeFinalization(stateDir, teamName, { command: 'MarkNotified', request_id: request.request_id, channel: 'tmux', }); recordRuntimeFinalization(stateDir, teamName, { command: 'MarkDelivered', request_id: request.request_id, }); } request.status = 'notified'; request.notified_at = nowIso; request.last_reason = result.reason; runtimeExec({ command: 'MarkNotified', request_id: request.request_id, channel: 'tmux' }, stateDir, teamName); if (request.kind === 'mailbox' && request.message_id) { await updateMailboxNotified(stateDir, teamName, request.to_worker, request.message_id).catch(() => {}); } if (result.delivery_confirmed === true) { runtimeExec({ command: 'MarkDelivered', request_id: request.request_id }, stateDir, teamName); request.status = 'delivered'; request.delivered_at = nowIso; } summary.processed += 1; mutated = true; await appendDispatchLog(logsDir, { type: result.delivery_confirmed === true ? 'dispatch_delivered' : 'dispatch_notified', team: teamName, request_id: request.request_id, worker: request.to_worker, message_id: request.message_id || null, reason: result.reason, ...buildDispatchAttemptEvidence(result), }); await appendDeliveryTelemetry(logsDir, { event: 'dispatch_result', team: teamName, request_id: request.request_id, message_id: request.message_id || null, to_worker: request.to_worker, transport: 'send-keys', result: result.delivery_confirmed === true ? 'delivered' : 'notified', reason: result.reason, }); } } else { request.status = 'failed'; request.failed_at = nowIso; request.last_reason = result.reason; runtimeExec({ command: 'MarkFailed', request_id: request.request_id, reason: result.reason }, stateDir, teamName); summary.processed += 1; summary.failed += 1; mutated = true; await appendDispatchLog(logsDir, { type: 'dispatch_failed', team: teamName, request_id: request.request_id, worker: request.to_worker, message_id: request.message_id || null, reason: result.reason, ...buildDispatchAttemptEvidence(result), }); await appendDeliveryTelemetry(logsDir, { event: 'dispatch_result', team: teamName, request_id: request.request_id, message_id: request.message_id || null, to_worker: request.to_worker, transport: 'send-keys', result: 'failed', reason: result.reason, }); await emitOperationalHookEvent(cwd, result.reason === LEADER_PANE_MISSING_DEFERRED_REASON ? 'handoff-needed' : 'failed', { team: teamName, worker: request.to_worker, request_id: request.request_id, message_id: request.message_id || null, command: request.trigger_message, reason: result.reason, ...(result.reason === LEADER_PANE_MISSING_DEFERRED_REASON ? { status: 'handoff-needed' } : { status: 'failed', error_summary: result.reason }), }); } if (!mutated) return; issueCooldownState.by_issue = issueCooldownByIssue; await writeJsonAtomic(issueCooldownStatePath(teamDirPath), issueCooldownState); triggerCooldownState.by_trigger = triggerCooldownByKey; await writeJsonAtomic(triggerCooldownStatePath(teamDirPath), triggerCooldownState); await writeJsonAtomic(requestsPath, requests); }); return summary; } function resolveWorkerCliForRequest(request, config) { const workers = Array.isArray(config?.workers) ? config.workers : []; const idx = Number.isFinite(request?.worker_index) ? Number(request.worker_index) : null; if (idx !== null) { const worker = workers.find((candidate) => Number(candidate?.index) === idx); const workerCli = safeString(worker?.worker_cli).trim().toLowerCase(); if (workerCli === 'claude') return 'claude'; } return 'codex'; } function capturedPaneContainsTrigger(captured, trigger) { if (!captured || !trigger) return false; const normalizeForDraftMatch = (value) => normalizeTmuxCapture(value).replace(/-\s+/g, '-'); return normalizeForDraftMatch(captured).includes(normalizeForDraftMatch(trigger)); } function capturedPaneContainsTriggerNearTail(captured, trigger, nonEmptyTailLines = 24) { if (!captured || !trigger) return false; const normalizeForDraftMatch = (value) => normalizeTmuxCapture(value).replace(/-\s+/g, '-'); const normalizedTrigger = normalizeForDraftMatch(trigger); if (!normalizedTrigger) return false; const lines = safeString(captured) .split('\n') .map((line) => line.replace(/\r/g, '').trim()) .filter((line) => line.length > 0); if (lines.length === 0) return false; const tail = lines.slice(-Math.max(1, nonEmptyTailLines)).join(' '); return normalizeForDraftMatch(tail).includes(normalizedTrigger); } function buildJoinedCapturePaneArgv(paneTarget, tailLines = 80) { // Join wrapped visual lines so long path-like trigger text split by tmux // remains comparable with the original trigger. return ['capture-pane', '-J', '-t', paneTarget, '-p', '-S', `-${tailLines}`]; } const INJECT_VERIFY_DELAY_MS = 250; const INJECT_VERIFY_ROUNDS = 3; async function injectDispatchRequest(request, config, cwd, stateDir) { const dispatchTarget = resolveDispatchTarget(request, config); if (dispatchTarget.failure) { if (dispatchTarget.failure !== 'exact_pane_mismatch' && dispatchTarget.failure !== 'invalid_explicit_pane') { return { ok: false, reason: dispatchTarget.failure, pane: dispatchTarget.paneId || null, pane_source: dispatchTarget.paneSource || null, exact_pane_proof: null, tmux_injection_attempted: false, }; } const paneProof = dispatchTarget.failure === 'exact_pane_mismatch' ? { reason: 'exact_pane_mismatch', proof: { status: 'unavailable', paneId: dispatchTarget.paneId, reason: 'pane_mismatch', detail: `configured pane is ${dispatchTarget.configuredPaneId}`, }, } : await verifyExactPaneLive(dispatchTarget.rawPaneId); return { ok: false, reason: paneProof.reason, pane: dispatchTarget.paneId || null, pane_source: dispatchTarget.paneSource || null, exact_pane_proof: paneProof.proof || null, tmux_injection_attempted: false, }; } const target = dispatchTarget.target; if (!target) { return { ok: false, reason: 'missing_tmux_target' }; } const leaderTargeted = request.to_worker === 'leader-fixed'; let resolution; if (dispatchTarget.exactPaneId) { resolution = { paneTarget: dispatchTarget.exactPaneId, reason: dispatchTarget.reason, source: dispatchTarget.source, }; } else if (target.type === 'session') { const paneId = await resolveSessionToPane(target.value).catch(() => null); resolution = paneId ? { paneTarget: paneId, reason: 'session_target_resolved' } : { paneTarget: null, reason: 'target_not_found' }; } else { resolution = await resolvePaneTarget(target, '', '', '', {}); } if (!resolution.paneTarget) { return { ok: false, reason: `target_resolution_failed:${resolution.reason}` }; } const exactPaneId = dispatchTarget.exactPaneId || normalizeExactPaneId(resolution.paneTarget); const exactPaneFailure = (paneProof, tmuxInjectionAttempted) => ({ ok: false, reason: paneProof.reason, pane: resolution.paneTarget, pane_source: resolution.source || null, exact_pane_proof: paneProof.proof || null, tmux_injection_attempted: tmuxInjectionAttempted, }); const paneGuard = await evaluatePaneInjectionReadiness(resolution.paneTarget, { skipIfScrolling: true, requireRunningAgent: leaderTargeted, requireReady: false, requireIdle: false, requireObservableState: leaderTargeted, exactPaneId, expectedPanePid: dispatchTarget.expectedPanePid, expectedPaneOwnerId: dispatchTarget.expectedPaneOwnerId, expectedHudPaneId: dispatchTarget.expectedHudPaneId, }); if (!paneGuard.ok) { return { ok: false, reason: paneGuard.reason, pane: resolution.paneTarget, pane_source: resolution.source || null, readiness_evidence: paneGuard.readinessEvidence || null, pane_current_command: paneGuard.paneCurrentCommand || null, exact_pane_proof: paneGuard.exactPaneProof || null, tmux_injection_attempted: false, }; } let exactPaneProof = paneGuard.exactPaneProof || null; const verifyExplicitPane = async () => { const paneProof = await verifyExactPaneOwnerLive(exactPaneId, dispatchTarget.expectedPanePid, dispatchTarget.expectedPaneOwnerId); exactPaneProof = paneProof.proof || null; return paneProof; }; const attemptCountAtStart = Number.isFinite(request.attempt_count) ? Math.max(0, Math.floor(request.attempt_count)) : 0; const submitKeyPresses = resolveWorkerCliForRequest(request, config) === 'claude' ? 1 : 2; let preCaptureHasTrigger = false; if (attemptCountAtStart >= 1) { const preCaptureProof = await verifyExplicitPane(); if (!preCaptureProof.ok) return exactPaneFailure(preCaptureProof, false); try { // Narrow capture (8 lines) to scope check to input area, not scrollback output const preCapture = await runProcess('tmux', buildJoinedCapturePaneArgv(resolution.paneTarget, 8), 2000); preCaptureHasTrigger = capturedPaneContainsTrigger(preCapture.stdout, request.trigger_message); } catch { preCaptureHasTrigger = false; } } // Retype whenever trigger text is NOT in the narrow input area, regardless of attempt count. // Pre-0.7.4 bug: 80-line capture matched trigger in scrollback output, falsely skipping retype. const shouldTypePrompt = attemptCountAtStart === 0 || !preCaptureHasTrigger; if (shouldTypePrompt && attemptCountAtStart >= 1) { // Clear stale text in input buffer before retyping (mirrors sync path tmux-session.ts:1270) const clearProof = await verifyExplicitPane(); if (!clearProof.ok) return exactPaneFailure(clearProof, false); await runProcess('tmux', ['send-keys', '-t', resolution.paneTarget, 'C-u'], 1000).catch(() => {}); await new Promise((r) => setTimeout(r, 50)); } const sendResult = await sendPaneInput({ paneTarget: resolution.paneTarget, exactPaneId, expectedPanePid: dispatchTarget.expectedPanePid, expectedPaneOwnerId: dispatchTarget.expectedPaneOwnerId, expectedHudPaneId: dispatchTarget.expectedHudPaneId, prompt: request.trigger_message, submitKeyPresses, submitKey: 'Enter', typePrompt: shouldTypePrompt, queueFirstSubmit: leaderTargeted, }); if (!sendResult.ok) { return { ok: false, reason: sendResult.error || sendResult.reason, pane: resolution.paneTarget, pane_source: resolution.source || null, readiness_evidence: paneGuard.readinessEvidence || null, pane_current_command: paneGuard.paneCurrentCommand || null, exact_pane_proof: sendResult.exactPaneProof || null, tmux_injection_attempted: true, }; } exactPaneProof = sendResult.exactPaneProof || exactPaneProof; // Post-injection verification: confirm the trigger text was consumed. // Fixes #391: without this, dispatch marks 'notified' even when the worker // pane is sitting on an unsent draft (Enter was not effectively applied). const verifyNarrowArgv = buildJoinedCapturePaneArgv(resolution.paneTarget, 8); const verifyWideArgv = buildJoinedCapturePaneArgv(resolution.paneTarget); for (let round = 0; round < INJECT_VERIFY_ROUNDS; round++) { await new Promise((r) => setTimeout(r, INJECT_VERIFY_DELAY_MS)); const narrowProof = await verifyExplicitPane(); if (!narrowProof.ok) return exactPaneFailure(narrowProof, true); let narrowCap; try { // Primary: trigger text no longer in narrow input area. // Secondary guard: also inspect the recent non-empty tail of wide capture. // This avoids false confirmations when Codex leaves the unsent draft just // above a large blank area (narrow capture misses it) while still avoiding // full-scrollback false positives. narrowCap = await runProcess('tmux', verifyNarrowArgv, 2000); } catch { // Capture failed; fall through to retry Enter. } if (narrowCap) { const wideProof = await verifyExplicitPane(); if (!wideProof.ok) return exactPaneFailure(wideProof, true); let wideCap; try { wideCap = await runProcess('tmux', verifyWideArgv, 2000); } catch { // Capture failed; fall through to retry Enter. } if (wideCap) { const triggerInNarrow = capturedPaneContainsTrigger(narrowCap.stdout, request.trigger_message); const triggerNearTail = capturedPaneContainsTriggerNearTail(wideCap.stdout, request.trigger_message); if (triggerInNarrow || triggerNearTail) { // Draft is still visible, so Enter has not actually submitted it yet. // Do not let transient spinner/active-task text mask an unsent draft. const retrySend = await sendPaneInput({ paneTarget: resolution.paneTarget, exactPaneId, expectedPanePid: dispatchTarget.expectedPanePid, expectedPaneOwnerId: dispatchTarget.expectedPaneOwnerId, expectedHudPaneId: dispatchTarget.expectedHudPaneId, prompt: request.trigger_message, submitKeyPresses, submitKey: 'Enter', typePrompt: false, }); if (!retrySend.ok && retrySend.reason === EXACT_PANE_UNAVAILABLE_REASON) { return exactPaneFailure({ reason: retrySend.reason, proof: retrySend.exactPaneProof }, true); } exactPaneProof = retrySend.exactPaneProof || exactPaneProof; continue; } // Worker is actively processing (mirrors sync path tmux-session.ts:1292-1294) if (paneHasActiveTask(wideCap.stdout)) { return { ok: true, reason: 'tmux_send_keys_confirmed_active_task', delivery_confirmed: true, pane: resolution.paneTarget, pane_source: resolution.source || null, readiness_evidence: paneGuard.readinessEvidence || null, pane_current_command: paneGuard.paneCurrentCommand || null, exact_pane_proof: exactPaneProof, tmux_injection_attempted: true, }; } // Do not declare success while a pane is not input-ready. Otherwise a // pre-ready send can be marked "confirmed" and later appear as a stuck // unsent draft once the UI finishes loading. This includes leader-fixed: // its Codex UI can show "tab to queue message" while busy, and marking // delivered before queue/consumption confirmation loses the orchestration // nudge until a human presses Tab manually. if (!paneLooksReady(wideCap.stdout)) { continue; } if (!triggerInNarrow && !triggerNearTail) { return { ok: true, reason: 'tmux_send_keys_confirmed', delivery_confirmed: true, pane: resolution.paneTarget, pane_source: resolution.source || null, readiness_evidence: paneGuard.readinessEvidence || null, pane_current_command: paneGuard.paneCurrentCommand || null, exact_pane_proof: exactPaneProof, tmux_injection_attempted: true, }; } } } // Draft still visible and no active task — retry Enter const retrySend = await sendPaneInput({ paneTarget: resolution.paneTarget, exactPaneId, expectedPanePid: dispatchTarget.expectedPanePid, expectedPaneOwnerId: dispatchTarget.expectedPaneOwnerId, expectedHudPaneId: dispatchTarget.expectedHudPaneId, prompt: request.trigger_message, submitKeyPresses, submitKey: 'Enter', typePrompt: false, queueFirstSubmit: leaderTargeted, }); if (!retrySend.ok && retrySend.reason === EXACT_PANE_UNAVAILABLE_REASON) { return exactPaneFailure({ reason: retrySend.reason, proof: retrySend.exactPaneProof }, true); } exactPaneProof = retrySend.exactPaneProof || exactPaneProof; } // Trigger text is still visible after all retry rounds. return { ok: true, reason: 'tmux_send_keys_unconfirmed', pane: resolution.paneTarget, pane_source: resolution.source || null, readiness_evidence: paneGuard.readinessEvidence || null, pane_current_command: paneGuard.paneCurrentCommand || null, exact_pane_proof: exactPaneProof, tmux_injection_attempted: true, }; } function shouldSkipRequest(request) { if (request.status !== 'pending') return true; const preference = safeString(request.transport_preference).trim(); return preference !== '' && preference !== 'hook_preferred_with_fallback'; } async function updateMailboxNotified(stateDir, teamName, workerName, messageId) { const teamDirPath = join(stateDir, 'team', teamName); const mailboxPath = join(teamDirPath, 'mailbox', `${workerName}.json`); return await withMailboxLock(teamDirPath, workerName, async () => { const mailbox = await readJson(mailboxPath, { worker: workerName, messages: [] }); if (!mailbox || mailbox.worker !== workerName || !Array.isArray(mailbox.messages)) return false; let msg = mailbox.messages.find((candidate) => candidate?.message_id === messageId && candidate.to_worker === workerName); if (process.env.OMX_RUNTIME_BRIDGE !== '0') { const records = getDefaultBridge(stateDir).readMailboxRecords(); const candidates = records.filter((candidate) => candidate.message_id === messageId && (candidate.team_name == null || candidate.team_name === teamName)); const legacyIds = candidates.length === 0 || candidates.some((record) => record.team_name == null || record.body === '' || mailbox.messages.some((message) => message?.message_id === messageId && message.created_at !== record.created_at)) ? await readLegacyMailboxMessageIds(teamName, workerName, mailbox, join(teamDirPath, 'events', 'events.ndjson'), records) : new Set(); const record = candidates.find((candidate) => mailboxRecordBelongsToTeam(candidate, teamName, workerName, legacyIds)); if (record) { const normalized = normalizeBridgeMailboxMessage(record, legacyIds.has(messageId) ? msg : undefined); msg = mailbox.messages.find((candidate) => mailboxMessageIdentity(candidate) === mailboxMessageIdentity(normalized) || ((legacyIds.has(messageId) || hasUniqueScopedMailboxMatch(teamName, normalized, records, mailbox.messages)) && mailboxMessageContentMatches(candidate, normalized))); if (!msg) { msg = normalized; mailbox.messages.push(msg); } runtimeExec({ command: 'MarkMailboxNotified', team_name: teamName, message_id: messageId }, stateDir, teamName); } else if (candidates.length > 0 || !legacyIds.has(messageId)) { return false; } } else { const legacyIds = await readLegacyMailboxMessageIds(teamName, workerName, mailbox, join(teamDirPath, 'events', 'events.ndjson')); if (!legacyIds.has(messageId)) return false; } if (!msg) return false; if (!msg.notified_at) msg.notified_at = new Date().toISOString(); await writeJsonAtomic(mailboxPath, mailbox); return true; }); } async function appendDispatchLog(logsDir, event) { const path = join(logsDir, `team-dispatch-${new Date().toISOString().slice(0, 10)}.jsonl`); await mkdir(logsDir, { recursive: true }).catch(() => {}); await appendFile(path, `${JSON.stringify({ timestamp: new Date().toISOString(), ...event })}\n`).catch(() => {}); } async function appendDeliveryTelemetry(logsDir, event) { await appendTeamDeliveryLog(logsDir, { source: 'notify-hook.team-dispatch', ...event, }).catch(() => {}); } function buildDispatchAttemptEvidence(result, fallback = {}) { return { pane_target: safeString(result?.pane || fallback.pane || '').trim() || null, pane_source: safeString(result?.pane_source || fallback.pane_source || '').trim() || null, readiness_evidence: safeString(result?.readiness_evidence || fallback.readiness_evidence || '').trim() || null, pane_current_command: safeString(result?.pane_current_command || fallback.pane_current_command || '').trim() || null, exact_pane_proof: result?.exact_pane_proof ?? fallback.exact_pane_proof ?? null, tmux_injection_attempted: typeof result?.tmux_injection_attempted === 'boolean' ? result.tmux_injection_attempted : (typeof fallback.tmux_injection_attempted === 'boolean' ? fallback.tmux_injection_attempted : null), }; } async function listTeamDirectories(teamRoot: string): Promise { const entries = await readdir(teamRoot, { withFileTypes: true }).catch(() => []); const names: string[] = []; for (const entry of entries) { if (entry.isDirectory()) { names.push(entry.name); continue; } if (!entry.isSymbolicLink()) continue; const resolved = await stat(join(teamRoot, entry.name)).catch(() => null); if (resolved?.isDirectory()) names.push(entry.name); } return names; } // Cache only a projection verified against unchanged event contents and state versions. const refreshedDispatchStates = new Map(); async function refreshDispatchProjection(stateDir) { const readStateVersion = async () => Promise.all(['events.json', 'dispatch.json'].map((name) => stat(join(stateDir, name), { bigint: true }).then((info) => [info.dev, info.ino, info.size, info.mtimeNs, info.ctimeNs].map(String).join(':')).catch((error) => { if (error.code === 'ENOENT') return null; throw error; }))); const before = await readStateVersion(); const cached = refreshedDispatchStates.get(stateDir); if (cached && cached.state[0] === before[0] && cached.state[1] === before[1]) return cached.records; refreshedDispatchStates.delete(stateDir); const eventsPath = join(stateDir, 'events.json'); const readEvents = () => readFile(eventsPath, 'utf8').catch((error) => { if (error.code === 'ENOENT') return null; throw error; }); const eventsBefore = await readEvents(); const snapshot = getDefaultBridge(stateDir).execCommand({ command: 'CaptureSnapshot' }); if (!Array.isArray(snapshot.dispatch_records)) return; // Older compatible runtimes do not return records. const eventsAfter = await readEvents(); const after = await readStateVersion(); let projected; try { projected = JSON.parse(await readFile(join(stateDir, 'dispatch.json'), 'utf8')).records; } catch { /* A concurrent writer cannot prove a cache entry. */ } if (eventsBefore === eventsAfter && isDeepStrictEqual(projected, snapshot.dispatch_records)) { refreshedDispatchStates.set(stateDir, { state: after, records: snapshot.dispatch_records }); } return snapshot.dispatch_records; } export async function drainPendingTeamDispatch({ cwd, stateDir = resolveBridgeStateDir(cwd), logsDir = join(cwd, '.omx', 'logs'), maxPerTick = 5, injector = injectDispatchRequest, }: { cwd?: string; stateDir?: string; logsDir?: string; maxPerTick?: number; injector?: typeof injectDispatchRequest; } = {}) { if (safeString(process.env.OMX_TEAM_WORKER)) { return { processed: 0, skipped: 0, failed: 0, reason: 'worker_context' }; } const teamRoot = join(stateDir, 'team'); if (!existsSync(teamRoot)) return { processed: 0, skipped: 0, failed: 0 }; // `.omx/state/team` also holds plain files (e.g. `notice-ledger.json`). // Only real directories are Team roots; treating a file as one makes every // dispatch-state write fail with ENOTDIR on `/dispatch`. const teams = await listTeamDirectories(teamRoot); let processed = 0; let skipped = 0; let failed = 0; const issueCooldownMs = resolveIssueDispatchCooldownMs(); const triggerCooldownMs = resolveDispatchTriggerCooldownMs(); for (const teamName of teams) { if (processed >= maxPerTick) break; const teamDirPath = join(teamRoot, teamName); const manifestPath = join(teamDirPath, 'manifest.v2.json'); const configPath = join(teamDirPath, 'config.json'); const requestsPath = join(teamDirPath, 'dispatch', 'requests.json'); const config = await readJson(existsSync(manifestPath) ? manifestPath : configPath, {}); const claims = []; await withDispatchLock(teamDirPath, async () => { await reconcileRuntimeFinalizations(stateDir, teamName); let authoritativeRecords; if (hasRuntimeAuthority(stateDir)) { if (process.env.OMX_RUNTIME_BRIDGE === '0') throw new Error('runtime dispatch authority requires an enabled bridge'); authoritativeRecords = await refreshDispatchProjection(stateDir); if (!authoritativeRecords) { const records = getDefaultBridge(stateDir).readCompatFile('dispatch.json')?.records; if (Array.isArray(records)) authoritativeRecords = records; } } const requests = await readBridgeDispatchRequests(stateDir, teamName, cwd, authoritativeRecords); if (!Array.isArray(requests)) return; if (requests.some((request) => request && typeof request === 'object' && !shouldSkipRequest(request))) { preflightRuntimeBridge(stateDir); } const issueCooldownState = await readIssueCooldownState(teamDirPath); const triggerCooldownState = await readTriggerCooldownState(teamDirPath); const issueCooldownByIssue = issueCooldownState.by_issue || {}; const triggerCooldownByKey = triggerCooldownState.by_trigger || {}; const nowMs = Date.now(); let mutated = false; for (const request of requests) { if (processed + claims.length >= maxPerTick) break; if (!request || typeof request !== 'object') continue; if (shouldSkipRequest(request)) { skipped += 1; continue; } if (request.to_worker === 'leader-fixed' && !resolveLeaderPaneId(config) && !explicitPaneIdentity(request.pane_id).provided) { const nowIso = new Date().toISOString(); const alreadyDeferred = safeString(request.last_reason).trim() === LEADER_PANE_MISSING_DEFERRED_REASON; request.updated_at = nowIso; request.last_reason = LEADER_PANE_MISSING_DEFERRED_REASON; request.status = 'pending'; skipped += 1; mutated = true; if (!alreadyDeferred) { await appendDispatchLog(logsDir, { type: 'dispatch_deferred', team: teamName, request_id: request.request_id, worker: request.to_worker, to_worker: request.to_worker, message_id: request.message_id || null, reason: LEADER_PANE_MISSING_DEFERRED_REASON, status: 'pending', tmux_session: safeString(config?.tmux_session).trim() || null, leader_pane_id: safeString(config?.leader_pane_id).trim() || null, tmux_injection_attempted: false, pane_target: null, pane_source: null, readiness_evidence: null, pane_current_command: null, }); await appendDeliveryTelemetry(logsDir, { event: 'dispatch_result', team: teamName, request_id: request.request_id, message_id: request.message_id || null, to_worker: request.to_worker, transport: 'send-keys', result: 'deferred', reason: LEADER_PANE_MISSING_DEFERRED_REASON, }); await appendLeaderNotificationDeferredEvent({ stateDir, teamName, request, reason: LEADER_PANE_MISSING_DEFERRED_REASON, nowIso, tmuxSession: safeString(config?.tmux_session).trim(), leaderPaneId: safeString(config?.leader_pane_id).trim(), sourceType: 'team_dispatch', }); } continue; } const issueKey = extractIssueKey(request.trigger_message); if (issueCooldownMs > 0 && issueKey) { const lastInjectedMs = Number(issueCooldownByIssue[issueKey]); if (Number.isFinite(lastInjectedMs) && lastInjectedMs > 0 && nowMs - lastInjectedMs < issueCooldownMs) { skipped += 1; continue; } } const triggerKey = normalizeTriggerKey(request.trigger_message); if (triggerCooldownMs > 0 && triggerKey) { const parsed = parseTriggerCooldownEntry(triggerCooldownByKey[triggerKey]); const withinCooldown = Number.isFinite(parsed.at) && parsed.at > 0 && nowMs - parsed.at < triggerCooldownMs; const sameRequestRetry = parsed.lastRequestId !== '' && parsed.lastRequestId === safeString(request.request_id).trim(); if (withinCooldown && !sameRequestRetry) { skipped += 1; continue; } } const lease = await tryAcquireDispatchRequestLease(teamDirPath, request.request_id); if (!lease) { skipped += 1; continue; } mutated = reserveDispatchCooldowns({ issueCooldownMs, triggerCooldownMs, issueCooldownByIssue, triggerCooldownByKey, issueKey, triggerKey, requestId: request.request_id, }) || mutated; claims.push({ request: { ...request }, authoritativeRecords, lease, }); } if (mutated) { issueCooldownState.by_issue = issueCooldownByIssue; await writeJsonAtomic(issueCooldownStatePath(teamDirPath), issueCooldownState); triggerCooldownState.by_trigger = triggerCooldownByKey; await writeJsonAtomic(triggerCooldownStatePath(teamDirPath), triggerCooldownState); await writeJsonAtomic(requestsPath, requests); } }); try { for (const claim of claims) { try { const result = await injector(claim.request, config, resolve(cwd), stateDir); const delta = await finalizeClaimedDispatchRequest({ claim, result, teamName, teamDirPath, config, cwd, stateDir, logsDir, issueCooldownMs, triggerCooldownMs, }); processed += delta.processed; skipped += delta.skipped; failed += delta.failed; } finally { claim.released = true; await releaseDispatchRequestLease(claim.lease); } } } finally { for (const claim of claims) { if (claim.released) continue; await releaseDispatchRequestLease(claim.lease); } } } return { processed, skipped, failed }; }