/**
 * V10 rectification turn runner over the durable V9 Case/Evidence domains.
 *
 * Tool activity is published as it happens so the browser can render stages.
 * The user-visible reply is the model's terminal `text-delta`, streamed live
 * once that step is the spoken answer. Provider thinking stays on
 * `reasoning-delta` and is not a public stream event. Server narration is
 * only the empty-stream fallback after tools. A retried attempt emits
 * `attempt.reset` first so the client discards the abandoned attempt.
 * Durable receipts, billing, and settled history still come only from the
 * successful attempt.
 */
import type { Agent } from "@mastra/core/agent";
import { RectificationAgentAction, resolveRectificationStepBudget } from "@/mastra/agentic-rectification";
import {
  createV10RunAttempt,
  finalizeV10RunAttempt,
  insertV9RunPhase,
  insertV9SkillRunReceipt,
  loadV9CaseDossier,
  loadV9CaseSkillIdentity,
  loadV9CaseCompute,
  RectificationToolServiceError,
  resolveV10ConversationFocus,
  setV9EvidenceDateReliability,
  type RectificationRpcClient,
  type V9CaseDossier,
} from "./tool-service";
import { RECTIFICATION_SKILL_NAME, RECTIFICATION_SKILL_VERSION } from "./case-status";
import { RECTIFICATION_AGENT_TOOLS } from "./public-receipt";
import { agentGenerationSettings, cachedSystemMessage, promptCacheUsage } from "../../agent-generation-settings.ts";
import { toAgentModelFinishReason } from "../../agent-observability.ts";
import { classifyDateReliabilityUtterance, isDateReliabilitySchema } from "./date-reliability.ts";
import { decideFromDossier } from "./decision-from-dossier";
import { ensureNonTerminalTurnExit, persistExhaustionGateTurn, persistNextInterviewIfIdle } from "./answer-choice";
import { alreadyDelivered } from "./delivery-turn-guard";
import { parseAgentChoiceCopy, isPersistedFocusId } from "./choice-card";
import {
  RECTIFICATION_USER_COPY,
  OPENING_COLLECT_DOMAINS,
  isAcceptableOpeningBody,
  openingRangeFromCandidateRange,
  openingSpokenBody,
  withCompareFailedRetryNotice,
  withRangeChangedAfterEvidence,
} from "../user-copy";
import { stripQuestionSentences, stripVerbalWindowChange, trimSpokenTurnForInterview } from "./collect-prompt";
import { focusSpokenPrompt } from "./turn-question";
import { previousInferenceFromReceipt } from "../core/compose-receipt.ts";
import {
  resolveExactSkillPackage,
  type ResolvedSkillPackageIdentity,
} from "../../skill-package-registry.ts";
import {
  activityChangedFromTool,
  mapStreamChunkToActivity,
  mapStreamChunkToPhase,
  streamToolNames,
  isPublicRectificationToolName,
  type PublicStreamEvent,
} from "./stream-mapping";
import { mapModelFinishToErrorCode, userFacingRunFailure } from "./run-diagnostic";
import {
  batchResultFromToolChunk,
  composeHostFallbackNarration,
  lastCompletedPublicTool,
  publicWriteToolCompleted,
  answerClaimsEvidenceRecorded,
  retryConstraintForAttempt,
  turnExpectsEvidenceWrite,
} from "./host-fallback";
import {
  applyStepAnswerChunk,
  createStepAnswerState,
  flushStepAnswerOnStreamFinish,
} from "./step-answer";
import {
  defaultMessageOrigin,
  isRectificationMessageOrigin,
  messageContentHash,
  type RectificationMessageOrigin,
} from "./message-origin";

export type V9RunBilling = Readonly<{
  reserve(): Promise<{ success: boolean; reason?: string; status: number }>;
  complete(input: {
    inputTokens: number;
    outputTokens: number;
    durationMs: number;
    cache?: ReturnType<typeof promptCacheUsage>;
  }): Promise<boolean>;
  release(): Promise<boolean>;
}>;

export type V9AgentRunOptions = Readonly<{
  userId: string;
  caseId: string;
  sessionId: string;
  requestId: string;
  action: RectificationAgentAction;
  message: string | null;
  messageOrigin?: RectificationMessageOrigin;
  clientActionId?: string | null;
  modelName: string;
  skillName?: string;
  skillVersion?: string;
  accounting: RectificationRpcClient;
  buildAgent(
    turnId: string,
    skillPackage: ResolvedSkillPackageIdentity,
    attemptId: string,
  ): Promise<Agent>;
  billing: V9RunBilling;
  emit(event: PublicStreamEvent): Promise<void> | void;
  signal?: AbortSignal;
  timeContext?: string;
  generationModel?: unknown;
  attemptTimeoutMs?: number;
  expectedWrite?: "evidence" | "none" | "unknown";
  collectIntent?: "classified" | "unclassified" | null;
}>;

export const RECTIFICATION_AGENT_ATTEMPT_TIMEOUT_MS = 210_000;
export const RECTIFICATION_AGENT_ROUTE_MAX_DURATION_S = 240;

export type V9AgentRunResult = Readonly<{
  ok: boolean;
  turnId: string;
  turnStatus: "completed" | "failed" | "retryable";
  skillLoaded: boolean;
  answerText: string;
  phases: readonly string[];
  toolsUsed: readonly string[];
  errorCode: string | null;
  previousFocusId: string | null;
}>;

