import type { Context } from "@earendil-works/chord";
import { BACKGROUND_CONTEXT } from "@earendil-works/chord/context";
import type { AssistantMessage, Message, ToolResultMessage } from "@earendil-works/pi-ai";
import {
  type AgentEvent,
  type AgentEventStream,
  type Conversation,
  createRegistry,
  defineExtension,
  type Extension,
  GenerationTask,
  Harness,
  type HarnessSettings,
  hook,
  type SettledSubmissionRecord,
  section,
  type ToolExecutionApi,
  type ToolExecutionResult,
  type ToolRegistration,
  ToolTask,
  type UsageState,
  watchEvents,
} from "@earendil-works/pi-durable";
import {
  type AgentBudgetKind,
  type AgentRunEvent,
  type AgentRunEventBody,
  type AgentUsageTotals,
  MODEL_IMAGE_KEY,
} from "@umt/contracts";
import type { TSchema } from "typebox";
import { isCapabilityError } from "../capability/errors.ts";
import type { CapabilityRegistry, RegisteredCapability } from "../capability/registry.ts";
import { toolNameToCapabilityId } from "../capability/tool-name.ts";
import { redactForModel, redactString, summarizeForLog } from "../redact.ts";
import { EventChannel } from "../runtime/event-channel.ts";
import {
  type AgentRunHandle,
  type AgentRuntimePort,
  type CreateRunSpec,
  emptyUsage,
  type InjectedEventBody,
  type RunOutcome,
} from "../runtime/port.ts";
import { type ResolvedAgentSettings, resolveAgentSettings } from "../settings.ts";
import { compactionFromSettings, type PiModelResolver, resolveModelFromSettings } from "./model.ts";
import { openRunStorage } from "./session.ts";
import {
  PiUpstreamClient,
  type UpstreamClientOptions,
  type UpstreamCredentialStore,
  type UpstreamSpec,
} from "./upstream.ts";

export interface PiAgentRuntimeOptions {
  registry: CapabilityRegistry;
  /** 默认按设置构造真实上游;测试用 FakeModelProvider 覆盖 */
  resolveModel?: PiModelResolver;
  /**
   * 按上游实例装配(W7-A):独立 Models + 只映射本上游 secret 的凭据存储(OAuth 按需刷新、锁内写回)。
   * 给了 resolveModel 时以 resolveModel 为准。
   */
  upstream?: { spec: UpstreamSpec; credentials: UpstreamCredentialStore; options?: UpstreamClientOptions };
  /**
   * 每个事件产生时回调(server 用它写 agent_run_event 再推 SSE)。
   * 同一 run 内按 seq 顺序串行调用(前一个 Promise 结束才调下一个);回调抛错只记入 onEventError,不影响运行。
   * close() 会等待已排队的回调全部完成。
   */
  onEvent?: (event: AgentRunEvent) => void | Promise<void>;
  onEventError?: (error: unknown, event: AgentRunEvent) => void;
  /**
   * 运行时内部错误(不影响本次调用结果、但应该看见的):本次运行扩展里的系统提示 section、能力工具、白名单 / 预算钩子抛错,
   * 以及 pi-durable 的 onReport(提交 / 进度写入等失败)。message 已脱敏。不能抛错。
   */
  onReport?: (report: RuntimeReport) => void;
  now?: () => Date;
}

export interface RuntimeReport {
  runId: string;
  /** 出错的位置:section:umt、tool:<工具名>、hook:beforeTool、hook:beforeRequest、harness(pi-durable onReport) */
  source: string;
  /** 已脱敏的错误信息 */
  message: string;
  error: unknown;
}

export const DEFAULT_COMPACTION_INSTRUCTIONS =
  "保留:本次目标、约束与授权范围、证据引用(对象 ID/记录)、已完成的动作及其结果(已提交≠已入库)、剩余事项。不要保留任何凭证。";

/** 可挂起重放的能力:首次执行时写入的调用级 memo(重放时据此区分) */
const STARTED_MEMO = "umt.started";
const SUMMARY_LIMIT = 600;
/** pi-durable 默认把工具文本截到 50KB / 2000 行;能力输出已自带分页与脱敏,沿用 0.99 时代不截断的行为 */
const TOOL_OUTPUT_LIMITS = { maxBytes: 8 * 1024 * 1024, maxLines: 1_000_000 } as const;

/**
 * Pi(pi-durable Harness)实现的 AgentRuntimePort。
 *
 * 一个 run = 一个 pi-durable 存储(sessionDir/<runId>/ 下的 JSONL)+ 其中的 root 对话。
 * Harness 只在本进程打开一次(单写者),close() 后其他进程才能接手续跑。
 */
