import { watchStop, TaskStopped } from './stop.ts';
import type { WorkspaceState } from '../workspace/manager.ts';
import type { PRTask, ParsedIntent } from './command.ts';
import { randomUUID } from 'node:crypto';
import { join } from 'node:path';
import { createGitHub } from '../github/client.ts';
import { capturePullRequest, type InspectedPR } from '../github/pull-request.ts';
import { resolveTaskDriver, type TaskProgress } from './tasks/driver.ts';
import { decideWriteback, performWriteback } from './writeback.ts';
import { applyRunPhase, assertRunPhase, createRunState } from './phases.ts';
import { workspaceChangesSince } from '../workspace/manager.ts';
import { ensureRepo, fetchPRState, disposeWorkspacePath } from '../workspace/repo-store.ts';
import { git, gitAuth } from '../workspace/git.ts';
import { Trace } from '../harness/trace.ts';
import { providerError } from '../harness/retry.ts';
import { seedContext } from '../harness/context/seed.ts';
import { budget } from '../harness/budget.ts';
import { claimRun, statePath, writeState, readState, workerStatus, type RunState } from './state.ts';
import { readArtifact, reviewCheckpoint } from './review-lifecycle.ts';
import { patchpawPaths } from '../config/paths.ts';
import { recoverLockedRun, reviewConnection } from './recovery.ts';
import { humanFeedback, hasHumanReplies } from './human-feedback.ts';
import { parsePRIntent } from './command.ts';
import { runClose } from './close.ts';
import { HumanHelpRequested } from '../tasks/human-help.ts';
import { type HumanReply } from './human-reply.ts';
import { type ConflictWorkspaceEvidence } from './workspace-evidence.ts';
import { retainWorkspace, readPaused, savePaused } from './resume.ts';
import { enqueueAndDeliverComment, OutboundPending } from './outbound.ts';
import { hasRunnableWork } from './runnable.ts';
import { bootstrapControlPlane, getRepositoryByName, getRepositoryByStorageKey, openControlPlaneDb } from '../control-plane/index.ts';
import type { ControlPlaneDb } from '../control-plane/db.ts';
import type { RuntimeExecution } from '../harness/runtime.ts';
import { classifyRunFailure, terminalStatusForFailure } from './failures.ts';
import {
  markCurrentConflictProposalStale,
  proposalPointerForState, readCurrentConflictProposal,
} from './conflict-proposals.ts';
import {
  readUnfinishedConflictApproval, updateConflictApproval, type ConflictApprovalRecord,
} from './conflict-approval.ts';
import { isSameRepoWriteback } from './command-approval.ts';
import { readCurrentApprovalPlan, readUnfinishedApprovalPlanClaim, type ApprovalPlan, type ApprovalPlanClaim } from './approval-plans.ts';
import { runLegacyConflictRepair } from './legacy-conflict-repair.ts';
import { recoverLegacyConflictApproval } from './legacy-conflict-recovery.ts';
import { prepareExecution, type ExecutionManifest } from './execution-preparation.ts';
import { runTaskDriver } from './task-execution.ts';
import { publishApprovalPlan as publishApprovalPlanDomain } from './approval-plan-publication.ts';
import { runConversationReply } from './conversation-runner.ts';
import { runReviewTask } from './review-runner.ts';
import { runConflictTask } from './conflict-runner.ts';
import { approvalPlanBinding, claimApprovalPlan } from './approval-plan-execution.ts';
import { prepareEntryLifecycle } from './entry-lifecycle.ts';
import { disposeTerminalWorkspace as disposeTerminalWorkspaceDomain, finishRun } from './run-finalizer.ts';
import { refuseCloseOnActiveTask } from './close-refusal.ts';
import { publishTaskWithLifecycle } from './task-publication.ts';
import { settleApprovalPlanClaim } from './approval-plan-settlement.ts';
import { runApprovalPlanTask } from './approval-plan-task.ts';
import { GitHubAdapter } from '../scm/github/adapter.ts';
import type { ChangeRequestSnapshot, ScmAdapter } from '../scm/types.ts';
import { saveScmSnapshot } from '../scm/snapshot.ts';
import { changeRequestRecipients } from '../scm/recipients.ts';
import type { OutboundConnection } from './outbound.ts';

function isApprovalComment(body: string) {
  return /^\s*@[^\s]+\s+\/(?:approval|approve)(?=$|\s)/i.test(body);
}

export type ScmRunContext = {
  adapter: ScmAdapter;
  projectId: string;
  snapshot: ChangeRequestSnapshot;
  snapshotPath?: string;
  connection?: () => Promise<OutboundConnection>;
};