type AttemptStatus = "completed" | "failed" | "retryable";
type Usage = Readonly<{ inputTokens: number; outputTokens: number; cache?: ReturnType<typeof promptCacheUsage> }>;
type AttemptOutcome = Readonly<{
  ok: boolean;
  status: AttemptStatus;
  errorCode: string | null;
  usage: Usage;
  answerText: string;
  answerDeltas: readonly string[];
  phases: readonly string[];
  toolsUsed: readonly string[];
  events: readonly PublicStreamEvent[];
  skillBound: boolean;
  caseLoaded: boolean;
  attemptId: string;
  settleBilling?: boolean;
}>;

const MAX_ATTEMPTS = 2;
const RETRYABLE_ERROR_CODES = new Set([
  "stream_aborted",
  "stream_unfinished",
  "skill_not_loaded",
  "skill_not_bound",
  "case_not_loaded",
]);

function streamFinishReason(chunk: {
  type: string;
  payload?: { stepResult?: { reason?: unknown }; reason?: unknown };
}): ReturnType<typeof toAgentModelFinishReason> | null {
  if (chunk.type !== "finish") return null;
  const raw = chunk.payload?.stepResult?.reason ?? chunk.payload?.reason;
  if (typeof raw !== "string" || !raw.trim()) return null;
  return toAgentModelFinishReason(raw);
}

function first(value: unknown): unknown {
  if (Array.isArray(value)) return value[0] ?? null;
  if (value && typeof value === "object" && "value" in value) {
    return (value as { value?: unknown }).value;
  }
  return value;
}

/**
 * Identity for a public tool-call chunk. Mastra may put the model input on
 * `args`, `input`, or omit it; missing input collapses to `{}` so a second
 * call of the same tool name still looks identical.
 */
function publicToolCallKey(toolName: string, payload: unknown): string {
  if (!payload || typeof payload !== "object") return `${toolName}:{}`;
  const record = payload as Record<string, unknown>;
  const args = record.args ?? record.input ?? record.toolArgs ?? {};
  try {
    return `${toolName}:${JSON.stringify(args)}`;
  } catch {
    return `${toolName}:{}`;
  }
}

async function rpcOf(
  accounting: RectificationRpcClient,
  fn: string,
  args: Record<string, unknown>,
): Promise<unknown> {
  const { data, error } = await accounting.rpc(fn, args);
  if (error) throw new RectificationToolServiceError(error.message);
  return first(data);
}

function safeErrorCode(error: unknown): string {
  const message = error instanceof Error ? error.message : String(error);
  for (const code of [
    "empty_stream",
    "evidence_not_written",
    "stream_aborted",
    "stream_unfinished",
    "skill_not_loaded",
    "skill_not_bound",
    "case_not_loaded",
    "repeated_tool_call",
    "focus_persistence_failed",
    "answer_truncated",
    "run_timeout",
    "max_steps",
    "provider_error",
  ]) {
    if (message.includes(code)) return code;
  }
  if (message.includes("agentic_rectification_case_terminal")) return "case_terminal";
  if (message.includes("agentic_rectification_case_not_found")) return "case_not_found";
  if (message.includes("agentic_rectification_case_session_mismatch")) return "case_session_mismatch";
  if (message.includes("Thinking mode does not support this tool_choice")) {
    return "thinking_tool_choice_unsupported";
  }
  return "run_failed";
}

function isRetryableError(errorCode: string): boolean {
  return RETRYABLE_ERROR_CODES.has(errorCode);
}

function shouldAutoRetry(
  errorCode: string,
  signal?: AbortSignal,
  status?: AttemptStatus,
): boolean {
  if (signal?.aborted) return false;
  if (errorCode === "empty_stream") return status === "retryable";
  if (errorCode === "evidence_not_written") return status === "retryable";
  return isRetryableError(errorCode);
}

function clockWindow(range: { start_time?: string | null; end_time?: string | null } | null | undefined): string | null {
  const start = range?.start_time?.trim().slice(0, 5) || "";
  const end = range?.end_time?.trim().slice(0, 5) || "";
  return start && end ? `${start}–${end}` : null;
}

export function buildOpeningBrief(dossier: V9CaseDossier, birthTimeClue?: string | null): string {
  const confirmed = dossier.evidence.filter((item) => item.status === "confirmed");
  const pending = dossier.evidence.filter((item) => item.status === "draft" || item.status === "pending_confirmation");
  const domains = [...new Set(confirmed.map((item) => item.domain))].slice(0, 6);
  const range = dossier.case.candidateRange;
  const window = clockWindow(range);
  const uncertaintyType = window
    ? `当前搜索窗口 ${window}，来自用户在资料里声明的不确定档`
    : "用户的出生时间精度仍需通过经历证据核对";
  const clue = typeof birthTimeClue === "string" && birthTimeClue.trim()
    ? birthTimeClue.trim()
    : "";
  return [
    "【服务端 opening brief】",
    `Case 状态：${dossier.case.status}。`,
    `当前搜索窗口：${window ?? "尚未锁定"}。来源：intake 声明的不确定档。`,
    `出生时间不确定类型：${uncertaintyType}。`,
    `已有证据摘要：已确认 ${confirmed.length} 条，待澄清或待确认 ${pending.length} 条${domains.length ? `；已覆盖 ${domains.join("、")}` : ""}。`,
    ...(clue
      ? [`家人或本人关于出生时段的线索（仅旁白建议，不得改搜索窗口）：${clue}`]
      : []),
    `做法要点：一句当前窗口与核对做法；一句「最后给区间和代表分钟，不给精确到秒」；一句「想到几件说几件，有大概年月就行」并点出${OPENING_COLLECT_DOMAINS.join("、")}。一条消息可以报多件，想到几件说几件。不得写具体年份，不得要求先准备材料。不要提问。先用 rectification-set-focus 的 spokenPrompt 写出当前采集题，题干写成「先说你最容易想起的一两件，年月大概就行」。`,
  ].join("\n");
}