export class PiAgentRuntime implements AgentRuntimePort {
  constructor(private readonly options: PiAgentRuntimeOptions) {}

  async createRun(spec: CreateRunSpec): Promise<AgentRunHandle> {
    const ctx = BACKGROUND_CONTEXT;
    const settings = resolveAgentSettings(spec.settings);
    const capabilities = this.options.registry.resolve(spec.capabilities);
    const upstream = this.options.upstream;
    const resolved = this.options.resolveModel
      ? await this.options.resolveModel(settings)
      : upstream
        ? await new PiUpstreamClient(upstream.spec, upstream.credentials, upstream.options).resolveRunModel(settings)
        : resolveModelFromSettings(settings);
    const { models, model } = resolved;
    const thinkingLevel = resolved.thinkingLevel ?? settings.thinkingLevel;
    const opened = await openRunStorage(spec.sessionDir, spec.runId, ctx);

    const run = new PiAgentRun(spec, settings, capabilities, this.options, opened.path);
    run.supportsImages = Array.isArray(model.input) && model.input.includes("image");
    const registry = createRegistry();
    registry.install(run.extension());
    const harnessSettings: HarnessSettings = {
      stream: { timeoutMs: settings.requestTimeoutMs, maxRetries: settings.maxRetries },
      retry: { enabled: settings.maxRetries > 0, maxRetries: settings.maxRetries, baseDelayMs: 1000 },
      compaction: compactionFromSettings(settings, model.contextWindow),
      toolExecution: "sequential",
    };
    let harness: Harness | undefined;
    try {
      harness = await Harness.open(
        opened.storage,
        { models, registry, settings: harnessSettings, onReport: (error) => run.report("harness", error) },
        ctx,
      );
      const agent = { model: { provider: model.provider, modelId: model.id }, thinkingLevel };
      const root = await harness.root(ctx, { agent });
      // 已有会话:模型 / 思考档位跟随本次设置(换了上游或模型后继续)
      await root.configure(agent, ctx);
      await run.attach(harness, root);
    } catch (error) {
      if (harness) await harness.close(ctx).catch(() => undefined);
      else await opened.storage.close(ctx).catch(() => undefined);
      throw error;
    }
    return run;
  }
}

type BudgetState = { kind: AgentBudgetKind; message: string };
type ContentBlock = AssistantMessage["content"][number];

class PiAgentRun implements AgentRunHandle {
  readonly runId: string;
  readonly sessionPath: string;
  pendingOperations = 0;
  /** 上游模型声明支持图片输入(model.input 含 image) */
  supportsImages = false;

  private readonly channel = new EventChannel<AgentRunEvent>();
  private readonly allowed = new Map<string, string>();
  private readonly rejected = new Set<string>();
  /** 已发出 tool_call_start 的调用 */
  private readonly started = new Set<string>();
  /** 打开会话时仍在执行中的调用(崩溃 / 挂起前未完成):恢复后的事件带 recovery */
  private readonly recovering = new Map<string, { toolName: string; taskId: unknown }>();
  /** 模型给出的调用参数(未经过 intent 的调用,结束时补发 start 用) */
  private readonly callArgs = new Map<string, unknown>();
  private readonly abortController = new AbortController();
  private harness!: Harness;
  private root!: Conversation;
  private stream: AgentEventStream | undefined;
  private readonly firstSeq: number;
  private seq: number;
  private delivery: Promise<void> = Promise.resolve();
  private toolCalls = 0;
  private usage: AgentUsageTotals = emptyUsage();
  private finalText = "";
  /** finalText 之后没有再调用工具(是收尾文字而不是"接下来我要…"的中间过程) */
  private finalIsTail = false;
  private budget: BudgetState | undefined;
  private userAbort: string | undefined;
  private startedAt: number | undefined;
  private deadlineTimer: NodeJS.Timeout | undefined;
  private closed = false;
  private closing: Promise<void> | undefined;
  /** 正在执行的工具数:工具内部触发 close()(挂起等待)时不能等 Harness 关闭,它要等这个工具结束 */
  private activeTools = 0;
  /** 流式中的 assistant 内容(由 message_start / message_update 维护)与各块已发出的字数 */
  private partial: ContentBlock[] | undefined;
  private shown: number[] = [];
  private batches = 0;
  private consuming = false;

  constructor(
    private readonly spec: CreateRunSpec,
    private readonly settings: ResolvedAgentSettings,
    private readonly capabilities: RegisteredCapability[],
    private readonly options: PiAgentRuntimeOptions,
    sessionPath: string,
  ) {
    this.runId = spec.runId;
    this.sessionPath = sessionPath;
    this.firstSeq = spec.firstSeq ?? 0;
    this.seq = this.firstSeq;
    for (const c of capabilities) this.allowed.set(c.toolName, c.descriptor.id);
  }

