src/runner/communication-scheduler.ts
Rama main · réplica de lectura del repositorio original.
import { createGitHub, type GitHubReader } from '../github/client.ts';
import { claimRun, readState, statePath, workerStatus } from './state.ts';
import { retryAfterMs } from '../harness/retry.ts';
import { drainDueDeliveries, OUTBOUND_LOCK_BUSY_RETRY_MS, safeError, type OutboundConnection, type OutboundItem, type StoredItem } from './outbound.ts';
import { verifyInboundNow } from './inbound-verification.ts';
import { createCommunicationWakeServer, registerCommunicationWake, type CommunicationWakeServer } from './communication-wake.ts';
import { closeCommunicationStore, openCommunicationStore, type CommunicationStore } from './communication-store.ts';
import type { InboundRecord } from './communication-types.ts';
import { patchpawPaths } from '../config/paths.ts';
import { withRuntimeLock } from '../migration/runtime-lock.ts';
import { reconcileLegacyOutbox } from './communication-legacy-reconciliation.ts';
import {
classifyFinalization, finalizeApprovalPlanPublication, finalizeReviewDelivery, finalizeRunNotice,
finalizeStaleReview, type FinalizationConnection,
} from './communication-finalization.ts';
import { finalizeLegacyConflictProposal, finalizeLegacyConflictRepair } from './communication-legacy-conflict-finalization.ts';
import { finalizeCloseCompletion, finalizeCloseRefusal, finalizeCloseStart } from './communication-close-finalization.ts';
import type { ScmInboundReader } from '../scm/types.ts';
export { reconcileLegacyOutbox } from './communication-legacy-reconciliation.ts';
export interface SchedulerConfig {
root: string;
legacyHome?: string;
appId: number;
privateKey: string;
operatorLogin?: string;
snapshotRoot: string;
appSlug?: string;
botLogin?: string;
wakeTransport?: 'unix' | 'pipe' | 'memory';
wakeSocketPath?: string;
/** Test/embedding seam; production leaves runtime wake failures fatal. */
onWakeFailure?: (error: Error) => void;
schedulerHooks?: {
afterDeadlineRead?: (deadline: string | undefined) => Promise<void> | void;
};
/** Platform-specific inbound verification, used by GitLab records while GitHub keeps its legacy reader. */
inboundScmReader?: ScmInboundReader | ((connectionId: string) => ScmInboundReader | undefined);
/** Platform-aware outbox connection resolver. GitHub remains the default resolver. */
connectionFor?: (item: OutboundItem) => Promise<OutboundConnection>;
}
export interface CommunicationSchedulerService {
ready: Promise<void>;
stop(): void;
}
export type OutboundSchedulerService = CommunicationSchedulerService;
class FinalizationBusy extends Error {
retryAfter = OUTBOUND_LOCK_BUSY_RETRY_MS / 1000;
constructor() { super('PR lifecycle is owned by an active worker'); this.name = 'FinalizationBusy'; }
}
async function finalizeDelayedDeliveryOwned(config: SchedulerConfig, stored: StoredItem, store: CommunicationStore,
connection?: FinalizationConnection) {
const item = stored.item;
const done = async () => (await store.markFinalized(item.delivery_id))?.item;
const resolveConnection = async () => typeof connection === 'function' ? connection() : connection;
const finalizationKind = classifyFinalization(item);
if (finalizationKind === 'stale_review') return finalizeStaleReview({ config, stored, store, resolveConnection, done });
if (finalizationKind === 'review') return finalizeReviewDelivery({ config, stored, store, resolveConnection, done });
if (finalizationKind === 'run_notice') return finalizeRunNotice({ config, stored, store, resolveConnection, done });
if (finalizationKind === 'approval_plan') return finalizeApprovalPlanPublication({ config, stored, store, resolveConnection, done });
if (finalizationKind === 'legacy_conflict_proposal') return finalizeLegacyConflictProposal({ config, stored, store, resolveConnection, done });
if (finalizationKind === 'legacy_conflict_repair') return finalizeLegacyConflictRepair({ config, stored, store, resolveConnection, done });
if (finalizationKind === 'close_refusal') return finalizeCloseRefusal({ config, stored, store, resolveConnection, done });
if (finalizationKind === 'close_completion') return finalizeCloseCompletion({ config, stored, store, resolveConnection, done });
if (finalizationKind === 'close_start') return finalizeCloseStart({ config, stored, store, resolveConnection, done });
return done();
}
export async function finalizeDelayedDelivery(config: SchedulerConfig, stored: StoredItem, connection?: FinalizationConnection) {
const store = await openCommunicationStore(config.root, config.legacyHome);
const path = statePath(patchpawPaths(config.root).state, stored.item.repo, stored.item.pr_number);
if (workerStatus(await readState(path)) === 'running') { await closeCommunicationStore(store); throw new FinalizationBusy(); }
const release = await claimRun(path);
if (!release) { await closeCommunicationStore(store); throw new FinalizationBusy(); }
try {
const result = await finalizeDelayedDeliveryOwned(config, stored, store, connection);
if (!result) throw new FinalizationBusy();
return result;
} finally { await release(); await closeCommunicationStore(store); }
}
function configuredBotLogin(config: SchedulerConfig) { return config.botLogin ?? (config.appSlug ? `${config.appSlug}[bot]` : undefined); }
type GitHubRuntime = ReturnType<typeof createGitHub> & GitHubReader;
export function startCommunicationScheduler(config: SchedulerConfig, githubOverride?: GitHubRuntime | GitHubReader,
onVerified: (reply: InboundRecord['reply']) => Promise<void> = async () => {}): CommunicationSchedulerService {
const github = (githubOverride ?? (config.appId && config.privateKey ? createGitHub(config) : undefined)) as GitHubRuntime | undefined;
let store: CommunicationStore | undefined;
let stopped = false;
let running = false;
let dirty = false;
let reconciled = false;
let timer: NodeJS.Timeout | undefined;
let wakeReady = config.wakeTransport === 'memory';
let wakeListening = false;
let wakeServer: CommunicationWakeServer | undefined;
let unregisterWake: (() => void) | undefined;
let activePump: Promise<void> | undefined;
let botLogin = configuredBotLogin(config);
let wakeFailure: Error | undefined;
const appLogin = async () => {
if (botLogin) return botLogin;
if (!github) throw new Error('GitHub App bot identity is unavailable');
const { data } = await github.app.rest.apps.getAuthenticated();
if (!data?.slug) throw new Error('GitHub App has no bot identity');
botLogin = `${data.slug}[bot]`;
return botLogin;
};
const connectionFor = async (item: OutboundItem): Promise<OutboundConnection> => {
if (config.connectionFor) return config.connectionFor(item);
if (!github) throw new Error('No SCM connection is configured for this delivery');
const [owner, repo] = item.repo.split('/');
const { data: installation } = await github.app.rest.apps.getRepoInstallation({ owner, repo });
return { client: github.installation(installation.id), botLogin: await appLogin() };
};
const processInbound = async () => {
if (!store || (!github?.readPullRequest && !config.inboundScmReader)) return;
for (const stored of await store.listDueInbound(new Date().toISOString(), 25)) {
await verifyInboundNow(config.root, stored, github, onVerified, config.inboundScmReader);
}
};
const processFinalizations = async () => {
if (!store) return;
for (const stored of await store.listDueFinalizations(new Date().toISOString(), 25)) {
const needsConnection = (stored.item.kind === 'review' && stored.item.status === 'cancelled_stale') || stored.item.purpose === 'close_start';
try {
await finalizeDelayedDelivery(config, stored, needsConnection ? () => connectionFor(stored.item) : undefined);
} catch (error) {
const fallback = Math.min(60 * 60_000, 15_000 * 2 ** Math.min(stored.item.finalization_attempt_count, 7));
const delay = error instanceof FinalizationBusy ? OUTBOUND_LOCK_BUSY_RETRY_MS : retryAfterMs(error) ?? fallback;
await store.deferFinalization(stored.item.delivery_id, safeError(error), new Date(Date.now() + delay).toISOString());
}
}
};
const armDeadline = async () => {
if (stopped || !store) return;
const deadline = await store.nextCommunicationDeadline();
await config.schedulerHooks?.afterDeadlineRead?.(deadline);
if (dirty || stopped) {
if (timer) clearTimeout(timer);
timer = undefined;
return;
}
if (timer) clearTimeout(timer);
timer = undefined;
if (!deadline) return;
timer = setTimeout(() => { timer = undefined; requestPump(); }, Math.max(0, Date.parse(deadline) - Date.now()));
timer.unref();
};
const pump = async () => {
if (stopped || running || !wakeReady || !store) return;
running = true;
try {
await withRuntimeLock(config.root, 'shared', false, async () => {
do {
dirty = false;
if (!reconciled) {
if (!(await store!.getMeta('legacy_reconciliation_v1'))) {
await reconcileLegacyOutbox(config, botLogin);
await store!.setMeta('legacy_reconciliation_v1', new Date().toISOString());
}
reconciled = true;
}
await store!.recoverExpiredSending(new Date().toISOString());
await processInbound();
await drainDueDeliveries(config.root, item => connectionFor(item), 25);
await processFinalizations();
} while (dirty || await store!.hasDueCommunication(new Date().toISOString()));
await armDeadline();
});
} catch (error) {
console.error(JSON.stringify({ status: 'communication_pump_failed', error_name: (error as Error).name,
error_code: (error as { code?: string }).code ?? null,
http_status: (error as { status?: number }).status ?? null }));
try { await armDeadline(); } catch { /* durable wake or restart can retry */ }
} finally {
running = false;
if (dirty && !stopped) requestPump();
}
};
function requestPump() {
dirty = true;
if (!running && wakeReady && !stopped) {
const current = pump();
activePump = current;
void current.finally(() => { if (activePump === current) activePump = undefined; });
}
}
const failWakeRuntime = (raw: unknown) => {
if (stopped || wakeFailure) return;
const error = raw instanceof Error ? raw : new Error(String(raw));
wakeFailure = error;
wakeReady = false;
wakeListening = false;
unregisterWake?.();
unregisterWake = undefined;
if (timer) clearTimeout(timer);
timer = undefined;
console.error(JSON.stringify({ status: 'communication_wake_runtime_failed', error_name: error.name,
error_code: (error as NodeJS.ErrnoException).code ?? null }));
if (config.onWakeFailure) {
config.onWakeFailure(error);
return;
}
process.nextTick(() => { throw error; });
};
const initialize = (async () => {
store = await openCommunicationStore(config.root, config.legacyHome);
if (config.wakeTransport !== 'memory') {
wakeServer = createCommunicationWakeServer(config.root, requestPump, config.wakeTransport, config.wakeSocketPath);
wakeServer.server.on('error', error => {
if (wakeListening) failWakeRuntime(error);
});
wakeServer.server.on('close', () => {
if (wakeListening && !stopped) failWakeRuntime(new Error('Communication wake socket closed unexpectedly'));
});
await wakeServer.listen();
wakeListening = true;
}
wakeReady = true;
unregisterWake = registerCommunicationWake(config.root, requestPump);
const initialPump = pump();
activePump = initialPump;
try { await initialPump; }
finally { if (activePump === initialPump) activePump = undefined; }
})();
void initialize.catch(async error => {
wakeReady = false;
console.error(JSON.stringify({ status: 'communication_wake_startup_failed', error_name: (error as Error).name,
code: (error as NodeJS.ErrnoException).code ?? null }));
await wakeServer?.close().catch(() => {});
if (store) await closeCommunicationStore(store).catch(() => {});
});
return {
ready: initialize,
stop() {
stopped = true; wakeReady = false; unregisterWake?.(); unregisterWake = undefined;
if (timer) clearTimeout(timer); timer = undefined;
void wakeServer?.close().catch(() => {});
void (async () => {
try { await activePump; } catch { /* the pump already logged its safe failure facts */ }
await store?.close();
})();
},
};
}
export function startOutboundScheduler(config: SchedulerConfig, legacyIntervalOrGithub?: number | GitHubRuntime,
githubOverride?: GitHubRuntime): OutboundSchedulerService {
const github = typeof legacyIntervalOrGithub === 'number' ? githubOverride : legacyIntervalOrGithub;
return startCommunicationScheduler(config, github);
}