export async function runV9AgentTurn(options: V9AgentRunOptions): Promise<V9AgentRunResult> {
  const {
    userId, caseId, sessionId, action, message, modelName,
    accounting, buildAgent, billing, emit, signal,
  } = options;
  const skillName = options.skillName ?? RECTIFICATION_SKILL_NAME;

  let dossier = await loadV9CaseDossier(accounting, userId, caseId);
  const reliabilityFocus = dossier.conversationSummary.activeFocus;
  const reliabilitySchema = reliabilityFocus?.expectedAnswerSchema ?? null;
  if (message && isDateReliabilitySchema(reliabilitySchema)) {
    const classified = classifyDateReliabilityUtterance(message);
    const evidenceId = String(reliabilitySchema.target_evidence_id ?? "");
    const focusId = reliabilityFocus?.id ?? "";
    try {
      if (classified && evidenceId) {
        await setV9EvidenceDateReliability(accounting, userId, caseId, evidenceId, classified);
        if (isPersistedFocusId(focusId)) {
          await resolveV10ConversationFocus(accounting, userId, caseId, {
            focusId,
            status: "resolved",
            evidenceId,
          });
        }
      } else if (isPersistedFocusId(focusId)) {
        await resolveV10ConversationFocus(accounting, userId, caseId, {
          focusId,
          status: "skipped",
        });
      }
      dossier = await loadV9CaseDossier(accounting, userId, caseId);
    } catch (error) {
      console.warn(
        `[rectification-v9] date reliability write deferred case=${caseId} reason=${
          error instanceof Error ? error.message : String(error)
        }`,
      );
    }
  }
  const previousFocusId = dossier.conversationSummary.activeFocus?.id ?? null;
  if (dossier.case.sessionId !== sessionId) {
    throw new RectificationToolServiceError("agentic_rectification_case_session_mismatch");
  }
  const boundIdentity = await loadV9CaseSkillIdentity(accounting, userId, caseId);
  if ((options.skillName && options.skillName !== boundIdentity.name)
    || (options.skillVersion && options.skillVersion !== boundIdentity.version)
    || dossier.case.skillName !== boundIdentity.name
    || dossier.case.skillVersion !== boundIdentity.version) {
    throw new RectificationToolServiceError("agentic_rectification_skill_identity_mismatch");
  }
  const resultId = dossier.latestResult?.resultId ?? "";
  if (alreadyDelivered({
    caseId,
    resultId,
    hasUserMessage: Boolean(message?.trim()),
  })) {
    return {
      ok: true,
      turnId: "",
      turnStatus: "completed",
      skillLoaded: true,
      answerText: "",
      phases: [],
      toolsUsed: [],
      errorCode: "already_delivered",
      previousFocusId,
    };
  }
  const skillPackage = resolveExactSkillPackage(
    boundIdentity.name,
    boundIdentity.version,
    boundIdentity.sha256,
  );
  if (skillPackage.sourceCommit !== boundIdentity.sourceCommit) {
    throw new RectificationToolServiceError("agentic_rectification_skill_identity_mismatch");
  }
  let birthTimeClue: string | null = null;
  if (action === "opening") {
    try {
      const compute = await loadV9CaseCompute(accounting, userId, caseId);
      const raw = compute.baselineBirthSnapshot.birth_time_clue;
      birthTimeClue = typeof raw === "string" && raw.trim() ? raw.trim() : null;
    } catch {
      birthTimeClue = null;
    }
  }

  const reserve = await billing.reserve();
  if (!reserve.success) {
    throw new RectificationToolServiceError(reserve.reason ?? "billing_denied");
  }

  let turnRow: unknown;
  try {
    turnRow = await rpcOf(accounting, "append_agentic_rectification_turn", {
      p_user_id: userId,
      p_case_id: caseId,
      p_user_message: message,
      p_assistant_message: null,
      p_model_name: modelName,
      p_model_version: null,
      p_status: "pending",
      p_request_id: options.requestId,
    });
  } catch (error) {
    await billing.release();
    throw error;
  }
  const turnRecord = turnRow && typeof turnRow === "object"
    ? turnRow as Record<string, unknown>
    : null;
  const turnId = typeof turnRecord?.turn_id === "string" ? turnRecord.turn_id : "";
  if (!turnId) {
    await billing.release();
    throw new RectificationToolServiceError("agentic_rectification_turn_incomplete");
  }

  try {
    await rpcOf(accounting, "record_agentic_rectification_turn_origin", {
      p_user_id: userId,
      p_case_id: caseId,
      p_turn_id: turnId,
      p_origin: isRectificationMessageOrigin(options.messageOrigin)
        ? options.messageOrigin
        : defaultMessageOrigin(options.action),
      p_client_action_id: options.clientActionId ?? options.requestId,
      p_content_hash: messageContentHash(message),
    });
  } catch {
    // Origin is audit metadata; a missing RPC must not fail the turn.
  }

  const shouldExecute = turnRecord?.should_execute === undefined
    ? true
    : turnRecord.should_execute === true;
  const existingStatus = typeof turnRecord?.status === "string" ? turnRecord.status : "pending";
  const existingAnswer = typeof turnRecord?.assistant_message === "string"
    ? turnRecord.assistant_message
    : "";
  if (!shouldExecute) {
    if (existingStatus === "completed" && existingAnswer.trim()) {
      await emit({ type: "run.started" });
      await emit({ type: "answer.delta", text: existingAnswer });
      await emit({ type: "run.completed", turnId });
      return {
        ok: true,
        turnId,
        turnStatus: "completed",
        skillLoaded: typeof turnRecord?.successful_attempt_id === "string",
        answerText: existingAnswer,
        phases: ["run.completed"],
        toolsUsed: [],
        errorCode: null,
        previousFocusId,
      };
    }
    if (existingStatus !== "pending") await billing.release();
    throw new RectificationToolServiceError(
      existingStatus === "pending"
        ? "agentic_rectification_turn_in_progress"
        : "agentic_rectification_turn_already_finalized",
    );
  }

  const startedAt = Date.now();
  await emit({ type: "run.started" });

  let lastAttemptError: string | null = null;
  let finalOutcome: AttemptOutcome | null = null;
  for (let attemptNumber = 1; attemptNumber <= MAX_ATTEMPTS; attemptNumber += 1) {
    const claim = await createV10RunAttempt(
      accounting,
      userId,
      caseId,
      turnId,
      attemptNumber,
    );
    const { attemptId } = claim;
    if (!claim.shouldExecute) {
      await billing.release();
      throw new RectificationToolServiceError(
        claim.alreadyInProgress
          ? "agentic_rectification_attempt_in_progress"
          : "agentic_rectification_attempt_already_finalized",
      );
    }
    let outcome: AttemptOutcome;
    try {
      outcome = await streamAttempt(attemptNumber, attemptId, lastAttemptError);
    } catch (error) {
      const errorCode = safeErrorCode(error);
      const reason = error instanceof Error ? error.message.slice(0, 180) : "UnknownError";
      console.error(
        `[rectification-v10] attempt failed case=${caseId} turn=${turnId} attempt=${attemptId} code=${errorCode} reason=${reason}`,
      );
      outcome = {
        ok: false,
        status: isRetryableError(errorCode) ? "retryable" : "failed",
        errorCode,
        usage: { inputTokens: 0, outputTokens: 0 },
        answerText: "",
        answerDeltas: [],
        phases: [],
        toolsUsed: [],
        events: [],
        skillBound: false,
        caseLoaded: false,
        attemptId,
      };
    }
    if (!outcome.ok) {
      await persistCommittedPhase("run.failed", null, attemptId, 1_000_000 + attemptNumber);
      await finalizeV10RunAttempt(
        accounting,
        userId,
        caseId,
        turnId,
        attemptId,
        outcome.status,
        outcome.errorCode,
        outcome.usage,
      );
    }
    finalOutcome = outcome;
    if (!outcome.ok) lastAttemptError = outcome.errorCode;
    if (outcome.ok
      || outcome.status === "failed"
      || !shouldAutoRetry(outcome.errorCode ?? "run_failed", signal, outcome.status)
      || attemptNumber === MAX_ATTEMPTS) break;
    await emit({ type: "attempt.reset" });
  }

  const outcome = finalOutcome ?? {
    ok: false,
    status: "failed" as const,
    errorCode: "run_failed",
    usage: { inputTokens: 0, outputTokens: 0 },
    answerText: "",
    answerDeltas: [],
    phases: [],
    toolsUsed: [],
    events: [],
    skillBound: false,
    caseLoaded: false,
    attemptId: "",
  };

  if (!outcome.ok) {
    await billing.release();
    await finalizeTurn(outcome.status, null, outcome.attemptId, null);
    await emit({
      type: "run.failed",
      code: outcome.errorCode ?? "run_failed",
      recoverable: outcome.status === "retryable",
      message: userFacingRunFailure(outcome.errorCode),
    });
    return {
      ok: false,
      turnId,
      turnStatus: outcome.status,
      skillLoaded: false,
      answerText: "",
      phases: outcome.phases,
      toolsUsed: outcome.toolsUsed,
      errorCode: outcome.errorCode,
      previousFocusId,
    };
  }

  const skipBilling = outcome.settleBilling === false;
  if (!skipBilling) {
    const durationMs = Date.now() - startedAt;
    const completed = await billing.complete({ ...outcome.usage, durationMs });
    if (!completed) {
      await persistCommittedPhase("run.failed", null, outcome.attemptId, outcome.phases.length + 1);
      await finalizeV10RunAttempt(
        accounting,
        userId,
        caseId,
        turnId,
        outcome.attemptId,
        "retryable",
        "usage_settlement_failed",
        outcome.usage,
      );
      await finalizeTurn("retryable", null, outcome.attemptId, null);
      await emit({
        type: "run.failed",
        code: "run_failed",
        recoverable: true,
        message: userFacingRunFailure("run_failed"),
      });
      return {
        ok: false,
        turnId,
        turnStatus: "retryable",
        skillLoaded: false,
        answerText: "",
        phases: [],
        toolsUsed: [],
        errorCode: "usage_settlement_failed",
        previousFocusId,
      };
    }

    await persistCommittedPhase(
      "billing.settled",
      null,
      outcome.attemptId,
      outcome.phases.length + 1,
      true,
    );
  } else {
    await billing.release();
  }

  await persistCommittedPhase(
    "run.completed",
    null,
    outcome.attemptId,
    outcome.phases.length + (skipBilling ? 1 : 2),
    true,
  );
  await finalizeV10RunAttempt(
    accounting,
    userId,
    caseId,
    turnId,
    outcome.attemptId,
    "completed",
    null,
    outcome.usage,
  );
  const answerText = outcome.answerText;
  let spokenAnswer = answerText;
  let interviewIdle: Awaited<ReturnType<typeof persistNextInterviewIfIdle>> | null = null;
  if (action === "opening" || action === "evidence") {
    try {
      interviewIdle = await persistNextInterviewIfIdle({ accounting, userId, caseId, askedTurnId: turnId });
      if (interviewIdle.terminalNote && interviewIdle.hostNarration && !turnId) {
        try {
          await persistExhaustionGateTurn({
            accounting,
            userId,
            caseId,
            askedTurnId: turnId,
            hostNarration: interviewIdle.hostNarration,
            resultId,
          });
        } catch (error) {
          console.warn(
            `[rectification-v9] persist exhaustion gate before collect attach failed case=${caseId} reason=${safeErrorCode(error)}`,
          );
        }
      }
    } catch (error) {
      console.warn(
        `[rectification-v9] persist interview before collect attach failed case=${caseId} reason=${safeErrorCode(error)}`,
      );
      try {
        interviewIdle = await ensureNonTerminalTurnExit({ accounting, userId, caseId });
      } catch (repairError) {
        console.warn(
          `[rectification-v9] nonterminal exit after idle failure failed case=${caseId} reason=${safeErrorCode(repairError)}`,
        );
      }
    }
  }
  if (action === "evidence") {
    spokenAnswer = trimSpokenTurnForInterview(answerText, interviewIdle?.terminalNote === true);
    if (spokenAnswer !== answerText) {
      await emit({ type: "answer.delta", text: spokenAnswer, replace: true });
    }
  }
  await finalizeTurn("completed", spokenAnswer, outcome.attemptId, outcome.attemptId, true);

  if (!skipBilling) await emit({ type: "billing.settled" });
  await emit({ type: "run.completed", turnId });

  return {
    ok: true,
    turnId,
    turnStatus: "completed",
    skillLoaded: outcome.skillBound,
    answerText: spokenAnswer,
    phases: skipBilling
      ? [...outcome.phases, "run.completed"]
      : [...outcome.phases, "billing.settled", "run.completed"],
    toolsUsed: outcome.toolsUsed,
    errorCode: null,
    previousFocusId,
  };

  async function streamAttempt(
    attemptNumber: number,
    attemptId: string,
    previousErrorCode: string | null,
  ): Promise<AttemptOutcome> {
    const agent = await buildAgent(turnId, skillPackage, attemptId);
    let frameworkSkill: unknown = null;
    try {
      frameworkSkill = await (agent as unknown as { getSkill(name: string): Promise<unknown> }).getSkill(skillName);
    } catch {
      frameworkSkill = null;
    }
    if (!frameworkSkill) {
      return failedAttempt(attemptId, "skill_not_loaded");
    }

    const rawSkillInstructions = (frameworkSkill as { instructions?: unknown }).instructions;
    const skillInstructions = typeof rawSkillInstructions === "string"
      ? rawSkillInstructions.trim()
      : "";
    if (!skillInstructions) {
      return failedAttempt(attemptId, "skill_not_loaded");
    }

    const messages = buildAgentMessages(
      options,
      attemptNumber,
      dossier,
      skillInstructions,
      birthTimeClue,
      lastAttemptError,
    );
    const maxSteps = resolveRectificationStepBudget(action);
    const abortController = new AbortController();
    const onAbort = () => abortController.abort();
    signal?.addEventListener("abort", onAbort, { once: true });
    let timedOut = false;
    const timeout = setTimeout(() => {
      timedOut = true;
      abortController.abort();
    }, options.attemptTimeoutMs ?? RECTIFICATION_AGENT_ATTEMPT_TIMEOUT_MS);

    let skillBound = true;
    let caseLoaded = false;
    let intentClassified = false;
    let streamFailed = false;
    let finished = false;
    let finishReason: ReturnType<typeof toAgentModelFinishReason> | null = null;
    let answerText = "";
    const answerDeltas: string[] = [];
    const phases: string[] = [];
    const toolsUsed = new Set<string>();
    const events: PublicStreamEvent[] = [];
    const toolTerminalStatus = new Map<string, "completed" | "failed">();
    let batchToolResult: unknown = null;
    let hostFallbackUsed = false;
    const emittedKeys = new Set<string>();
    const emittedActivities = new Set<string>();
    const repeatedCalls = new Map<string, number>();
    let phaseSequence = 0;
    const rangeBeforeCompare = previousInferenceFromReceipt(
      dossier.latestResult?.decisionReceipt ?? null,
    )?.credible_range ?? null;

    const recordPhase = async (phase: string, tool: string | null = null) => {
      if (
        phase === "answer.delta"
        || phase === "thinking.delta"
        || phase === "activity.changed"
        || phase === "choice.applied"
        || phase === "attempt.reset"
        || emittedKeys.has(`${phase}:${tool ?? ""}`)
      ) return;
      emittedKeys.add(`${phase}:${tool ?? ""}`);
      phases.push(phase);
      phaseSequence += 1;
      await persistCommittedPhase(phase, tool, attemptId, phaseSequence);
    };

    const publish = async (event: PublicStreamEvent) => {
      events.push(event);
      await emit(event);
    };

    try {
      await recordPhase("run.started");
      await insertV9SkillRunReceipt(
        accounting,
        userId,
        caseId,
        turnId,
        attemptId,
        "turn",
        skillPackage,
      );
      await recordPhase("skill.bound");
      await publish({ type: "skill.bound" });
      emittedKeys.add("event:skill.bound::");

      const generation = agentGenerationSettings(options.generationModel, {
        thinking: "enabled",
        answerTokens: 8_192,
        thinkingTokens: 8_192,
      });
      const result = await (agent as unknown as {
        stream(
          messages: unknown[],
          streamOptions: {
            maxSteps: number;
            abortSignal: AbortSignal;
            modelSettings?: { maxOutputTokens?: number };
            providerOptions?: Record<string, unknown>;
            prepareStep: (input: { stepNumber: number }) => {
              activeTools: string[];
              toolChoice: "auto";
            };
          },
        ): Promise<{
          fullStream: AsyncIterable<{
            type: string;
            payload?: {
              toolName?: unknown;
              text?: unknown;
              args?: unknown;
              error?: unknown;
              stepResult?: { reason?: unknown };
              reason?: unknown;
            };
            object?: unknown;
          }>;
          totalUsage?: Promise<Record<string, unknown>>;
        }>;
      }).stream(messages, {
        maxSteps,
        abortSignal: abortController.signal,
        ...generation,
        // Thinking-mode providers reject named/required tool_choice. Restrict
        // the first step to read-case and keep tool_choice auto; the runner
        // still refuses any other public tool before case.loaded.
        prepareStep: ({ stepNumber }) => stepNumber === 0
          ? {
              activeTools: ["rectification-read-case"],
              toolChoice: "auto",
            }
          : {
              activeTools: [...RECTIFICATION_AGENT_TOOLS],
              toolChoice: "auto",
            },
      });

      const stepAnswer = createStepAnswerState();

      let spokenRaw = "";
      let visibleEmitted = "";

      const emitVisibleSpoken = async (visible: string) => {
        if (!caseLoaded) return;
        if (visible === visibleEmitted) {
          answerText = visible;
          return;
        }
        if (visible.startsWith(visibleEmitted)) {
          const growth = visible.slice(visibleEmitted.length);
          if (!growth) return;
          answerText = visible;
          answerDeltas.push(growth);
          visibleEmitted = visible;
          await emit({ type: "answer.delta", text: growth });
          return;
        }
        answerText = visible;
        answerDeltas.push(visible);
        visibleEmitted = visible;
        await emit({ type: "answer.delta", text: visible, replace: true });
      };

      const publishSpokenStep = async (pieces: readonly string[], live = false) => {
        const joined = pieces.join("");
        if (!joined) return;
        if (!caseLoaded) return;
        const spoken = live ? joined : joined.trim();
        if (!spoken) return;
        spokenRaw += spoken;
        await emitVisibleSpoken(spokenRaw);
      };

      const retractSpoken = async () => {
        const hadVisible = Boolean(visibleEmitted || answerText);
        spokenRaw = "";
        visibleEmitted = "";
        answerText = "";
        answerDeltas.length = 0;
        if (hadVisible && caseLoaded) {
          await emit({ type: "answer.delta", text: "", replace: true });
        }
      };

      const applyHostFallback = async (): Promise<boolean> => {
        if (answerText.trim()) return false;
        if (toolTerminalStatus.get("rectification-record-evidence-batch") !== "completed") {
          return false;
        }
        const spoken = composeHostFallbackNarration(batchToolResult ?? {});
        if (!spoken) return false;
        hostFallbackUsed = true;
        await recordPhase("answer.host_fallback");
        await publish({ type: "answer.host_fallback" });
        await emitVisibleSpoken(spoken);
        return true;
      };

      try {
      for await (const chunk of result.fullStream) {
        const rawToolName = typeof chunk.payload?.toolName === "string" ? chunk.payload.toolName : "";
        // Identical public tool-call + args are idempotent. Throwing
        // `repeated_tool_call` (BUG-368 P0-3) aborted the turn after
        // evidence / diagnostics / compare had already committed, because
        // the model often re-issued compare with the same caseId. Bound
        // loops with maxSteps / timeout instead; do not attempt.reset.
        let skipDuplicateToolCallReceipt = false;
        if (chunk.type === "tool-call") {
          if (rawToolName === "rectification-read-case" && !skillBound) {
            throw new Error("skill_not_bound");
          }
          if (isPublicRectificationToolName(rawToolName) && rawToolName !== "rectification-read-case" && !caseLoaded) {
            throw new Error("case_not_loaded");
          }
          if (isPublicRectificationToolName(rawToolName)) {
            const key = publicToolCallKey(rawToolName, chunk.payload);
            const count = (repeatedCalls.get(key) ?? 0) + 1;
            repeatedCalls.set(key, count);
            skipDuplicateToolCallReceipt = count > 1;
          }
        }

        const stepEffect = applyStepAnswerChunk(
          stepAnswer,
          chunk,
          isPublicRectificationToolName,
        );
        if (stepEffect.kind === "live") await publishSpokenStep([stepEffect.text], true);
        if (stepEffect.kind === "publish") await publishSpokenStep(stepEffect.pieces);
        if (stepEffect.kind === "retract") await retractSpoken();

        const activityEvent = skipDuplicateToolCallReceipt
          ? null
          : mapStreamChunkToActivity(chunk as never);
        if (activityEvent) {
          await publish(activityEvent);
          if (activityEvent.status === "started") {
            const changed = activityChangedFromTool(activityEvent.tool);
            if (!emittedActivities.has(changed.activity)) {
              emittedActivities.add(changed.activity);
              await publish(changed);
            }
          }
          if (activityEvent.status === "failed" || activityEvent.status === "completed") {
            toolTerminalStatus.set(activityEvent.tool, activityEvent.status);
          }
        }
        const capturedBatch = batchResultFromToolChunk(chunk as never);
        if (capturedBatch != null) batchToolResult = capturedBatch;
        const phaseEvent = skipDuplicateToolCallReceipt
          ? null
          : mapStreamChunkToPhase(chunk as never);
        if (phaseEvent) {
          if (phaseEvent.type === "skill.bound" && !skillBound) {
            skillBound = true;
            await insertV9SkillRunReceipt(
              accounting,
              userId,
              caseId,
              turnId,
              attemptId,
              "turn",
              skillPackage,
            );
          }
          if (phaseEvent.type === "case.loaded" && !skillBound) {
            throw new Error("skill_not_bound");
          }
          await recordPhase(phaseEvent.type, phaseEvent.tool ?? null);
          const key = `${phaseEvent.type}:${phaseEvent.tool ?? ""}:${(phaseEvent.methods ?? []).join(",")}`;
          if (!emittedKeys.has(`event:${key}`)) {
            emittedKeys.add(`event:${key}`);
            await publish(phaseEvent);
          }
          if (phaseEvent.type === "case.loaded") {
            caseLoaded = true;
            if (!intentClassified) {
              intentClassified = true;
              await recordPhase("intent.classified");
              await publish({ type: "intent.classified" });
            }
          }
        }
        for (const toolName of streamToolNames(chunk as never)) toolsUsed.add(toolName);
        if (chunk.type === "error" || chunk.type === "abort") streamFailed = true;
        if (chunk.type === "finish") {
          finished = true;
          finishReason = streamFinishReason(chunk);
        }
      }
      } catch (error) {
        if (!timedOut && !abortController.signal.aborted) throw error;
        streamFailed = true;
      }

      if (finished && !streamFailed) {
        const flushReason = finishReason === "length"
          ? "length"
          : finishReason === "stop" || finishReason === "unknown" || finishReason === null
            ? "stop"
            : finishReason;
        const flushed = flushStepAnswerOnStreamFinish(stepAnswer, flushReason);
        if (flushed.kind === "publish") await publishSpokenStep(flushed.pieces);
      }

      if (toolTerminalStatus.get("rectification-compare-candidates") === "failed") {
        await emitVisibleSpoken(withCompareFailedRetryNotice(answerText));
      }

      const completeAttempt = async (settleBilling = true): Promise<AttemptOutcome> => {
        let inputTokens = 0;
        let outputTokens = 0;
        let cache: ReturnType<typeof promptCacheUsage> = null;
        try {
          const raw = await (result.totalUsage ?? Promise.resolve({ inputTokens: 0, outputTokens: 0 }));
          inputTokens = Math.max(0, Math.trunc(typeof raw.inputTokens === "number" ? raw.inputTokens : 0));
          outputTokens = Math.max(0, Math.trunc(typeof raw.outputTokens === "number" ? raw.outputTokens : 0));
          cache = promptCacheUsage(raw);
        } catch {
          // Timeout/abort can leave provider usage unread.
        }
        await recordPhase("answer.composed");
        await publish({ type: "answer.composed" });
        return {
          ok: true,
          status: "completed",
          errorCode: null,
          usage: { inputTokens, outputTokens, ...(cache ? { cache } : {}) },
          answerText,
          answerDeltas,
          phases,
          toolsUsed: [...toolsUsed],
          events,
          skillBound,
          caseLoaded,
          attemptId,
          settleBilling,
        };
      };

      if (!skillBound) return failedAttempt(attemptId, "skill_not_loaded");
      if (!caseLoaded) return failedAttempt(attemptId, "case_not_loaded");
      const mapped = mapModelFinishToErrorCode({
        finishReason,
        aborted: abortController.signal.aborted,
        timedOut,
        answerText,
        stepCount: toolsUsed.size,
        maxSteps,
      });
      if (mapped === "run_timeout") return failedAttempt(attemptId, "run_timeout");
      if (streamFailed || abortController.signal.aborted) return failedAttempt(attemptId, mapped ?? "stream_aborted");
      if (!finished) return failedAttempt(attemptId, mapped ?? "stream_unfinished");
      if (mapped === "answer_truncated") {
        if (!await applyHostFallback()) {
          return {
            ok: false,
            status: "failed",
            errorCode: "answer_truncated",
            usage: { inputTokens: 0, outputTokens: 0 },
            answerText,
            answerDeltas,
            phases,
            toolsUsed: [...toolsUsed],
            events,
            skillBound,
            caseLoaded,
            attemptId,
          };
        }
      } else if (mapped === "max_steps" || mapped === "provider_error") {
        if (mapped !== "max_steps" || !await applyHostFallback()) {
          return failedAttempt(attemptId, mapped);
        }
      } else if (!answerText.trim() && !await applyHostFallback()) {
        const retryable = !publicWriteToolCompleted(toolTerminalStatus);
        return {
          ok: false,
          status: retryable ? "retryable" : "failed",
          errorCode: "empty_stream",
          usage: { inputTokens: 0, outputTokens: 0 },
          answerText: "",
          answerDeltas: [],
          phases: [...phases],
          toolsUsed: [...toolsUsed],
          events,
          skillBound,
          caseLoaded,
          attemptId,
        };
      }
      const needsWrite = turnExpectsEvidenceWrite(action, options.expectedWrite)
        || (action !== "opening" && action !== "read_only" && answerClaimsEvidenceRecorded(answerText));
      if (needsWrite && !publicWriteToolCompleted(toolTerminalStatus) && !hostFallbackUsed) {
        await retractSpoken();
        if (attemptNumber < MAX_ATTEMPTS) {
          return {
            ok: false,
            status: "retryable",
            errorCode: "evidence_not_written",
            usage: { inputTokens: 0, outputTokens: 0 },
            answerText: "",
            answerDeltas: [],
            phases: [...phases],
            toolsUsed: [...toolsUsed],
            events,
            skillBound,
            caseLoaded,
            attemptId,
          };
        }
        hostFallbackUsed = true;
        await recordPhase("answer.host_fallback");
        await publish({ type: "answer.host_fallback" });
        await emitVisibleSpoken(RECTIFICATION_USER_COPY.evidenceNotRecorded);
        return completeAttempt(false);
      }
      let latestDossier: V9CaseDossier;
      try {
        latestDossier = await loadV9CaseDossier(accounting, userId, caseId);
      } catch {
        return failedAttempt(attemptId, "state_invariant_failed");
      }
      const decision = decideFromDossier(latestDossier);
      if (decision.nextAction === "ask_candidate_discriminator") {
        const openFocus = latestDossier.conversationSummary.activeFocus;
        if (!(
          openFocus
          && isPersistedFocusId(openFocus.id)
          && parseAgentChoiceCopy(openFocus.expectedAnswerSchema)
        )) {
          return failedAttempt(attemptId, "state_invariant_failed");
        }
      }
      const askedFocus = latestDossier.conversationSummary.activeFocus;
      if (askedFocus?.askedTurnId === turnId) {
        const stem = focusSpokenPrompt(askedFocus.expectedAnswerSchema);
        if (stem) {
          const stripped = stripQuestionSentences(answerText, stem);
          const next = stripped || RECTIFICATION_USER_COPY.collectHandoff;
          if (next !== answerText) answerText = next;
        }
      }
      if (action === "opening") {
        const openingRange = openingRangeFromCandidateRange(latestDossier.case.candidateRange);
        if (!isAcceptableOpeningBody(answerText)) {
          answerText = openingSpokenBody(openingRange);
        }
      }
      const rangeAfterEvidence = previousInferenceFromReceipt(
        latestDossier.latestResult?.decisionReceipt ?? null,
      )?.credible_range ?? null;
      answerText = withRangeChangedAfterEvidence(
        answerText,
        rangeBeforeCompare,
        rangeAfterEvidence,
      );
      answerText = stripVerbalWindowChange(answerText)
        || RECTIFICATION_USER_COPY.declaredWindowLockedReply;
      if (answerText !== visibleEmitted) await emitVisibleSpoken(answerText);
      return completeAttempt();
    } finally {
      clearTimeout(timeout);
      signal?.removeEventListener("abort", onAbort);
      console.info(JSON.stringify({
        scope: "RectificationRunDiagnostic",
        runId: attemptId,
        modelId: options.modelName,
        finishReason: finishReason ?? "unknown",
        inputTokens: null,
        reasoningTokens: null,
        outputTokens: null,
        stepCount: toolsUsed.size,
        toolCallCount: toolsUsed.size,
        readCasePayloadBytes: null,
        elapsedMs: Date.now() - startedAt,
        lastCompletedTool: lastCompletedPublicTool(toolTerminalStatus),
        stateMutationCommitted: publicWriteToolCompleted(toolTerminalStatus) || hostFallbackUsed,
        expectedWrite: options.expectedWrite ?? null,
        collectIntent: options.collectIntent ?? null,
      }));
    }
  }

  function failedAttempt(attemptId: string, errorCode: string): AttemptOutcome {
    return {
      ok: false,
      status: isRetryableError(errorCode) ? "retryable" : "failed",
      errorCode,
      usage: { inputTokens: 0, outputTokens: 0 },
      answerText: "",
      answerDeltas: [],
      phases: [],
      toolsUsed: [],
      events: [],
      skillBound: false,
      caseLoaded: false,
      attemptId,
    };
  }

  async function persistCommittedPhase(
    phase: string,
    toolName: string | null,
    attemptId: string,
    sequence: number,
    strict = false,
  ) {
    try {
      await insertV9RunPhase(
        accounting,
        userId,
        caseId,
        turnId,
        phase,
        toolName,
        sequence,
        attemptId,
      );
    } catch (error) {
      if (strict) throw error;
      // Non-terminal activity receipts remain best effort. The completion
      // receipts above are strict because they are part of business truth.
    }
  }

  async function finalizeTurn(
    status: AttemptStatus,
    assistantText: string | null,
    attemptId: string,
    successfulAttemptId: string | null,
    strict = false,
  ) {
    try {
      await rpcOf(accounting, "finalize_agentic_rectification_turn", {
        p_user_id: userId,
        p_case_id: caseId,
        p_turn_id: turnId,
        p_attempt_id: attemptId,
        p_status: status,
        p_assistant_message: status === "completed" ? assistantText : null,
        p_successful_attempt_id: successfulAttemptId,
      });
    } catch (error) {
      if (strict) throw error;
      console.warn(`[rectification-v10] turn finalize failed turn=${turnId} status=${status} reason=${safeErrorCode(error)}`);
    }
  }
}