  /** 本次运行的 pi-durable 扩展:系统提示 + 能力工具 + 白名单 / 预算钩子 */
  extension(): Extension {
    const systemPrompt = this.spec.systemPrompt;
    return defineExtension({
      name: "umt",
      sections: [
        section(
          "umt",
          this.reported("section:umt", () => systemPrompt),
          { tag: false },
        ),
      ],
      tools: this.capabilities.map((c) => this.toTool(c)),
      hooks: [
        hook(ToolTask, {
          beforeTool: this.reported("hook:beforeTool", (call: { id: string; name: string }) =>
            this.beforeTool(call.id, call.name),
          ),
        }),
        hook(GenerationTask, {
          beforeRequest: this.reported("hook:beforeRequest", () => {
            this.checkDeadline();
            this.checkCost();
            if (this.budget) this.scheduleAbort();
            return undefined;
          }),
        }),
      ],
    });
  }

  /** 记一条运行时内部错误(脱敏后交给 onReport;onReport 自身出错忽略) */
  report(source: string, error: unknown): void {
    try {
      const message = redactString(error instanceof Error ? error.message : String(error));
      this.options.onReport?.({ runId: this.runId, source, message, error });
    } catch {
      // 上报失败不影响运行
    }
  }

  /** 包一层:抛错时先上报(带位置)再原样抛出,pi-durable 的处理语义不变 */
  private reported<A extends unknown[], R>(source: string, fn: (...args: A) => R): (...args: A) => R {
    return (...args: A) => {
      try {
        return fn(...args);
      } catch (error) {
        this.report(source, error);
        throw error;
      }
    };
  }

  /** 能力 → Pi 工具。执行时统一走注册中心 invoke(schema/作用域/授权/幂等/审计) */
  toTool(entry: RegisteredCapability): ToolRegistration {
    const { descriptor } = entry;
    const registry = this.options.registry;
    const gate = this.spec.replayGate;
    // 可挂起重放(见 CreateRunSpec.replayGate):对 harness 声明 safe,执行时用 memo 判断是否重放
    const gated = descriptor.replay === "never" && !!gate;
    return {
      name: entry.toolName,
      description: `${descriptor.description}\n[能力 ${descriptor.id} v${descriptor.version};类别 ${descriptor.kind};风险 ${descriptor.riskLevel}]`,
      parameters: registry.inputJsonSchema(descriptor.id) as unknown as TSchema,
      replay: gated || descriptor.replay === "safe" ? "safe" : "unsafe",
      outputLimits: TOOL_OUTPUT_LIMITS,
      ...(descriptor.kind === "execute" ? { executionMode: "sequential" as const } : {}),
      execute: (params, api, context) =>
        this.executeTool(entry, gated, params, api, context).catch((error: unknown) => {
          if (!this.closed && !context.abortSignal?.aborted) this.report(`tool:${entry.toolName}`, error);
          throw error;
        }),
    };
  }

  private async executeTool(
    entry: RegisteredCapability,
    gated: boolean,
    params: unknown,
    api: ToolExecutionApi,
    context: Context,
  ): Promise<ToolExecutionResult> {
    const { descriptor } = entry;
    const registry = this.options.registry;
    const actor = this.spec.actor;
    const gate = this.spec.replayGate;
    const signals = [this.abortController.signal, ...(context.abortSignal ? [context.abortSignal] : [])];
    const signal = AbortSignal.any(signals);
    // 稳定的调用 ID:同一存储内唯一,挂起 / 崩溃后重放时不变(审批、子会话等待按它对应)
    const invocationId = `${api.taskId}:${api.callId}`;
    this.activeTools += 1;
    try {
      if (gated) {
        const replayed = (await api.memo<boolean>(STARTED_MEMO, context)) === true;
        if (!replayed) await api.memo(STARTED_MEMO, true, context);
        else if (!gate?.(descriptor.id, invocationId))
          return errorResult(
            JSON.stringify({
              error: {
                code: "interrupted",
                message: "执行被中断,外部结果未知;先用查询工具核对实际状态,不要直接重复提交",
              },
            }),
          );
      }
      try {
        const output = await registry.invoke(descriptor.id, params, {
          actorId: actor.actorId,
          taskId: actor.taskId,
          scopeIds: actor.scopeIds,
          ...(actor.allowedKinds ? { allowedKinds: actor.allowedKinds } : {}),
          signal,
          source: "agent",
          invocationId,
          agentRunId: this.runId,
          modelImages: this.supportsImages,
        });
        // 会话已在本调用内关闭(挂起等待):结果不落盘,恢复后重放本调用
        if (this.closed) throw new Error("会话已挂起");
        const content = toolContent(output, this.supportsImages);
        const notice = this.budgetNotice();
        if (notice) content.push({ type: "text", text: notice });
        return { content };
      } catch (error) {
        // 中止 / 挂起:交给 harness(中止写 aborted 结果;挂起时调用保持未完成)
        if (this.closed || context.abortSignal?.aborted) throw error;
        const notice = this.budgetNotice();
        const tail = notice ? `\n${notice}` : "";
        if (isCapabilityError(error))
          return errorResult(
            JSON.stringify({ error: { code: error.code, message: redactString(error.message) } }) + tail,
          );
        return errorResult(redactString(error instanceof Error ? error.message : String(error)) + tail);
      }
    } finally {
      this.activeTools -= 1;
    }
  }

