src/runner/review-lifecycle.ts

Rama main · réplica de lectura del repositorio original.

import { readFile } from 'node:fs/promises';
import { join } from 'node:path';
import { z } from 'zod';
import type { Octokit } from '@octokit/rest';
import { reviewPayloadSchema } from '../tasks/review/result.ts';
import { ReviewStale } from '../scm/errors.ts';
import { Trace } from '../harness/trace.ts';
import { writeState, type RunState } from './state.ts';
import { deliverImmediately, enqueueReviewDelivery, OutboundPending } from './outbound.ts';
import { assertRunPhase, type RunPhase } from './phases.ts';
import type { ScmAdapter } from '../scm/types.ts';

const shaSchema = z.string().regex(/^[a-f0-9]{40}$/);
export const reviewArtifactSchema = z.object({ head_sha: shaSchema }).and(reviewPayloadSchema);
const publicationSchema = z.object({
  id: z.number().int().positive(), html_url: z.url(), commit_id: shaSchema, published_at: z.string().min(1),
  run_id: z.string(), kind: z.literal('review'), recovered: z.boolean(), reused: z.boolean(),
});
export async function readArtifact(dir: string, name: string): Promise<any | null> {
  try { return JSON.parse(await readFile(join(dir, name), 'utf8')); }
  catch (error) { if ((error as NodeJS.ErrnoException).code === 'ENOENT') return null; throw error; }
}

export async function reviewCheckpoint(dir: string): Promise<RunPhase> {
  const result = await readArtifact(dir, 'result.json');
  if (result?.status === 'review_completed') return 'review_completed';
  if (result?.status === 'review_stale') return 'review_stale';
  if (await readArtifact(dir, 'review-publication.json')) return 'review_publishing';
  if (await readArtifact(dir, 'review.json')) return 'review_ready';
  return 'review_running';
}
export async function needsReviewRecovery(state: RunState, dir: string) {
  // review-publication.json is written only after confirmed remote evidence. Once it is
  // durable, the publication can never be "lost" again — even if a later /stop overwrote
  // result.json with a stopped terminal state. Such a run is closed; continuation belongs
  // to the retained workspace and an explicit matching /review, not to recovery hijack.
  const settled = await readArtifact(dir, 'review-publication.json');
  if (settled) return state.active || state.waiting_for_ci;
  const point = await reviewCheckpoint(dir);
  return point === 'review_ready' || point === 'review_publishing'
    || (point === 'review_completed' && (state.active || state.phase !== point || state.waiting_for_ci));
}

export interface ReviewLifecycle {
  trace: Trace; state: RunState; path: string; runtimeHome: string; recovered: boolean;
  // Set when this execution resumes a workspace whose prior run already settled the review publication.
  settled_run_id?: string;
  connection: () => Promise<{ adapter?: ScmAdapter; projectId?: string; botLogin: string; mentions: string[]; client?: Octokit }>;
}
async function checkpoint(ctx: ReviewLifecycle, phase: RunPhase) {
  ctx.state.phase = phase; ctx.state.waiting_for_ci = phase === 'ci';
  await writeState(ctx.path, ctx.state);
  ctx.trace.emit('phase', { phase, recovered: ctx.recovered });
}

export async function finalizeReview(ctx: ReviewLifecycle, result: Record<string, unknown>) {
  // Result first: death before the state write is recoverable without any remote action.
  ctx.trace.save('result.json', result);
  ctx.state.active = false; // pid is retained only as historical worker identity.
  await checkpoint(ctx, assertRunPhase(String(result.status)));
  return result;
}

export async function ensureReviewPublished(ctx: ReviewLifecycle) {
  const { trace, state } = ctx;
  const review = reviewArtifactSchema.parse(await readArtifact(trace.dir, 'review.json'));
  const saved = await readArtifact(trace.dir, 'review-publication.json');
  if (saved) {
    const publication = publicationSchema.parse(saved);
    // A resumed execution may carry the settled publication of the run it resumes;
    // identity is that run's id, and the commit match still pins it to this review.
    if ((publication.run_id !== state.run_id && publication.run_id !== ctx.settled_run_id) || publication.commit_id !== review.head_sha) {
      throw new Error('Publication artifact does not match run/review identity');
    }
    return { review, publication }; // Confirmed remote evidence: finalization only, zero GitHub actions.
  }
  await checkpoint(ctx, 'review_ready');
  const connection = await ctx.connection();
  if (!connection.adapter && !connection.client) throw new Error('Review publication has no SCM connection');
  const { adapter, projectId, client, botLogin, mentions } = connection;
  await checkpoint(ctx, 'review_publishing');
  const stored = await enqueueReviewDelivery({ root: ctx.runtimeHome, repo: state.repo, prNumber: state.pr_number,
    semanticKey: `review:${state.run_id}:${review.head_sha}`, headSha: review.head_sha, review, mentions, botLogin,
    runId: state.run_id, allowLegacy: ctx.recovered, source: { run_id: state.run_id, head_sha: review.head_sha,
      ...(adapter && projectId ? { project_id: projectId } : {}), recovered: ctx.recovered ? 1 : 0 } });
  const attempt = await deliverImmediately(ctx.runtimeHome, stored, { adapter, client, botLogin });
  if (attempt.item.status !== 'delivered') throw new OutboundPending(attempt.item);
  const published = attempt.item.receipt!;
  const publication = publicationSchema.parse({ ...published, run_id: state.run_id, kind: 'review', recovered: ctx.recovered });
  trace.save('review-publication.json', publication);
  trace.emit('review_published', publication);
  return { review, publication };
}

export async function completeReview(ctx: ReviewLifecycle) {
  const existing = await readArtifact(ctx.trace.dir, 'result.json');
  if (existing?.status === 'review_completed') {
    if (existing.run_id !== ctx.state.run_id || existing.repo !== ctx.state.repo || existing.pr_number !== ctx.state.pr_number) {
      throw new Error('Completed result does not match recovery target');
    }
    ctx.state.current_head_sha = existing.final_head_sha;
    ctx.state.active = false;
    await checkpoint(ctx, 'review_completed');
    return { ...existing, status: 'already_completed' };
  }
  try {
    const { review, publication } = await ensureReviewPublished(ctx);
    ctx.state.current_head_sha = review.head_sha;
    return await finalizeReview(ctx, { status: 'review_completed', run_id: ctx.state.run_id,
      repo: ctx.state.repo, pr_number: ctx.state.pr_number, final_head_sha: review.head_sha,
      recovered: ctx.recovered, review, publication });
  } catch (error) {
    if (error instanceof ReviewStale) {
      const evidence = { expected_head: error.expectedHead, actual_head: error.actualHead, pr_state: error.prState };
      ctx.trace.save('review-stale.json', evidence);
      return await finalizeReview(ctx, { status: 'review_stale', run_id: ctx.state.run_id, repo: ctx.state.repo,
        pr_number: ctx.state.pr_number, final_head_sha: error.expectedHead, recovered: ctx.recovered, ...evidence });
    }
    // Leave the durable model output available for retry, not a terminal Agent failure.
    ctx.state.active = false;
    await checkpoint(ctx, await reviewCheckpoint(ctx.trace.dir));
    ctx.trace.save('publication-error.json', { status: 'publication_interrupted',
      http_status: (error as { status?: number }).status ?? null, error_name: (error as Error).name });
    throw error;
  }
}