export function buildAgentMessages(
  options: V9AgentRunOptions,
  attempt: number,
  dossier: V9CaseDossier,
  skillInstructions: string,
  birthTimeClue: string | null = null,
  previousErrorCode: string | null = null,
): unknown[] {
  const timeContext = options.timeContext
    ?? `服务端当前时间（权威）：${new Date().toISOString()}。涉及“现在、今天、今年、未来几个月”等相对时间时，以此为准。`;
  const caseContext = `【服务端 Case ID】${options.caseId}。所有 rectification 工具调用的 caseId 必须原样使用此值。`;
  const bootstrapContent = [
    "【服务器已绑定当前 Case 的精确 Skill】运行器已在本 attempt 内加载并核验下列指令；不要重复调用 skill。第一步必须调用 rectification-read-case。",
    skillInstructions,
    ...(attempt > 1 ? [retryConstraintForAttempt(previousErrorCode)] : []),
  ].join("\n\n");
  const bootstrap = cachedSystemMessage(bootstrapContent, options.generationModel)
    ?? { role: "system" as const, content: bootstrapContent };
  if (options.action === "opening") {
    return [bootstrap, {
      role: "user",
      content: [timeContext, caseContext, buildOpeningBrief(dossier, birthTimeClue)].join("\n"),
    }];
  }
  return [bootstrap, {
    role: "user",
    content: [timeContext, caseContext, options.message ?? ""].join("\n"),
  }];
}

export { RECTIFICATION_SKILL_NAME, RECTIFICATION_SKILL_VERSION };