  async attach(harness: Harness, root: Conversation): Promise<void> {
    const ctx = BACKGROUND_CONTEXT;
    this.harness = harness;
    this.root = root;
    const inspection = await harness.inspect(ctx);
    const inputs = inspection.submissions.filter((s) => s.conversationId === root.id && s.type === "input");
    this.pendingOperations = inputs.length > 0 ? inputs.length : inspection.tasks.length > 0 ? 1 : 0;
    const stream = await watchEvents(harness, root.id, ctx);
    for (const slot of stream.snapshot.tools)
      if (slot.status === "running") this.recovering.set(slot.callId, { toolName: slot.name, taskId: slot.taskId });
    this.usage = usageTotals(stream.snapshot.usage);
    this.stream = stream;
    stream.start(async (events) => {
      this.batches += 1;
      this.consuming = true;
      try {
        for (const event of events) await this.onAgentEvent(event);
      } finally {
        this.consuming = false;
      }
    });
  }

  /** 白名单二次校验 + 工具调用次数预算(ToolTask.beforeTool) */
  private beforeTool(toolCallId: string, toolName: string): { block: string } | undefined {
    const decision = checkToolCall(toolName, this.allowed);
    if (!decision.allowed) {
      this.reject(toolCallId, toolName, decision.reason);
      return { block: decision.reason };
    }
    if (this.budget) {
      // 已超预算、中止尚未落地:不再执行任何工具
      this.reject(toolCallId, toolName, this.budget.message);
      return { block: this.budget.message };
    }
    this.toolCalls += 1;
    const extCalls = this.spec.externalUsage?.().toolCalls ?? 0;
    if (this.toolCalls + extCalls > this.settings.maxToolCalls) {
      const reason =
        extCalls > 0
          ? `工具调用次数超出预算(${this.settings.maxToolCalls},含子会话 ${extCalls} 次)`
          : `工具调用次数超出预算(${this.settings.maxToolCalls})`;
      this.exceed("tool_calls", reason);
      this.rejected.add(toolCallId);
      this.emit({
        type: "tool_call_rejected",
        toolCallId,
        toolName,
        capabilityId: decision.capabilityId,
        reason,
      });
      return { block: reason };
    }
    return undefined;
  }