export async function runPullRequest(config: { appId?: number; privateKey?: string; snapshotRoot: string; root: string; operatorLogin?: string; appSlug?: string; botLogin?: string; controlPlaneDb?: ControlPlaneDb; gitlabConnections?: Array<{ id: string; instanceUrl: string; projectIds: string[]; token?: string; botUserId?: string; botLogin?: string }> }, repo: string, number: number, scmContext?: ScmRunContext) {
  if (repo.startsWith('gitlab:') && !scmContext) {
    const { runGitLabMergeRequest } = await import('../scm/gitlab/runner.ts');
    return runGitLabMergeRequest(config, repo, number);
  }
  const path = statePath(patchpawPaths(config.root).state, repo, number);
  if (workerStatus(await readState(path)) === 'running') return { status: 'already_running' };
  if (!await hasRunnableWork(config.root, repo, number)) return { status: 'mention_required' };
  const release = await claimRun(path);
  if (!release) return { status: 'already_running' };
  const github = scmContext ? undefined : (() => {
    if (config.appId === undefined || !config.privateKey) throw new Error('GitHub App configuration is unavailable');
    return createGitHub({ appId: config.appId, privateKey: config.privateKey });
  })();
  let controlPlane: ControlPlaneDb | undefined = config.controlPlaneDb;
  let ownsControlPlane = false;
  let botLogin = scmContext?.adapter.botLogin ?? config.botLogin ?? (config.appSlug ? `${config.appSlug}[bot]` : '');
  let appSlug = config.appSlug ?? config.botLogin?.replace(/\[bot\]$/, '');
  let recoveryTarget: string | undefined;
  let priorConflictProposal: RunState['conflict_proposal'];
  let approvalRecovery: ConflictApprovalRecord | undefined;
  // A durable generic ApprovalPlan claim is an ordered PR-local job just like a legacy Conflict
  // Approval. It must wake the runner without a new human comment and must never consume a later
  // comment while it is being recovered.
  let genericApprovalRecovery: Awaited<ReturnType<typeof readUnfinishedApprovalPlanClaim>> = null;
  let priorClose: Pick<RunState, 'closed_through_comment_id' | 'close_start_notice_id' | 'close_comment_id' | 'close_mentions' | 'completion_notice_id' | 'completion_notice_status' | 'pending_close_refusal'> | undefined;
  let closeResult: Awaited<ReturnType<typeof runClose>> | undefined;
  const projectId = scmContext?.projectId ?? repo;
  let entryConnection: (() => Promise<OutboundConnection>) | undefined;
  try {
    entryConnection = scmContext?.connection ?? (async (): Promise<OutboundConnection> => {
      if (!github) throw new Error('GitHub connection is unavailable');
      const [owner, name] = repo.split('/');
      const { data: installation } = await github.app.rest.apps.getRepoInstallation({ owner, repo: name });
      return { client: github.installation(installation.id), botLogin };
    });
    const entry = await prepareEntryLifecycle({ config, resolveConnection: entryConnection,
      repo, prNumber: number, path, appSlug, botLogin, projectId });
    appSlug = entry.appSlug;
    botLogin = entry.botLogin;
    closeResult = entry.closeResult;
    recoveryTarget = entry.recoveryTarget;
    priorClose = entry.priorClose;
    priorConflictProposal = entry.priorConflictProposal;

  } catch (error) {
    await release(); throw error;
  }
  if (closeResult) {
    try { return closeResult; }
    finally { await release(); }
  }
  if (recoveryTarget) {
    const recoveryConnection = scmContext
      ? reviewConnection(config, repo, number, { adapter: scmContext.adapter, projectId: scmContext.projectId })
      : undefined;
    try { return await recoverLockedRun(config, repo, number, recoveryTarget, recoveryConnection); }
    finally { await release(); }
  }
  try {
    // Load durable approval work only after entry lifecycle has retired compatibility residue
    // from historical /close tombstones. Otherwise an old claim could win over the fresh comment.
    approvalRecovery = await readUnfinishedConflictApproval(path) ?? undefined;
    genericApprovalRecovery = await readUnfinishedApprovalPlanClaim(path);
  } catch (error) {
    await release();
    throw error;
  }
  if (!await hasHumanReplies(path) && !approvalRecovery && !genericApprovalRecovery) {
    // Pure outbound retries belong to the independent scheduler. A direct
    // worker invocation must not allocate a new Agent run for them.
    try { return { status: 'mention_required' }; }
    finally { await release(); }
  }
  const runId = `${new Date().toISOString().replace(/[:.]/g, '-')}-${randomUUID().slice(0, 8)}`;
  const trace = new Trace(join(patchpawPaths(config.root).runs, runId));
  if (config.privateKey) trace.secret(config.privateKey);
  const priorState = await readState(path);
  const priorHandledCommentIds = priorState?.handled_comment_ids ?? [];
  const priorLastPatchpawCommit = priorState?.last_patchpaw_commit ?? null;
  const state: RunState = createRunState({ repo, pr_number: number, run_id: runId, current_head_sha: '', repair_attempts: 0,
    last_patchpaw_commit: null, active: true, pid: process.pid,
    // Carry the retired-comment high-water and close communication facts across generations.
    // `closed_at` belongs only to the tombstone for the generation that /close retired.
    ...(priorClose ?? {}), ...(priorConflictProposal ? { conflict_proposal: priorConflictProposal } : {}) }, 'inspect');
  let inspected: InspectedPR | undefined;
  let snapshotPath: string | undefined = scmContext?.snapshotPath;
  let changeRequest: ChangeRequestSnapshot | undefined = scmContext?.snapshot;
  let scm: ScmAdapter | undefined = scmContext?.adapter;
  let confirmedRemoteHead: string | undefined;
  let executionId = 1;
  let workspaceNotice = '';
  let activeWorkspace: WorkspaceState | undefined;
  let conflictEvidence: { evidence: ConflictWorkspaceEvidence; evidenceSha256: string } | undefined;
  let approvalRequested = false;
  let approvedConflictRepair = false;
  let approvalRecord: ConflictApprovalRecord | undefined = approvalRecovery;
  let genericApprovalClaim: ApprovalPlanClaim | undefined;
  // Assigned inside the run body; the catch path needs it to settle a crashed approved write.
  let settleGenericApprovalClaim: (phase: 'completed' | 'interrupted') => Promise<void> = async () => undefined;
  let runtimeExecution: RuntimeExecution | undefined;
  let manifest: ExecutionManifest;
  // Cleanup ownership of a fresh worktree begins the moment its creation is attempted: a
  // half-created or prepare-failed worktree must never leak outside the run's lifecycle.
  let createdWorkspace: string | undefined;
  let activeTask: PRTask = 'conversation';
  let stop: ReturnType<typeof watchStop> | undefined;
  let feedback: Awaited<ReturnType<typeof humanFeedback>>['context'];
  const requireStop = () => {
    if (!stop) throw new Error('Run stop controller is unavailable');
    return stop;
  };
  const requireControlPlane = () => {
    if (!controlPlane) throw new Error('Control-plane database is unavailable');
    return controlPlane;
  };
  const requireConflictEvidence = () => {
    if (!conflictEvidence) throw new Error('Conflict workspace evidence is unavailable');
    return conflictEvidence;
  };
  const recipients = () => {
    const newCommentIds = new Set(feedback?.new_comment_ids ?? []);
    return changeRequestRecipients(changeRequest, config.operatorLogin,
      (feedback?.comments ?? []).filter(comment => newCommentIds.has(comment.comment_id)).map(comment => comment.author));
  };
  const deliverComment = (purpose: string, semanticKey: string, body: string, mentions: string[], source: Record<string, string | number | null | undefined> = {}) => {
    if (!scm || !botLogin) throw new Error('SCM adapter and bot identity are required before publishing a task report');
    return enqueueAndDeliverComment({ root: config.root, repo, prNumber: number, purpose, semanticKey, body, mentions,
      botLogin, source: { project_id: projectId, ...source } }, { adapter: scm, botLogin });
  };
  const phase = async (raw: string) => { const value = assertRunPhase(raw); applyRunPhase(state, value);
    await writeState(path, state); trace.emit('phase', { phase: value }); console.log(JSON.stringify({ run_id: runId, phase: value })); };
  const finish = (status: string, extra: { reason?: string; message?: string; [key: string]: unknown } = {}, persistedPhase = status) => finishRun({
    root: config.root, repo, prNumber: number, projectId, runId, executionId, statePath: path, state, trace, scm, resolveConnection: entryConnection, botLogin, activeTask, approvedConflictRepair,
    activeWorkspace, changeRequest, confirmedRemoteHead, stop, workspaceNotice, recipients,
    genericApprovalClaim: genericApprovalClaim ? approvalPlanBinding(genericApprovalClaim) : undefined, phase, settleGenericApprovalClaim,
  }, status, extra, persistedPhase);
  const disposeTerminalWorkspace = () => disposeTerminalWorkspaceDomain({
    root: config.root, repo, path, trace, activeWorkspace, createdWorkspace,
  });

  try {
    // A claimed Approval is an ordered PR-local job. Do not consume a later human comment while
    // recovering it; the later comment remains in the inbox for the next generation.
    const recoveryEntry = approvalRecovery || genericApprovalRecovery;
    const replies = recoveryEntry
      ? { handledIds: (await readState(path))?.handled_comment_ids ?? [], context: undefined }
      : await humanFeedback(path, patchpawPaths(config.root).runs, true);
    feedback = replies.context; state.handled_comment_ids = replies.handledIds;
    if (!feedback && !recoveryEntry) return { status: 'mention_required' };
    // Production configuration supplies the app slug. Keeping identity in local config lets
    // failure notices be prepared durably before any GitHub installation/app lookup; identity
    // discovery remains a delivery-time fallback for old callers without the configured slug.
    if (!appSlug) {
      const configured = config.botLogin?.replace(/\[bot\]$/, '');
      if (configured) appSlug = configured;
    }
    if (!appSlug && scm) appSlug = botLogin.replace(/\[bot\]$/i, '');
    if (!appSlug) throw new Error('SCM Bot identity is unavailable');
    botLogin = botLogin || `${appSlug}[bot]`;
    // Do not persist the approval comment's high-water mark during the initial inspect phase.
    // The approval record must win the crash race: once that record is durable, prepare below
    // retires the source comment atomically with the claimed state; before then it remains wakeable.
    const newCommentIds = new Set(feedback?.new_comment_ids ?? []);
    const pendingApprovalComment = feedback?.comments.find(comment => newCommentIds.has(comment.comment_id) && isApprovalComment(comment.body));
    if (pendingApprovalComment) state.handled_comment_ids = priorHandledCommentIds.filter((id: number) => id !== pendingApprovalComment.comment_id);
    const triggeringComment = feedback?.comments.find(comment => newCommentIds.has(comment.comment_id));
    let task: PRTask = 'conversation';
    let intent: ParsedIntent = { kind: 'conversation', repositoryId: '' };
    activeTask = task;
    // An /close arriving mid-task is refused mechanically by the owning worker (no model, no
    // queued destruction). Durable retirement comes FIRST: the comment must never become an
    // executable /close after this task ends, even when the refusal notice cannot be published.
    // A failed notice waits in the tiny state outbox (pending_close_refusal) and the next
    // PatchPaw entry retries it deterministically.
    const refuseClose = (comment: HumanReply) => refuseCloseOnActiveTask({
    root: config.root, repo, prNumber: number, runId, statePath: path, state, trace, projectId, botLogin, connection: async () => {
      if (scm) return { adapter: scm, botLogin };
      if (!github) throw new Error('GitHub connection is unavailable');
      const [owner, name] = repo.split('/');
      const { data: installation } = await github.app.rest.apps.getRepoInstallation({ owner, repo: name });
      return { client: github.installation(installation.id), botLogin };
    }, comment,
    });
    stop = watchStop(path, botLogin, () => state.handled_comment_ids ?? [], refuseClose);
    if (feedback) trace.save('human-feedback.json', feedback);
    await phase('inspect');
    // Approval needs to explain a closed change request without turning the inspection failure
    // into a generic harness error. GitHub keeps its capture artifact; GitLab enters with the
    // already-resolved normalized snapshot and adapter.
    const pr = github ? await capturePullRequest(github, repo, number, config.snapshotRoot, { allowClosed: true }) : undefined;
    if (pr) {
      inspected = pr;
      snapshotPath = pr.snapshotPath;
    }
    if (!scm) {
      if (!pr) throw new Error('SCM inspection did not produce a GitHub capture');
      scm = new GitHubAdapter(pr.client, botLogin);
    }
    changeRequest = await scm.readChangeRequest(projectId, number, { allowClosed: true });
    snapshotPath = (await saveScmSnapshot(config.snapshotRoot, changeRequest, `run-${runId}`)).path;
    state.current_head_sha = changeRequest.source.sha;
    confirmedRemoteHead = changeRequest.source.sha;
    const adapter = scm;
    const currentChangeRequest = changeRequest;
    // Bootstrap only fills genuinely missing control-plane records. Existing repository
    // edits remain authoritative, so every new invocation resolves the current config.
    controlPlane = config.controlPlaneDb ?? await openControlPlaneDb(config.root);
    ownsControlPlane = !config.controlPlaneDb;
    await bootstrapControlPlane({ root: config.root, repositories: scmContext ? [{ fullName: changeRequest.repository.pathWithNamespace,
      displayName: changeRequest.repository.pathWithNamespace, scmKind: changeRequest.kind, connectionId: changeRequest.repository.connectionId,
      remoteProjectId: changeRequest.repository.remoteProjectId, pathWithNamespace: changeRequest.repository.pathWithNamespace,
      webUrl: changeRequest.repository.webUrl, cloneUrl: changeRequest.repository.cloneUrl, storageKey: repo }] : [{ fullName: repo }], controlPlaneDb: controlPlane });
    const repository = scmContext ? await getRepositoryByStorageKey(controlPlane, repo) : await getRepositoryByName(controlPlane, repo);
    if (!repository) throw new Error(`Control-plane repository is missing after bootstrap: ${repo}`);
    const preparedControlPlane = requireControlPlane();
    intent = triggeringComment
      ? await parsePRIntent(controlPlane, repository.id, triggeringComment.body, appSlug)
      : { kind: 'control', control: 'approval' };
    let currentApprovalPlan = await readCurrentApprovalPlan(path);
    const requireApprovalPlan = () => {
      if (!currentApprovalPlan) throw new Error('Approval Plan is unavailable');
      return currentApprovalPlan;
    };
    task = intent.kind === 'command' ? intent.executionType : intent.kind === 'control' && intent.control === 'approval' && currentApprovalPlan
      ? currentApprovalPlan.plan.execution_type : intent.kind === 'control' && intent.control === 'stop' ? 'stop'
      : intent.kind === 'control' && intent.control === 'close' ? 'close'
      : intent.kind === 'control' && intent.control === 'approval' ? 'conflict' : 'conversation';
    activeTask = task;
    approvalRequested = intent.kind === 'control' && intent.control === 'approval';
    if (!['open', 'opened'].includes(changeRequest.state) && !approvalRequested) throw new Error('Change request is not open or has no source repository');
    if (intent.kind === 'control' && intent.control === 'close') {
      return await finish('needs_human', { reason: '/close 是机械生命周期命令，但本条评论未能在前置生命周期入口处理；没有启动模型或修复。请重发一次单独的 /close。' });
    }
    const token = async () => {
      const installationGitToken = scm?.installationGitToken;
      if (!installationGitToken) throw new Error('SCM installation Git token is unavailable');
      const value = await installationGitToken.call(scm);
      trace.secret(value); trace.secret(Buffer.from(`x-access-token:${value}`).toString('base64'));
      return value;
    };
    await requireStop().guard();
    const retained = await readPaused(path);
    const currentConflictProposal = await readCurrentConflictProposal(path);
    currentApprovalPlan = currentApprovalPlan ?? await readCurrentApprovalPlan(path);
    const retainedForDiscussion = retained && ['budget_exhausted', 'needs_human', 'stopped', 'publication_pending', 'awaiting_approval', 'claimed', 'repairing'].includes(retained.status);
    const discussingConflict = task === 'conversation' && !!retainedForDiscussion && !!currentConflictProposal
      && retained?.workspace.path === currentConflictProposal.pointer.workspace_path
      && ['draft', 'publication_pending', 'published'].includes(currentConflictProposal.proposal.status);
    const discussingPause = (task === 'conversation' && !!retainedForDiscussion && !currentConflictProposal)
      || discussingConflict;
    if (task === 'stop' && currentConflictProposal) {
      // /stop is mechanical even while a proposal is awaiting publication/approval. Preserve the
      // proposal pointer and retained checkout, but never turn a stop comment into a model turn.
      state.conflict_proposal = proposalPointerForState(currentConflictProposal.proposal, currentConflictProposal.pointer.workspace_path);
      return await finish('stopped', { reason: 'Conflict Proposal 已保留；/stop 只停止当前本地会话，不会调用模型或启动修复。' });
    }
    if (task === 'stop' && !discussingPause) return await finish('stopped', { reason: '当前没有正在执行的任务或可继续的暂停工作区。没有启动新任务。' });
    if (retained && ['budget_exhausted', 'needs_human', 'stopped'].includes(retained.status)
      && !discussingPause && task !== retained.task) {
      return await finish('needs_human', { reason: `当前有暂停的 /${retained.task} 工作区。请先沟通并用 /${retained.task} 继续该任务；此次没有另建工作区。` });
    }
    await phase('workspace');
    // Every run refreshes the one shared object store for this GitHub repo BEFORE the
    // resume/fresh decision; fetching never mutates an existing paused worktree's HEAD/index.
    await ensureRepo(config.root, repo, changeRequest.repository.cloneUrl, trace);
    const fetchPlan = scm.workspaceFetchPlan(changeRequest);
    const fetchedBase = await fetchPRState(config.root, repo, { headSha: changeRequest.source.sha, baseRef: changeRequest.target.ref,
      headFetch: fetchPlan.head }, trace, gitAuth(await token(), changeRequest.repository.cloneUrl, 'x-access-token', scm.connection.instanceUrl));
    const currentBase = { ref: fetchedBase.baseRef, sha: fetchedBase.currentBaseTipSha };
    const legacyRecovery = await recoverLegacyConflictApproval({
      root: config.root, statePath: path, repo, prNumber: number, runId, executionId, botLogin, trace, state, changeRequest, scm, projectId, currentBase,
      retained: retained ?? undefined, currentConflictProposal: currentConflictProposal ?? undefined, approvalRequested, approvalRecovery,
      controlPlane, recipients, finish, confirmedRemoteHead,
      setApprovalRecord: value => { approvalRecord = value; },
      setConfirmedRemoteHead: value => { confirmedRemoteHead = value; },
    });
    approvalRecord = legacyRecovery.approvalRecord;
    confirmedRemoteHead = legacyRecovery.confirmedRemoteHead;
    if (legacyRecovery.handled) return legacyRecovery.result;

    if (discussingConflict && currentConflictProposal) {
      const basis = currentConflictProposal.proposal.basis;
      const staleReason = basis.pr_head_sha !== changeRequest.source.sha ? 'pr_head_changed'
        : basis.current_base_tip_sha !== currentBase.sha ? 'current_base_tip_changed'
        : basis.base_ref !== currentBase.ref ? 'base_ref_changed' : undefined;
      if (staleReason) {
        if (!retained) throw new Error('Conflict discussion is missing its retained workspace');
        await markCurrentConflictProposalStale(path, retained.workspace.path, staleReason);
        await savePaused(path, { ...retained, status: 'stale', reason: staleReason });
        state.conflict_proposal = { ...proposalPointerForState(currentConflictProposal.proposal, retained.workspace.path), status: 'stale' };
        await writeState(path, applyRunPhase({ ...state, active: false }, 'stale'));
        try { await disposeWorkspacePath(config.root, repo, retained.workspace.path, trace); }
        catch (error) { trace.emit('stale_workspace_dispose_failed', { workspace: retained.workspace.path, reason: staleReason, message: (error as Error).message }); }
        return await finish('stale', { reason: `当前 Git 事实已变化（${staleReason}），旧 Conflict Proposal 已标记 stale；不会基于旧工作区继续讨论或发布修订。请重新发送 /conflict。` });
      }
    }
    const preparedExecution = await prepareExecution({
      root: config.root, repo, prNumber: number, runId, trace, state, changeRequest: currentChangeRequest, scm: adapter, projectId,
      snapshotPath, path, task, intent, repository, currentApprovalPlan,
      approvalRequested, approvedConflictRepair, approvalRecord, approvalRecovery, triggeringComment, feedback,
      currentBase, retained, currentConflictProposal, discussingPause, discussingConflict, botLogin, stop: requireStop(), workspaceNotice, executionId, phase, finish,
      controlPlane: preparedControlPlane,
      setCreatedWorkspace: value => { createdWorkspace = value; },
      setActiveWorkspace: value => { activeWorkspace = value; },
      setExecutionId: value => { executionId = value; },
    });
    if (preparedExecution.handled) return preparedExecution.result;
    approvedConflictRepair = preparedExecution.approvedConflictRepair;
    approvalRecord = preparedExecution.approvalRecord;
    executionId = preparedExecution.executionId;
    runtimeExecution = preparedExecution.runtimeExecution;
    const preparedRuntimeExecution = runtimeExecution;
    if (!preparedRuntimeExecution) throw new Error('Runtime execution is unavailable after preparation');
    manifest = preparedExecution.manifest;
    activeWorkspace = preparedExecution.ws;
    workspaceNotice = preparedExecution.workspaceNotice;
    const { genericApprovalWrite, resumed, ws, snapshot, snapshotReference, conflictEvidence: preparedConflictEvidence } = preparedExecution;
    conflictEvidence = preparedConflictEvidence;

    const permissionPhase = genericApprovalWrite ? 'approved_write' as const : snapshot.command?.permission === 'read_write_approval' ? 'planning' as const : 'normal' as const;
    const taskOptions = { task, prompt: '', ws, trace, runId, currentBase, execution: preparedRuntimeExecution, permissionPhase,
      approvalMessage: genericApprovalWrite ? '用户批准了你的计划' : undefined,
      preventGitPush: snapshot.command?.permission !== 'read_only',
      prMemory: { root: patchpawPaths(config.root).memory, repo, number },
      repairStartHead: resumed ? (resumed.remote_head ?? ws.initialHead) : undefined,
      stopSignal: stop.signal, stopRequest: stop.request,
      onCloseout: () => phase('repair_closeout') };
    // Harness-owned writeback applies to every execution type that has write capability.
    // read_write_approval only becomes writable in the approved_write phase.
    // Whether this execution may write back is the writeback domain's decision, not a boolean
    // assembled here. It runs before the Agent receives write tools, so a fork is refused before
    // any capability exists rather than after something has already been changed.
    const writeback = decideWriteback({ permission: snapshot.command?.permission, approvedWrite: genericApprovalWrite,
      sameRepo: isSameRepoWriteback(currentChangeRequest, repo) });
    if (writeback.kind === 'refused') {
      return await finish('needs_human', { reason: writeback.reason, approval_rejection_code: writeback.code });
    }
    const seed = async () => ({ ...await seedContext(currentChangeRequest, ws, trace, currentBase), human_feedback: feedback,
      execution_id: executionId, resumed: !!resumed, retained_for_discussion: !!discussingPause, workspace_notice: workspaceNotice, previous_stop_report: resumed ? await readArtifact(trace.dir, 'resume-stop-report.json') : null,
      previous_closeout: resumed ? await readArtifact(trace.dir, 'resume-closeout.json') : null,
      conflict_workspace_evidence: conflictEvidence?.evidence ?? null,
      pending_conflict_proposal: discussingConflict || approvedConflictRepair ? currentConflictProposal?.proposal ?? null : null,
      conflict_approval: approvedConflictRepair ? approvalRecord ?? null : null,
      approved_conflict_proposal: approvedConflictRepair ? currentConflictProposal?.proposal ?? null : null });
    const beginTask = async (task: string) => { manifest.task_chain.push(task); trace.save('manifest.json', manifest);
      await requireStop().guard();
      await phase(task === 'review' ? 'review_running' : task); };
    const assertCurrent = async () => {
      const current = await adapter.readChangeRequest(projectId, number, { allowClosed: true });
      if (!['open', 'opened'].includes(current.state) || current.source.sha !== state.current_head_sha
          || current.source.ref !== currentChangeRequest.source.ref
          || current.source.pathWithNamespace.toLowerCase() !== currentChangeRequest.source.pathWithNamespace.toLowerCase()
          || current.target.ref !== currentChangeRequest.target.ref || current.target.sha !== ws.mainSha) {
        throw new Error('Change request head, target, source, or state changed during the run');
      }
    };
    // The durable generic approval claim only exists while an approved write is in flight. It is
    // settled once the write has a durable outcome, so a crash-recovered worker never replays it.
    settleGenericApprovalClaim = phase => settleApprovalPlanClaim({ path, trace, claim: genericApprovalClaim, phase });
    const deliver = (status: 'custom_completed' | 'conflict_completed' | 'ci_completed' | 'repair_completed', body: string) => publishTaskWithLifecycle({
      root: config.root, repo, prNumber: number, runId, path, workspacePath: ws.path, status, body, workspaceNotice,
      headSha: state.current_head_sha, projectId, mentions: recipients(), botLogin, adapter, trace,
      approvalPlan: genericApprovalClaim ? approvalPlanBinding(genericApprovalClaim) : undefined, guard: () => requireStop().guard(), finish,
    });
    const publishApprovalPlan = async (body: string) => publishApprovalPlanDomain({
      root: config.root, repo, prNumber: number, runId, executionId, path, task: task as ApprovalPlan['execution_type'], body,
      snapshot, snapshotSha256: snapshotReference.snapshot_sha256, currentApprovalPlan,
      setCurrentApprovalPlan: value => { currentApprovalPlan = value; },
      conflictEvidence: requireConflictEvidence(), currentBase, source: currentChangeRequest, adapter, projectId, workspace: ws, state, trace,
      mentions: recipients(), botLogin, finish,
    });
    if (genericApprovalWrite) {
      const execution = await claimApprovalPlan({
        root: config.root, repo, prNumber: number, runId, executionId, path, trace, state, source: currentChangeRequest, adapter, projectId, currentBase, workspace: ws,
        snapshot, snapshotSha256: snapshotReference.snapshot_sha256, currentApprovalPlan: requireApprovalPlan(),
        recovery: genericApprovalRecovery, claim: genericApprovalClaim, setClaim: value => { genericApprovalClaim = value; },
        triggeringComment, botLogin, priorLastPatchpawCommit, finish,
      });
      genericApprovalClaim = execution.claim;
      if (execution.handled) return execution.result;
    }
    // A Conflict run whose merge is already satisfied is a mechanical no-op under every
    // Permission, including the read-only planning phase of read_write_approval.
    if (task === 'conflict' && !approvedConflictRepair && !ws.mergePending
        && (!resumed || (resumed.local_head === currentChangeRequest.source.sha && !(await git(ws.path, ['status', '--porcelain'])).stdout.trim()))) {
      return await deliver('conflict_completed', '当前 PR 已包含目标分支，无需修复合并冲突；没有修改或提交代码。');
    }
    const writebackPermit = writeback.kind === 'allowed' ? writeback.permit : undefined;
    const writebackEnabled = writebackPermit !== undefined;
    // Harness-owned writeback. The sequence lives in the writeback domain; this closure only
    // records its outcome in run state, which is the runner's business.
    const publish = async (kind: string) => {
      if (!writebackPermit) throw new Error('Writeback was requested without an enabled permission capability');
      const outcome = await performWriteback({ workspace: ws, trace, scm: adapter, projectId,
        changeRequestNumber: number, expectedBaseSha: currentBase.sha, sameRepo: isSameRepoWriteback(currentChangeRequest, repo), kind,
        permit: writebackPermit, previousHead: state.current_head_sha, baseRef: currentChangeRequest.target.ref, headRef: currentChangeRequest.source.ref,
        headRepo: currentChangeRequest.source.pathWithNamespace, remoteUrl: currentChangeRequest.repository.cloneUrl,
        credentialScopeUrl: adapter.connection.instanceUrl, token, guard: () => requireStop().guard(), beforePush: async (candidateSha) => {
          state.current_head_sha = candidateSha;
          state.last_patchpaw_commit = candidateSha;
          await phase('publishing');
        } });
      state.current_head_sha = outcome.confirmedRemoteHead; state.last_patchpaw_commit = outcome.commitSha;
      confirmedRemoteHead = outcome.confirmedRemoteHead;
    };
    const hasWorkspaceChanges = async () => (await workspaceChangesSince(ws, state.current_head_sha, trace)).has_changes;
    // A task whose execution this runner cannot express generically runs through its driver. The
    // driver decides what the task needs from freshly-read real state; this runner performs every
    // durable effect it asks for — the Agent turn, writeback, publication, the terminal result.
    // The runner therefore never branches on which task this is: the loop below is identical for
    // any driver, and the only task-shaped facts it handles are the ones the driver declares.
    const taskDriver = resolveTaskDriver(snapshot.template_type);
    if (taskDriver) {
      const progress: TaskProgress = { turns: [] };
      const execution = await runTaskDriver({
        driver: taskDriver, phase: permissionPhase, writable: writebackEnabled, repo, scm: adapter,
        projectId, changeRequestNumber: number, trace,
        signal: stop.signal, options: taskOptions, repairAttempts: () => state.repair_attempts,
        repairBudget: budget.repairAttempts, resumed: resumed ? { pausePhase: resumed.pause_phase } : null, progress,
        guardCycle: async (phaseLabel: string) => { await phase(phaseLabel); await assertCurrent(); await requireStop().guard(); },
        beginTask, seed, finish, deliver, hasWorkspaceChanges, publish,
        headSha: () => state.current_head_sha,
        saveRepairAttempts: async () => { state.repair_attempts++; await writeState(path, state); },
        publishApprovalPlan,
      });
      if (execution.handled) return execution.result;
    }
    if (!approvalRequested && snapshot.command?.permission === 'read_write_approval') {
      const planning = await runApprovalPlanTask({ task, taskOptions, seed, beginTask,
        publishApprovalPlan, finish });
      if (planning.handled) return planning.result;
      throw new Error(`No planning path for execution type ${snapshot.template_type}; it needs a task driver.`);
    }
    if (task === 'conversation' || task === 'stop') {
      return await runConversationReply({
        repo, prNumber: number, runId, trace, scm: adapter, projectId,
        taskOptions, seed, beginTask, deliverComment, recipients, finish,
      });
    }
    if (task === 'review') {
      return await runReviewTask({
        root: config.root, repo, prNumber: number, trace, state, path, resumed: resumed ? { run_id: resumed.run_id } : null,
        currentHead: () => state.current_head_sha, taskOptions, seed, workspaceNotice,
        writable: writebackEnabled, hasWorkspaceChanges, publish, stopGuard: () => requireStop().guard(),
        savePausedCompleted: async () => { const paused = await readPaused(path); if (paused?.workspace.path === ws.path) await savePaused(path, { ...paused, status: 'completed' }); },
        beginTask, connection: async () => ({ adapter, projectId, botLogin, mentions: recipients() }), mentions: recipients, finish, settleApproval: settleGenericApprovalClaim,
      });
    }
    if (task === 'conflict') {
      if (approvedConflictRepair) {
        if (!approvalRecord || !currentConflictProposal) throw new Error('Approved Conflict repair is missing its durable approval or proposal.');
        await beginTask('conflict-repair');
        const outcome = await runLegacyConflictRepair({
        root: config.root, repo, prNumber: number, runId, executionId, path, trace, changeRequest: currentChangeRequest, scm: adapter, projectId, currentBase,
          retained: retained ?? (() => { throw new Error('Approved Conflict repair is missing its retained workspace.'); })(), workspace: ws, state, approvalRecord, proposal: currentConflictProposal.proposal, confirmedRemoteHead,
          taskOptions, seed, phase, finish, token, guard: () => requireStop().guard(), botLogin, recipients,
          setApprovalRecord: value => { approvalRecord = value; },
          setConfirmedRemoteHead: sha => { confirmedRemoteHead = sha; },
        });
        approvalRecord = outcome.approvalRecord;
        confirmedRemoteHead = outcome.confirmedRemoteHead;
        return outcome.result;
      }
      if (!ws.mergePending && (!resumed || (resumed.local_head === currentChangeRequest.source.sha && !(await git(ws.path, ['status', '--porcelain'])).stdout.trim()))) return await deliver('conflict_completed', '当前 change request 已包含目标分支，无需修复合并冲突；没有修改或提交代码。');
      return await runConflictTask({
        repo, prNumber: number, scm: adapter, projectId,
        baseSha: currentChangeRequest.diffBaseSha ?? currentChangeRequest.target.sha, baseRef: currentChangeRequest.target.ref, taskOptions, seed, trace, workspace: ws, path, runId, beginTask,
        retainWorkspace: async input => retainWorkspace(path, trace, input),
        executionId, writebackEnabled, hasWorkspaceChanges, publish, deliver, finish,
        currentHead: () => state.current_head_sha,
      });
    }
  } catch (error) {
    if (error instanceof TaskStopped || stop?.signal.aborted) {
      if (approvalRecord?.status === 'accepted' && approvalRecord.phase !== 'completed') {
        approvalRecord = await updateConflictApproval(path, approvalRecord.approval_id, { phase: 'interrupted' }) ?? approvalRecord;
        trace.emit('conflict_approval_interrupted', { approval_id: approvalRecord.approval_id, reason: 'human_stop' });
      }
      await settleGenericApprovalClaim('interrupted');
      if (activeWorkspace && changeRequest && ['conflict', 'ci', 'repair', 'review'].includes(activeTask)) {
        await retainWorkspace(path, trace, { run_id: runId, execution_id: executionId, task: activeTask as 'conflict' | 'ci' | 'repair' | 'review',
          base_sha: changeRequest.diffBaseSha ?? changeRequest.target.sha, base_ref: changeRequest.target.ref, workspace: activeWorkspace, pause_reason: 'human_stop',
          remote_head: confirmedRemoteHead, pause_phase: state.phase });
      }
      return await finish('stopped', { reason: error instanceof TaskStopped ? error.message : new TaskStopped().message });
    }
    if (approvalRecord?.status === 'accepted' && approvalRecord.phase === 'repairing') {
      approvalRecord = await updateConflictApproval(path, approvalRecord.approval_id, { phase: 'interrupted' }) ?? approvalRecord;
      trace.emit('conflict_approval_interrupted', { approval_id: approvalRecord.approval_id, reason: 'harness_or_provider_failure' });
    }
    if (error instanceof HumanHelpRequested) {
      await settleGenericApprovalClaim('interrupted');
      return await finish('needs_human', { reason: error.message });
    }
    trace.emit('run_error', { phase: state.phase, ...providerError(error), message: (error as Error).message });
    if (await readArtifact(trace.dir, 'review.json')) {
      if (error instanceof OutboundPending) trace.save('result.json', { status: 'review_publication_pending', run_id: runId,
        repo, pr_number: number, final_head_sha: state.current_head_sha, delivery_id: error.delivery.delivery_id });
      state.active = false; await phase(await reviewCheckpoint(trace.dir));
      return { status: 'publication_interrupted', run_id: runId };
    }
    await settleGenericApprovalClaim('interrupted');
    const failure = classifyRunFailure(error);
    return await finish(terminalStatusForFailure(failure), { failed_phase: state.phase,
      error: providerError(error), message: failure.message, failure });
  } finally {
    try { await stop?.close(); }
    finally { try { await disposeTerminalWorkspace(); } finally { try { if (ownsControlPlane) controlPlane?.close(); } finally { await release(); } } }
  }
}