import { readFile, readdir } from 'node:fs/promises'; import { join } from 'node:path'; import { z } from 'zod'; import { closeCompleteBodyFor, closeStartBodyFor } from './close.ts'; import { closeoutMarker } from './closeout-publication.ts'; import { enqueueCommentDelivery, enqueueReviewDelivery } from './outbound.ts'; import { patchpawPaths } from '../config/paths.ts'; import { readState, statePath } from './state.ts'; import { reviewPayloadSchema } from '../tasks/review/result.ts'; import { runNoticeBody, type RunNotice } from './run-notice.ts'; import { recipientLogins } from '../scm/recipients.ts'; export interface LegacyReconciliationConfig { root: string; operatorLogin?: string; } type JsonObject = Record; function isJsonObject(value: unknown): value is JsonObject { return typeof value === 'object' && value !== null && !Array.isArray(value); } async function json(path: string): Promise { try { const value: unknown = JSON.parse(await readFile(path, 'utf8')); if (!isJsonObject(value)) throw new Error(`Expected JSON object at ${path}`); return value; } catch (error) { if ((error as NodeJS.ErrnoException).code === 'ENOENT') return null; throw error; } } function stringValue(value: unknown) { return typeof value === 'string' ? value : undefined; } function numberValue(value: unknown) { return typeof value === 'number' && Number.isSafeInteger(value) ? value : undefined; } function strings(value: unknown) { return Array.isArray(value) ? value.filter((item): item is string => typeof item === 'string') : []; } function runDir(root: string, runId: string) { return join(patchpawPaths(root).runs, runId); } const runNoticeSchema = z.object({ run_id: z.string(), head: z.string(), status: z.string(), phase: z.string(), reason: z.string(), mentions: z.array(z.string()), bot_login: z.string().optional(), platform: z.enum(['github', 'gitlab']).optional(), failure: z.object({ code: z.string(), category: z.enum(['provider', 'model', 'scm', 'workspace', 'internal']), retryable: z.boolean(), user_action: z.enum(['retry', 'check_configuration', 'human_review', 'inspect_logs']), message: z.string(), upstream_status: z.number().optional(), upstream_code: z.union([z.string(), z.number()]).optional(), attempts: z.number().optional(), scm_platform: z.enum(['github', 'gitlab']).optional(), }).optional(), }).passthrough(); function parseRunNotice(value: JsonObject): RunNotice { return runNoticeSchema.parse(value); } async function legacyStatePaths(root: string) { const stateRoot = patchpawPaths(root).state; let repoKeys: string[]; try { repoKeys = await readdir(stateRoot); } catch (error) { if ((error as NodeJS.ErrnoException).code === 'ENOENT') return []; throw error; } const paths: string[] = []; for (const repoKey of repoKeys) { let names: string[]; try { names = await readdir(join(stateRoot, repoKey)); } catch { continue; } for (const name of names.filter(value => /^pr-\d+\.json$/.test(value))) paths.push(join(stateRoot, repoKey, name)); } return paths; } /** One migration pass for pre-outbox state artifacts. State JSON is authoritative for repo/PR. */ export async function reconcileLegacyOutbox(config: LegacyReconciliationConfig, botLogin?: string) { const runsRoot = patchpawPaths(config.root).runs; let runs: string[]; try { runs = await readdir(runsRoot); } catch (error) { if ((error as NodeJS.ErrnoException).code === 'ENOENT') runs = []; else throw error; } for (const runId of runs) { const dir = runDir(config.root, runId); const manifest = await json(join(dir, 'manifest.json')); const repo = stringValue(manifest?.repo); const prNumber = numberValue(manifest?.pr_number); if (!repo || prNumber === undefined) continue; const state = await readState(statePath(patchpawPaths(config.root).state, repo, prNumber)); const review = await json(join(dir, 'review.json')); const publication = await json(join(dir, 'review-publication.json')); if (review && !publication && ['review_ready', 'review_publishing'].includes(state?.phase ?? 'review_ready')) { const headSha = stringValue(review.head_sha); if (!headSha) throw new Error(`Legacy review artifact is missing head_sha: ${dir}`); const { head_sha: _headSha, ...reviewPayload } = review; const mentions = recipientLogins([stringValue(manifest?.request_author), config.operatorLogin]); await enqueueReviewDelivery({ root: config.root, repo, prNumber, semanticKey: `review:${runId}:${headSha}`, headSha, review: reviewPayloadSchema.parse(reviewPayload), mentions, botLogin, runId, allowLegacy: true, source: { run_id: runId, head_sha: headSha, recovered: 1 } }); } const noticeRaw = await json(join(dir, 'run-notice.json')); const notification = await json(join(dir, 'notification.json')); if (noticeRaw && notification?.status !== 'published') { const notice = parseRunNotice(noticeRaw); await enqueueCommentDelivery({ root: config.root, repo, prNumber, purpose: 'run_notice', semanticKey: `run-notice:${notice.run_id}`, body: runNoticeBody(notice), mentions: recipientLogins(notice.mentions), botLogin: notice.bot_login ?? botLogin, legacyMarkers: ['budget_exhausted', 'stopped'].includes(notice.status) ? [closeoutMarker(notice.run_id)] : undefined, source: { run_id: notice.run_id, status: notice.status } }); } } for (const path of await legacyStatePaths(config.root)) { const state = await readState(path); if (!state?.repo || !state.pr_number) continue; const mentions = recipientLogins(state.close_mentions ?? []); if (state.completion_notice_status === 'pending' && state.closed_through_comment_id) { await enqueueCommentDelivery({ root: config.root, repo: state.repo, prNumber: state.pr_number, purpose: 'close_completion', semanticKey: `close-completion:${state.close_comment_id ?? state.closed_through_comment_id}`, body: closeCompleteBodyFor(state.repo), mentions, botLogin, source: { close_comment_id: state.close_comment_id ?? state.closed_through_comment_id } }); } if (state.pending_close_refusal) { await enqueueCommentDelivery({ root: config.root, repo: state.repo, prNumber: state.pr_number, purpose: 'close_refusal', semanticKey: `close-refusal:${state.pending_close_refusal.comment_id}`, body: '当前任务仍在运行,请先 /stop,再执行 /close。', mentions: recipientLogins([state.pending_close_refusal.author]), botLogin, source: { comment_id: state.pending_close_refusal.comment_id } }); } const journal = await json(`${path}.close.json`); if (journal?.status === 'closing' && !journal.start_notice_id) { const closeCommentId = typeof journal.close_comment_id === 'string' || typeof journal.close_comment_id === 'number' ? journal.close_comment_id : null; await enqueueCommentDelivery({ root: config.root, repo: state.repo, prNumber: state.pr_number, purpose: 'close_start', semanticKey: `close-start:${String(closeCommentId ?? '')}`, body: closeStartBodyFor(state.repo), mentions: recipientLogins(strings(journal.mentions)), botLogin, source: { close_comment_id: closeCommentId } }); } } }