  /** pi-durable 事件(每次提交一批)→ AgentRunEvent */
  private async onAgentEvent(e: AgentEvent): Promise<void> {
    switch (e.type) {
      case "message_start": {
        if (e.message.role !== "assistant") return;
        this.partial = structuredClone(e.message.content);
        this.shown = [];
        this.flushDeltas(this.partial);
        return;
      }
      case "message_update": {
        for (const c of e.changes) {
          if (c.type === "message") {
            this.partial = structuredClone(c.message.content);
            continue;
          }
          if (!this.partial) continue;
          if (
            c.type === "text_start" ||
            c.type === "thinking_start" ||
            c.type === "toolcall_start" ||
            c.type === "block"
          )
            this.partial[c.contentIndex] = structuredClone(c.block);
          else if (c.type === "text_delta" || c.type === "thinking_delta") {
            const block = this.partial[c.contentIndex];
            if (block?.type === "text") block.text += c.delta;
            else if (block?.type === "thinking") block.thinking += c.delta;
          }
        }
        if (this.partial) this.flushDeltas(this.partial);
        return;
      }
      case "message_end": {
        const message = e.entry.model?.[0] as Message | undefined;
        if (message?.role !== "assistant") return;
        this.flushDeltas(message.content);
        this.partial = undefined;
        this.shown = [];
        const text = assistantText(message);
        const calls = message.content.filter((c) => c.type === "toolCall");
        for (const c of calls) this.callArgs.set(c.id, c.arguments);
        if (text.trim()) {
          this.finalText = text;
          this.finalIsTail = calls.length === 0;
        } else if (calls.length > 0) this.finalIsTail = false;
        return;
      }
      case "tool_execution_start": {
        if (this.rejected.has(e.toolCallId) || this.started.has(e.toolCallId)) return;
        const capabilityId = this.allowed.get(e.toolName);
        if (!capabilityId) return;
        this.emitToolStart(e.toolCallId, capabilityId, e.args);
        return;
      }
      case "tool_execution_end": {
        if (this.rejected.has(e.toolCallId)) return;
        const capabilityId = this.allowed.get(e.toolName);
        if (!capabilityId) {
          // 模型调用了本次没提供的工具(例如 bash):harness 已合成 tool_unavailable 错误,这里只产生拒绝事件
          this.reject(e.toolCallId, e.toolName, `工具 ${e.toolName} 不在本次白名单内`);
          return;
        }
        // 参数校验失败、恢复后才结束的调用没有 start:先补一条
        if (!this.started.has(e.toolCallId))
          this.emitToolStart(e.toolCallId, capabilityId, this.callArgs.get(e.toolCallId) ?? {});
        const result = e.entry?.model?.[0] as Message | undefined;
        const tool = result?.role === "toolResult" ? (result as ToolResultMessage) : undefined;
        const text = tool
          ? tool.content
              .map((c) => (c.type === "text" ? c.text : `[${c.type}]`))
              .join("")
              .trim()
          : "工具调用没有结果(执行被中断)";
        this.emit({
          type: "tool_call_end",
          toolCallId: e.toolCallId,
          capabilityId,
          isError: tool?.isError ?? true,
          outputSummary: summarizeForLog(text, SUMMARY_LIMIT),
          recovery: this.recovering.has(e.toolCallId),
          ...(tool?.durationMs !== undefined ? { durationMs: tool.durationMs } : {}),
        });
        return;
      }
      case "usage_changed":
        this.usage = usageTotals(e.usage);
        this.checkCost();
        return;
      case "compaction_start":
        this.emit({ type: "compaction", phase: "start", reason: e.reason });
        return;
      case "compaction_end": {
        const task = await this.harness.getTask(e.taskId, BACKGROUND_CONTEXT).catch(() => undefined);
        const state = task?.state;
        const outcome = state && state.status === "terminal" ? state.outcome.status : undefined;
        const status =
          outcome === "completed" ? "completed" : outcome === "aborted" ? "aborted" : outcome ? "failed" : undefined;
        this.emit({ type: "compaction", phase: "end", reason: e.reason, ...(status ? { status } : {}) });
        return;
      }
      default:
        return;
    }
  }

  /** 流式文本:按块补发尚未发出的部分 */
  private flushDeltas(content: readonly ContentBlock[]): void {
    content.forEach((block, i) => {
      if (block.type !== "text" && block.type !== "thinking") return;
      const text = block.type === "text" ? block.text : block.thinking;
      const done = this.shown[i] ?? 0;
      if (text.length <= done) return;
      this.shown[i] = text.length;
      const delta = text.slice(done);
      this.emit(block.type === "text" ? { type: "text_delta", delta } : { type: "thinking_delta", delta });
    });
  }

  private emitToolStart(toolCallId: string, capabilityId: string, args: unknown): void {
    this.started.add(toolCallId);
    this.finalIsTail = false;
    this.emit({
      type: "tool_call_start",
      toolCallId,
      capabilityId,
      inputSummary: summarizeForLog(args, SUMMARY_LIMIT),
      recovery: this.recovering.has(toolCallId),
    });
  }

  /** 恢复时为打开会话时仍在执行的调用补发 start(带 recovery),参数取自工具任务的 intent */
  private async announceRecovery(): Promise<void> {
    for (const [callId, slot] of this.recovering) {
      if (this.started.has(callId) || this.rejected.has(callId)) continue;
      const capabilityId = this.allowed.get(slot.toolName) ?? toolNameToCapabilityId(slot.toolName) ?? slot.toolName;
      let args: unknown = this.callArgs.get(callId) ?? {};
      if (slot.taskId !== undefined) {
        const task = await this.harness.getTask(slot.taskId as never, BACKGROUND_CONTEXT).catch(() => undefined);
        const checkpoint = task?.state.checkpoint as { arguments?: unknown } | undefined;
        if (checkpoint?.arguments) args = checkpoint.arguments;
      }
      this.emitToolStart(callId, capabilityId, args);
    }
  }

  events(fromSeq = this.firstSeq): AsyncIterable<AgentRunEvent> {
    return this.channel.iterate(fromSeq - this.firstSeq);
  }

  history(): AgentRunEvent[] {
    return this.channel.snapshot();
  }

  async prompt(text?: string): Promise<RunOutcome> {
    const message = text ?? this.spec.goal;
    return this.drive(false, async () => {
      const submission = await this.root.submit({ type: "input", content: message }, BACKGROUND_CONTEXT);
      return submission.wait(BACKGROUND_CONTEXT);
    });
  }

  async resume(): Promise<RunOutcome> {
    return this.drive(true, async () => {
      const ctx = BACKGROUND_CONTEXT;
      const inspection = await this.harness.inspect(ctx);
      const inputs = inspection.submissions.filter((s) => s.conversationId === this.root.id && s.type === "input");
      if (inspection.tasks.length === 0 && inputs.length === 0) return undefined;
      await this.announceRecovery();
      this.harness.resume();
      let last: SettledSubmissionRecord | undefined;
      for (const s of inputs) {
        const submission = await this.harness.submission(s.id, ctx);
        if (submission) last = await submission.wait(ctx);
      }
      if (!last) await this.root.waitForIdle(ctx);
      return last;
    });
  }

  async abort(reason = "用户取消"): Promise<void> {
    this.userAbort = reason;
    this.abortController.abort(reason);
    await this.root?.abort(BACKGROUND_CONTEXT).catch(() => undefined);
  }

  appendEvent(body: InjectedEventBody): boolean {
    if (this.closed) return false;
    this.emit(body);
    return true;
  }

  stats(): { toolCalls: number; usage: AgentUsageTotals } {
    return { toolCalls: this.toolCalls, usage: { ...this.usage } };
  }

  async compact(customInstructions = DEFAULT_COMPACTION_INSTRUCTIONS): Promise<void> {
    const ctx = BACKGROUND_CONTEXT;
    const id = await this.root.compact(customInstructions, ctx);
    const settled = await this.harness.waitForTask(id, ctx);
    const outcome = settled.state.outcome;
    if (outcome.status === "failed") throw new Error(`压缩失败: ${outcome.error.message}`);
    if (outcome.status === "completed" && outcome.result.submissionId !== undefined) {
      const placed = await this.harness.submission(outcome.result.submissionId, ctx);
      await placed?.wait(ctx);
    }
    await this.settleEvents();
  }

  /**
   * 关闭会话(Harness + 存储)。挂起等待时由工具内部调用:此时 Harness 关闭要等这个工具结束,
   * 所以只封口(之后的提交一律被拒,本调用在存储里保持未完成)就返回,由外层再次 close() 等待真正关完。
   */
  async close(): Promise<void> {
    if (!this.closing) {
      this.closed = true;
      this.clearDeadline();
      // 同步调用:立即封口,工具随后的结果提交会被拒绝
      const harnessClosed = this.harness ? this.harness.close(BACKGROUND_CONTEXT) : Promise.resolve();
      this.closing = (async () => {
        await harnessClosed.catch(() => undefined);
        await this.stream?.stop().catch(() => undefined);
        this.channel.close();
        await this.delivery;
      })();
    }
    if (this.activeTools > 0) return;
    await this.closing;
  }

  // ---- internals ----

  private async drive(
    resumed: boolean,
    start: () => Promise<SettledSubmissionRecord | undefined>,
  ): Promise<RunOutcome> {
    if (this.closed) throw new Error(`run ${this.runId} 已关闭`);
    this.startedAt ??= Date.now();
    this.emit({ type: "run_started", goal: this.spec.goal, resumed });
    this.checkDeadline();
    if (this.budget) return this.finish(undefined);
    this.armDeadline();
    let record: SettledSubmissionRecord | undefined;
    try {
      record = await start();
    } finally {
      this.clearDeadline();
    }
    await this.settleEvents();
    this.pendingOperations = 0;
    return this.finish(record);
  }

  /** 等事件流把已提交的批次处理完(批次在提交时同步生成,经微任务投递) */
  private async settleEvents(): Promise<void> {
    for (let i = 0; i < 1000; i++) {
      const before = this.batches;
      await new Promise((r) => setTimeout(r, 0));
      while (this.consuming) await new Promise((r) => setTimeout(r, 1));
      if (this.batches === before) return;
    }
  }

  private finish(record: SettledSubmissionRecord | undefined): RunOutcome {
    let outcome: RunOutcome;
    if (this.budget) {
      outcome = this.failed("budget", this.budget.message, this.budget.kind);
    } else if (this.userAbort || (record?.status === "unanswered" && record.reason === "aborted")) {
      outcome = this.failed("aborted", this.userAbort ?? "已中止");
    } else if (record?.status === "unanswered") {
      outcome = this.failed("error", redactString(unansweredMessage(record)));
    } else {
      outcome = { status: "completed", finalText: this.finalText, usage: this.usage };
      this.emit({ type: "run_completed", finalText: outcome.finalText, usage: this.usage });
      return outcome;
    }
    this.emit({
      type: "run_failed",
      reason: outcome.reason ?? "error",
      message: outcome.message ?? "",
      usage: this.usage,
      ...(outcome.budget ? { budget: outcome.budget } : {}),
    });
    return outcome;
  }

  private failed(reason: RunOutcome["reason"] & string, message: string, budget?: AgentBudgetKind): RunOutcome {
    return {
      status: "failed",
      reason,
      message,
      // 被预算 / 中止打断时,工具调用之前的文字只是中间过程("我再查一下…"),不作为汇报
      finalText: (reason === "budget" || reason === "aborted") && !this.finalIsTail ? "" : this.finalText,
      usage: this.usage,
      ...(budget ? { budget } : {}),
    };
  }

  private reject(toolCallId: string, toolName: string, reason: string): void {
    if (this.rejected.has(toolCallId)) return;
    this.rejected.add(toolCallId);
    const capabilityId = toolNameToCapabilityId(toolName);
    this.emit({ type: "tool_call_rejected", toolCallId, toolName, ...(capabilityId ? { capabilityId } : {}), reason });
  }

  private exceed(kind: AgentBudgetKind, message: string): void {
    if (this.budget) return;
    this.budget = { kind, message };
    this.scheduleAbort();
  }

  /** 不能在事件/钩子回调里同步等待 abort(会与事件投递互相等待),放到下一个宏任务 */
  private scheduleAbort(): void {
    setTimeout(() => {
      this.abortController.abort("budget");
      void this.root.abort(BACKGROUND_CONTEXT).catch(() => undefined);
    }, 0);
  }

  /**
   * 预算将尽提醒(AG-3):工具调用只剩 ≤2 次,或 token / 费用 / 时长用到 85%,在工具结果末尾追加一句,
   * 让模型停止探查、先写中文小结,避免被硬截断后只剩中间过程。
   */
  private budgetNotice(): string | null {
    const out: string[] = [];
    const ext = this.spec.externalUsage?.();
    const left = this.settings.maxToolCalls - this.toolCalls - (ext?.toolCalls ?? 0);
    if (left <= 2) out.push(`工具调用只剩 ${Math.max(0, left)} 次`);
    const cost = this.settings.costBudget;
    const tokens = this.usage.totalTokens + (ext?.totalTokens ?? 0);
    const usd = this.usage.costUsd + (ext?.costUsd ?? 0);
    if (cost?.maxTotalTokens !== undefined && tokens >= cost.maxTotalTokens * 0.85)
      out.push(`token 已用 ${tokens}/${cost.maxTotalTokens}`);
    if (cost?.maxCostUsd !== undefined && usd >= cost.maxCostUsd * 0.85)
      out.push(`费用已用 $${usd.toFixed(4)}/$${cost.maxCostUsd}`);
    if (this.startedAt !== undefined && Date.now() - this.startedAt >= this.settings.maxDurationMs * 0.85)
      out.push("运行时长接近上限");
    if (out.length === 0) return null;
    return `【预算提醒】${out.join("；")}。停止探查，下一条回复直接用中文写小结（已完成什么、证据、还缺什么）。`;
  }

  /** token / 费用预算:自身用量 + 子会话用量(externalUsage) */
  private checkCost(): void {
    const cost = this.settings.costBudget;
    if (!cost) return;
    const ext = this.spec.externalUsage?.();
    const tokens = this.usage.totalTokens + (ext?.totalTokens ?? 0);
    const usd = this.usage.costUsd + (ext?.costUsd ?? 0);
    const note = ext && (ext.totalTokens > 0 || ext.costUsd > 0) ? "(含子会话)" : "";
    if (cost.maxTotalTokens !== undefined && tokens > cost.maxTotalTokens) {
      this.exceed("tokens", `token 用量 ${tokens}${note} 超出预算 ${cost.maxTotalTokens}`);
    } else if (cost.maxCostUsd !== undefined && usd > cost.maxCostUsd) {
      this.exceed("cost", `费用 $${usd.toFixed(4)}${note} 超出预算 $${cost.maxCostUsd}`);
    }
  }

  private checkDeadline(): void {
    if (this.startedAt !== undefined && Date.now() - this.startedAt > this.settings.maxDurationMs) {
      this.exceed("duration", `运行时长超出预算(${this.settings.maxDurationMs}ms)`);
    }
  }

  private armDeadline(): void {
    const remaining = (this.startedAt ?? Date.now()) + this.settings.maxDurationMs - Date.now();
    this.deadlineTimer = setTimeout(
      () => this.exceed("duration", `运行时长超出预算(${this.settings.maxDurationMs}ms)`),
      Math.max(0, remaining),
    );
  }

  private clearDeadline(): void {
    if (this.deadlineTimer) clearTimeout(this.deadlineTimer);
    this.deadlineTimer = undefined;
  }

  private emit(body: AgentRunEventBody): void {
    const event = { ...body, runId: this.runId, seq: this.seq++, at: new Date().toISOString() } as AgentRunEvent;
    this.channel.push(event);
    const onEvent = this.options.onEvent;
    if (onEvent) {
      // 串行投递:落库顺序与 seq 一致
      this.delivery = this.delivery
        .then(() => onEvent(event))
        .catch((error) => this.options.onEventError?.(error, event));
    }
  }
}

/**
 * 能力输出 → 工具结果内容块。输出里的 modelImage(见 @umt/contracts MODEL_IMAGE_KEY)摘出来:
 * 上游支持图片 → image content;不支持 → 丢弃并在文本里提示。图片数据不进 JSON 文本、不经文本脱敏。
 */
export function toolContent(
  output: unknown,
  vision: boolean,
): Array<{ type: "text"; text: string } | { type: "image"; data: string; mimeType: string }> {
  let rest = output;
  let image: { data: string; mimeType: string } | null = null;
  if (output && typeof output === "object" && !Array.isArray(output) && MODEL_IMAGE_KEY in output) {
    const { [MODEL_IMAGE_KEY]: img, ...others } = output as Record<string, unknown>;
    rest = others;
    const m = img as { data?: unknown; mimeType?: unknown } | null | undefined;
    if (m && typeof m.data === "string" && typeof m.mimeType === "string")
      image = { data: m.data, mimeType: m.mimeType };
  }
  const text = JSON.stringify(redactForModel(rest) ?? null);
  if (!image) return [{ type: "text", text }];
  if (!vision)
    return [
      { type: "text", text },
      { type: "text", text: "(截图未发送:当前模型不支持图片输入;用 browser.snapshot 读取页面文字)" },
    ];
  return [
    { type: "text", text },
    { type: "image", data: image.data, mimeType: image.mimeType },
  ];
}

/** before_tool 白名单判定(纯函数,便于单测) */
export function checkToolCall(
  toolName: string,
  allowed: ReadonlyMap<string, string>,
): { allowed: true; capabilityId: string } | { allowed: false; reason: string; capabilityId?: string } {
  const capabilityId = allowed.get(toolName);
  if (capabilityId) return { allowed: true, capabilityId };
  const mapped = toolNameToCapabilityId(toolName);
  return {
    allowed: false,
    reason: `工具 ${toolName} 不在本次白名单内`,
    ...(mapped ? { capabilityId: mapped } : {}),
  };
}

function errorResult(text: string): ToolExecutionResult {
  return { content: [{ type: "text", text }], isError: true };
}

/** 没有得到回答的输入 → 失败原因 */
function unansweredMessage(record: SettledSubmissionRecord): string {
  const detail = typeof record.detail === "string" ? record.detail : record.detail ? JSON.stringify(record.detail) : "";
  switch (record.reason) {
    case "model_error":
      return `模型调用失败: ${detail || "未知错误"}`;
    case "no_model":
      return "模型不可用:检查模型服务与模型设置";
    case "faulted":
      return `运行出错: ${detail || "未知错误"}`;
    default:
      return `本轮没有完成(${record.reason}${detail ? `: ${detail}` : ""})`;
  }
}

function assistantText(message: AssistantMessage): string {
  return message.content
    .filter((c) => c.type === "text")
    .map((c) => c.text)
    .join("");
}

/** pi.usage(按 provider/model 与工具分桶)→ 合计 */
function usageTotals(u: UsageState): AgentUsageTotals {
  const t = emptyUsage();
  for (const x of [...Object.values(u.models), ...Object.values(u.tools)]) {
    t.inputTokens += x.input;
    t.outputTokens += x.output;
    t.cacheReadTokens += x.cacheRead;
    t.cacheWriteTokens += x.cacheWrite;
    t.totalTokens += x.totalTokens;
    t.costUsd += x.cost.total;
  }
  return t;
}