opencode/.opencode/tool/agentpaas_run.ts

/// <reference path="../env.d.ts" />
import { tool } from "@opencode-ai/plugin"
import { mkdir } from "node:fs/promises"
import { homedir } from "node:os"
import path from "node:path"

const DEFAULT_AGENTPAAS_URL = "http://127.0.0.1:8000"

type SSEEvent = { event: string; data: unknown }
type AgentListItem = { id: string; name: string }
type ConversationState = { threadID: string; runID: string }
type ProgressStatus = "completed" | "running" | "pending" | "failed"
type ProgressMilestone = {
  id: string
  label: string
  status: ProgressStatus
  done?: Set<string>
  total?: number
  detail?: string
}
type ProgressStage = {
  id: string
  label: string
  status: ProgressStatus
  milestones: ProgressMilestone[]
}

export default tool({
  description: `把任务转发给远端 AgentPaaS 智能体执行，等待完成后返回最终答案。

参数说明：
- agent_name：远端 AgentPaaS 智能体名称，工具会动态解析当前 ID，必填
- input：要执行的用户任务指令
- mode：iterate（多轮迭代，默认）/ chat（单轮）/ edit（单轮并指定目标子智能体）
- target_subagent：mode=edit 时的目标子智能体 ID
- work_dir：可选的服务端可见工作目录；留空时由 AgentPaaS 创建隔离 workspace
- conversation：continue（默认，复用本 OpenCode session 的远端 thread/workspace）/ new（新课题）
- thread_id：可选的远端会话线程 ID；显式值优先于自动映射
- parameters：额外参数，原样传给远端智能体

适用于调用远端科研流程智能体（如 research-67 系列）的场景。`,
  args: {
    input: tool.schema.string().describe("用户任务指令"),
    agent_name: tool.schema.string().describe("远端 AgentPaaS 智能体名称"),
    mode: tool.schema.enum(["iterate", "chat", "edit"]).default("iterate"),
    target_subagent: tool.schema.string().optional().describe("mode=edit 时的目标子智能 ID"),
    work_dir: tool.schema.string().optional().describe("服务端可见工作目录；通常留空，由 AgentPaaS 创建"),
    thread_id: tool.schema.string().optional().describe("远端会话线程 ID；通常留空，由 OpenCode session 自动维护"),
    conversation: tool.schema.enum(["continue", "new"]).default("continue").describe("复用当前对话或开始新课题"),
    parameters: tool.schema
      .record(tool.schema.string(), tool.schema.any())
      .optional()
      .describe("额外参数，原样传给远端智能体"),
  },
  async execute(args, context) {
    const config = await loadAgentPaaSConfig()
    if (args.mode === "edit" && !args.target_subagent) {
      throw new Error("mode=edit 时必须提供 target_subagent")
    }
    const agentID = await resolveAgentID(config, args.agent_name, context.abort)
    const conversation = args.conversation ?? "continue"
    const conversationKey = `${config.url}\n${agentID}`
    const savedConversation =
      conversation === "new" ? undefined : await loadConversationState(context.sessionID, conversationKey)
    const threadID = args.thread_id?.trim() || savedConversation?.threadID || ""
    const continueRunID =
      conversation === "continue" && (!args.thread_id?.trim() || args.thread_id.trim() === savedConversation?.threadID)
        ? savedConversation?.runID || ""
        : ""
    // OpenCode 的宿主机目录不会自动挂载进 AgentPaaS Docker。只有调用方
    // 明确给出服务端可见路径时才覆盖，否则让 PaaS 创建并隔离 run workspace。
    const workDir = args.work_dir
    const url = `${config.url}/api/v1/agents/${encodeURIComponent(agentID)}/run/stream`
    let resp: Response
    try {
      resp = await fetch(url, {
        method: "POST",
        headers: {
          Authorization: `Bearer ${config.apiKey}`,
          "Content-Type": "application/json",
          Accept: "text/event-stream",
        },
        body: JSON.stringify({
          input: args.input,
          mode: args.mode,
          target_subagent: args.target_subagent ?? "",
          thread_id: threadID,
          parameters: args.parameters ?? {},
          context: {
            ...(workDir ? { work_dir: workDir } : {}),
            ...(continueRunID ? { run_id: continueRunID } : {}),
          },
        }),
        signal: context.abort,
      })
    } catch (err) {
      throw new Error(`AgentPaaS 请求失败: ${err instanceof Error ? err.message : String(err)}`)
    }
    if (!resp.ok) {
      const detail = await resp.text().catch(() => "")
      throw new Error(`AgentPaaS HTTP ${resp.status}: ${detail.slice(0, 500)}`)
    }
    if (!resp.body) throw new Error("AgentPaaS 响应没有 body")

    const errors: string[] = []
    const liveMetadata: Record<string, unknown> = {
      agent_name: args.agent_name,
      agent_id: agentID,
      status: "running",
    }
    let done: Record<string, unknown> | null = null
    let startedRunID = ""
    let sawCancelled = false
    let lastThinkUpdate = 0
    const progress = createResearch67Progress()
    const updateMetadata = (title: string, next: Record<string, unknown>) => {
      Object.assign(liveMetadata, next)
      const snapshot = progress.snapshot()
      context.metadata({
        title: snapshot?.title ?? title,
        metadata: { ...liveMetadata, ...(snapshot ? { research67_progress: snapshot } : {}) },
      })
    }
    let remoteCancellation: Promise<void> | undefined
    const cancelOnAbort = () => {
      if (!startedRunID || remoteCancellation) return
      remoteCancellation = cancelRemoteRun(config, agentID, startedRunID).catch(() => undefined)
    }
    context.abort.addEventListener("abort", cancelOnAbort, { once: true })
    for await (const ev of parseSSE(resp.body)) {
      const data = asRecord(ev.data)
      if (ev.event === "started") {
        startedRunID = getString(data?.run_id)
        const remoteThreadID = getString(data?.thread_id)
        if (remoteThreadID && startedRunID) {
          await saveConversationState(context.sessionID, conversationKey, {
            threadID: remoteThreadID,
            runID: startedRunID,
          })
        }
        if (context.abort.aborted) cancelOnAbort()
        updateMetadata("Research67 正在运行", {
          run_id: data?.run_id,
          thread_id: data?.thread_id,
          latest_event: ev.event,
        })
      } else if (ev.event === "stage_started") {
        const stageName = getString(data?.stage_name) || getString(data?.stage_id) || "未知阶段"
        progress.startStage(getString(data?.stage_id), getPositiveInteger(data?.attempt))
        updateMetadata(`Research67 正在进行 ${stageName}`, {
          pipeline_id: data?.pipeline_id,
          stage_id: data?.stage_id,
          stage_name: stageName,
          stage_status: data?.status,
          attempt: data?.attempt,
          latest_event: ev.event,
        })
      } else if (ev.event === "stage_passed") {
        const stageName = getString(data?.stage_name) || getString(data?.stage_id) || "当前阶段"
        progress.finishStage(getString(data?.stage_id), "completed")
        updateMetadata(`Research67 已通过 ${stageName}`, {
          pipeline_id: data?.pipeline_id,
          stage_id: data?.stage_id,
          stage_name: stageName,
          stage_status: data?.status,
          attempt: data?.attempt,
          latest_event: ev.event,
        })
      } else if (ev.event === "stage_failed") {
        const stageName = getString(data?.stage_name) || getString(data?.stage_id) || "当前阶段"
        progress.finishStage(getString(data?.stage_id), "failed")
        updateMetadata(`Research67 阶段失败：${stageName}`, {
          pipeline_id: data?.pipeline_id,
          stage_id: data?.stage_id,
          stage_name: stageName,
          stage_status: data?.status,
          stage_error: data?.error,
          attempt: data?.attempt,
          latest_event: ev.event,
          status: "failed",
        })
      } else if (ev.event === "pipeline_completed") {
        progress.finishPipeline()
        updateMetadata("Research67 流水线已完成", {
          pipeline_id: data?.pipeline_id,
          pipeline_status: data?.status,
          latest_event: ev.event,
          status: "completed",
        })
      } else if (ev.event === "confirm_required") {
        const runID = getString(data?.run_id) || getString(liveMetadata.run_id)
        const toolName = getString(data?.tool) || "高风险工具"
        const reason = getString(data?.reason)
        const input = getString(data?.input)
        if (!runID) throw new Error("AgentPaaS confirm_required 事件缺少 run_id")

        progress.note(`等待确认 · ${toolName}`)
        updateMetadata(`Research67 正在等待确认：${toolName}`, {
          latest_event: ev.event,
          pending_confirmation: true,
          confirmation_tool: toolName,
          confirmation_reason: reason,
          confirmation_input: input,
        })
        let approved = true
        try {
          await context.ask({
            permission: "agentpaas_confirm",
            patterns: [`${agentID}:${toolName}`],
            always: [`${agentID}:${toolName}`],
            metadata: { run_id: runID, agent_name: args.agent_name, tool: toolName, reason, input },
          })
        } catch {
          approved = false
        }
        await resolveConfirmation(config, runID, approved, context.abort)
        progress.note(approved ? "确认已通过，继续执行" : "确认被拒绝，等待流程处理")
        updateMetadata(`Research67 已${approved ? "批准" : "拒绝"} ${toolName}`, {
          latest_event: "confirmation_resolved",
          pending_confirmation: false,
          confirmation_approved: approved,
        })
      } else if (ev.event === "tool_call") {
        const toolName = getString(data?.tool) || "工具"
        progress.toolCall(toolName, getString(data?.content))
        updateMetadata(withStage("Research67 正在处理当前步骤", liveMetadata), {
          latest_event: ev.event,
          latest_tool: toolName,
          step: data?.step,
        })
      } else if (ev.event === "tool_result") {
        const toolName = getString(data?.tool) || getString(liveMetadata.latest_tool) || "工具"
        progress.toolResult(toolName, getString(data?.content))
        updateMetadata(withStage("Research67 已更新当前步骤", liveMetadata), {
          latest_event: ev.event,
          latest_tool: toolName,
          step: data?.step,
        })
      } else if (ev.event === "think" || ev.event === "think_chunk") {
        const now = Date.now()
        if (ev.event === "think" || now - lastThinkUpdate >= 1000) {
          lastThinkUpdate = now
          updateMetadata(withStage("Research67 正在思考", liveMetadata), {
            latest_event: ev.event,
            step: data?.step,
          })
        }
      } else if (ev.event === "answer") {
        updateMetadata(withStage("Research67 正在整理结果", liveMetadata), {
          latest_event: ev.event,
          step: data?.step,
        })
      } else if (ev.event === "heartbeat") {
        updateMetadata(withStage("Research67 仍在运行", liveMetadata), { latest_event: ev.event })
      } else if (ev.event === "error") {
        const message = extractMessage(ev.data)
        errors.push(message)
        updateMetadata("Research67 执行失败", {
          latest_event: ev.event,
          status: "failed",
          error_code: data?.code,
          error_message: message,
          workspace_path: data?.workspace_path ?? liveMetadata.workspace_path,
        })
      } else if (ev.event === "cancelled") {
        sawCancelled = true
        progress.cancel()
        updateMetadata("Research67 已取消", { latest_event: ev.event, status: "cancelled" })
      } else if (ev.event === "done" && data) {
        done = data
        break
      }
    }
    context.abort.removeEventListener("abort", cancelOnAbort)
    if (!done) {
      const hint = sawCancelled ? "（收到 cancelled 事件）" : "（未收到 done 事件，连接可能中断）"
      throw new Error(`AgentPaaS 任务未正常结束${hint}${errors.length ? "：" + errors.join("; ") : ""}`)
    }
    const status = typeof done.status === "string" ? done.status : "unknown"
    const progressMetadata = progress.snapshot()
    const metadata = {
      agent_name: args.agent_name,
      agent_id: agentID,
      run_id: done.run_id ?? liveMetadata.run_id,
      thread_id: done.thread_id ?? liveMetadata.thread_id,
      status,
      steps: done.steps,
      total_tokens: done.total_tokens,
      cost_usd: done.cost_usd,
      workspace_path: done.workspace_path ?? liveMetadata.workspace_path,
      error_code: liveMetadata.error_code,
      error_message: liveMetadata.error_message,
      ...(progressMetadata ? { research67_progress: progressMetadata } : {}),
    }
    if (status === "completed")
      return { title: `AgentPaaS ${args.agent_name}`, output: String(done.output ?? ""), metadata }
    if (status === "cancelled") return { title: `AgentPaaS ${args.agent_name}`, output: "任务已被取消。", metadata }
    return {
      title: `AgentPaaS ${args.agent_name}（失败）`,
      output: [
        `[AgentPaaS 执行失败] ${errors.join("; ") || String(done.output ?? "") || "未知错误"}`,
        metadata.workspace_path
          ? `已有产物保留在：${metadata.workspace_path}`
          : "已有产物已保留；请根据 run_id 查询工作区。",
        "失败严禁自动重跑、拆分任务或调用其他 subagent；必须先向用户报告真实错误和已有产物，等待用户明确指令。",
      ].join("\n"),
      metadata,
    }
  },
})

async function resolveAgentID(config: { url: string; apiKey: string }, agentName: string, signal: AbortSignal) {
  const name = agentName.trim()
  if (!name) throw new Error("agent_name 不能为空")

  let resp: Response
  try {
    resp = await fetch(`${config.url}/api/v1/agents`, {
      headers: { Authorization: `Bearer ${config.apiKey}`, Accept: "application/json" },
      signal,
    })
  } catch (err) {
    throw new Error(`获取 AgentPaaS Agent 列表失败: ${err instanceof Error ? err.message : String(err)}`)
  }
  if (!resp.ok) {
    const detail = await resp.text().catch(() => "")
    throw new Error(`获取 AgentPaaS Agent 列表失败: HTTP ${resp.status}: ${detail.slice(0, 500)}`)
  }

  const payload: unknown = await resp.json().catch(() => null)
  if (!payload || typeof payload !== "object" || !("agents" in payload) || !Array.isArray(payload.agents)) {
    throw new Error("AgentPaaS Agent 列表响应格式无效")
  }
  const agents = (payload.agents as unknown[]).flatMap((item): AgentListItem[] => {
    if (!item || typeof item !== "object" || !("id" in item) || !("name" in item)) return []
    if (typeof item.id !== "string" || typeof item.name !== "string") return []
    return [{ id: item.id, name: item.name }]
  })
  const matches = agents.filter((item) => item.name === name)
  if (matches.length === 0) throw new Error(`AgentPaaS 中没有名为“${name}”的 active Agent`)
  if (matches.length > 1)
    throw new Error(`AgentPaaS 中存在 ${matches.length} 个名为“${name}”的 active Agent，无法确定调用目标`)
  if (!matches[0].id.startsWith("ag_")) throw new Error(`AgentPaaS 为“${name}”返回了无效 ID: ${matches[0].id}`)
  return matches[0].id
}

const conversationCache = new Map<string, ConversationState>()

async function loadConversationState(sessionID: string, key: string): Promise<ConversationState | undefined> {
  const cacheKey = `${sessionID}\n${key}`
  const cached = conversationCache.get(cacheKey)
  if (cached) return cached

  const file = Bun.file(conversationStatePath(sessionID))
  if (!(await file.exists())) return undefined
  try {
    const parsed: unknown = await file.json()
    if (!parsed || typeof parsed !== "object" || Array.isArray(parsed)) return undefined
    const value = (parsed as Record<string, unknown>)[key]
    if (!value || typeof value !== "object" || Array.isArray(value)) return undefined
    const threadID = getString((value as Record<string, unknown>).thread_id)
    const runID = getString((value as Record<string, unknown>).run_id)
    if (!threadID || !runID) return undefined
    const state = { threadID, runID }
    conversationCache.set(cacheKey, state)
    return state
  } catch {
    return undefined
  }
}

async function saveConversationState(sessionID: string, key: string, state: ConversationState) {
  conversationCache.set(`${sessionID}\n${key}`, state)
  const filePath = conversationStatePath(sessionID)
  const file = Bun.file(filePath)
  let saved: Record<string, unknown> = {}
  if (await file.exists()) {
    try {
      const parsed: unknown = await file.json()
      if (parsed && typeof parsed === "object" && !Array.isArray(parsed)) saved = parsed as Record<string, unknown>
    } catch {
      saved = {}
    }
  }
  saved[key] = { thread_id: state.threadID, run_id: state.runID }
  try {
    await mkdir(agentPaaSConversationDir(), { recursive: true, mode: 0o700 })
    await Bun.write(filePath, `${JSON.stringify(saved, null, 2)}\n`)
  } catch {
    // The in-memory mapping still preserves continuity for this CLI process.
  }
}

function conversationStatePath(sessionID: string) {
  return path.join(agentPaaSConversationDir(), `${sessionID.replace(/[^A-Za-z0-9._-]/g, "_")}.json`)
}

async function loadAgentPaaSConfig() {
  const configPath = agentPaaSConfigPath()
  const file = Bun.file(configPath)
  let saved: Record<string, unknown> = {}
  if (await file.exists()) {
    try {
      const parsed: unknown = await file.json()
      if (!parsed || typeof parsed !== "object" || Array.isArray(parsed)) throw new Error("根节点必须是 JSON 对象")
      saved = parsed as Record<string, unknown>
    } catch (err) {
      throw new Error(`AgentPaaS 配置文件无效：${configPath}\n${err instanceof Error ? err.message : String(err)}`)
    }
  }

  const savedUrl = typeof saved.server === "string" ? saved.server.trim() : ""
  const savedApiKey = typeof saved.api_key === "string" ? saved.api_key.trim() : ""
  const url = (process.env.AGENTPAAS_URL?.trim() || savedUrl || DEFAULT_AGENTPAAS_URL).replace(/\/+$/, "")
  const apiKey = process.env.AGENTPAAS_API_KEY?.trim() || savedApiKey
  if (apiKey) return { url, apiKey }

  throw new Error(`AgentPaaS 尚未配置。请创建 ${configPath}：
{
  "server": "${DEFAULT_AGENTPAAS_URL}",
  "api_key": "你的 AgentPaaS API Key"
}
也可以通过 AGENTPAAS_URL 和 AGENTPAAS_API_KEY 环境变量临时覆盖。`)
}

function agentPaaSHome() {
  return process.env.HOME?.trim() || homedir()
}

function agentPaaSConfigPath() {
  return path.join(agentPaaSHome(), ".agentpaas", "config.json")
}

function agentPaaSConversationDir() {
  return path.join(agentPaaSHome(), ".agentpaas", "opencode-conversations")
}

async function resolveConfirmation(
  config: { url: string; apiKey: string },
  runID: string,
  approved: boolean,
  signal: AbortSignal,
) {
  const resp = await fetch(`${config.url}/api/v1/traces/${encodeURIComponent(runID)}/confirm`, {
    method: "POST",
    headers: {
      Authorization: `Bearer ${config.apiKey}`,
      "Content-Type": "application/json",
    },
    body: JSON.stringify({ approved }),
    signal,
  })
  if (resp.ok) return
  const detail = await resp.text().catch(() => "")
  throw new Error(`AgentPaaS 确认请求失败: HTTP ${resp.status}: ${detail.slice(0, 500)}`)
}

async function cancelRemoteRun(config: { url: string; apiKey: string }, agentID: string, runID: string) {
  const signal = AbortSignal.timeout(5000)
  const resp = await fetch(
    `${config.url}/api/v1/agents/${encodeURIComponent(agentID)}/runs/${encodeURIComponent(runID)}/cancel`,
    {
      method: "POST",
      headers: { Authorization: `Bearer ${config.apiKey}`, Accept: "application/json" },
      signal,
    },
  )
  if (resp.ok || resp.status === 409) return
  const detail = await resp.text().catch(() => "")
  throw new Error(`AgentPaaS 取消请求失败: HTTP ${resp.status}: ${detail.slice(0, 500)}`)
}

async function* parseSSE(body: ReadableStream<Uint8Array>): AsyncGenerator<SSEEvent> {
  const reader = body.getReader()
  const decoder = new TextDecoder()
  let buffer = ""
  while (true) {
    const { done, value } = await reader.read()
    if (done) break
    buffer += decoder.decode(value, { stream: true })
    buffer = buffer.replace(/\r\n/g, "\n")
    let idx: number
    while ((idx = buffer.indexOf("\n\n")) >= 0) {
      const chunk = buffer.slice(0, idx)
      buffer = buffer.slice(idx + 2)
      const event = parseSSEField(chunk, "event")
      const dataRaw = parseSSEField(chunk, "data")
      if (!event && chunk.split("\n").some((line) => line.trim() === ": heartbeat")) {
        yield { event: "heartbeat", data: undefined }
        continue
      }
      let data: unknown = dataRaw
      if (dataRaw !== "") {
        try {
          data = JSON.parse(dataRaw)
        } catch {
          /* 非 JSON 当纯文本 */
        }
      }
      if (event) yield { event, data }
    }
  }
}
function parseSSEField(chunk: string, field: string): string {
  const lines = chunk.split("\n")
  const parts: string[] = []
  for (const line of lines) {
    if (line.startsWith(":")) continue
    const m = line.match(new RegExp(`^${field}:\\s?(.*)$`))
    if (m) parts.push(m[1])
  }
  return parts.join("\n")
}
function extractMessage(data: unknown): string {
  if (data && typeof data === "object" && "message" in data) return String((data as Record<string, unknown>).message)
  return String(data)
}

function asRecord(value: unknown): Record<string, unknown> | undefined {
  if (!value || typeof value !== "object" || Array.isArray(value)) return undefined
  const result: Record<string, unknown> = {}
  for (const [key, item] of Object.entries(value)) result[key] = item
  return result
}

function getString(value: unknown): string {
  return typeof value === "string" ? value : ""
}

function getPositiveInteger(value: unknown): number {
  return typeof value === "number" && Number.isInteger(value) && value > 0 ? value : 1
}

function getWriteFilePath(content: string): string {
  try {
    const filePath = getString(asRecord(JSON.parse(content))?.file_path)
    if (filePath) return filePath
  } catch {}
  return content.match(/["']file_path["']\s*:\s*(["'])(.*?)\1/)?.[2] ?? ""
}

function withStage(title: string, metadata: Record<string, unknown>): string {
  const stageName = getString(metadata.stage_name)
  return stageName ? `${title}（${stageName}）` : title
}

function createResearch67Progress() {
  const stages: ProgressStage[] = [
    {
      id: "ontology-seed",
      label: "本体构建 · seed ontology",
      status: "pending",
      milestones: [
        { id: "parse", label: "解析问题与学科路由", status: "pending", done: new Set(), total: 1 },
        { id: "seed", label: "生成基础本体", status: "pending", done: new Set(), total: 1 },
      ],
    },
    {
      id: "ontology-audit",
      label: "ontology-auditor · 本体审计",
      status: "pending",
      milestones: [
        { id: "schema", label: "检查 schema 和关系", status: "pending", done: new Set(), total: 1 },
        { id: "gaps", label: "识别证据与机制缺口", status: "pending", done: new Set(), total: 1 },
      ],
    },
    {
      id: "ontology-enrichment",
      label: "ontology-enricher · 知识补全",
      status: "pending",
      milestones: [
        { id: "kb", label: "召回本地知识库与教材锚点", status: "pending", done: new Set(), total: 1 },
        { id: "search", label: "检索与来源核验", status: "pending", done: new Set(), detail: "尚未开始" },
        { id: "patch", label: "生成本体补丁", status: "pending", done: new Set(), total: 1 },
      ],
    },
    {
      id: "hypothesis-build",
      label: "hypothesis-builder · 假设构建",
      status: "pending",
      milestones: [
        { id: "hypothesis", label: "生成可证伪假设", status: "pending", done: new Set(), total: 1 },
        { id: "falsifiers", label: "生成可证伪条件", status: "pending", done: new Set(), total: 1 },
      ],
    },
    {
      id: "research-plan",
      label: "plan-builder · 研究计划",
      status: "pending",
      milestones: [
        { id: "plan", label: "形成研究计划本体", status: "pending", done: new Set(), total: 1 },
        { id: "dag", label: "编译 Action DAG", status: "pending", done: new Set(), total: 1 },
      ],
    },
    {
      id: "evaluation",
      label: "action-reasoner · 证据评估",
      status: "pending",
      milestones: [
        { id: "assess", label: "评估假设支持状态", status: "pending", done: new Set(), total: 1 },
        { id: "synthesis", label: "形成边界化结论", status: "pending", done: new Set(), total: 1 },
      ],
    },
    {
      id: "experiment-design",
      label: "exp-planner · 实验规划 / 授权包",
      status: "pending",
      milestones: [
        { id: "protocol", label: "生成实验协议", status: "pending", done: new Set(), total: 1 },
        { id: "authorization", label: "生成授权与数据回传要求", status: "pending", done: new Set(), total: 1 },
      ],
    },
    {
      id: "experiment-execution",
      label: "exp-executor · 授权实验执行",
      status: "pending",
      milestones: [
        { id: "auth", label: "确认授权与执行边界", status: "pending", done: new Set(), total: 1 },
        { id: "run", label: "运行计算/数据实验", status: "pending", done: new Set(), total: 1 },
        { id: "receipt", label: "保存执行日志与结果", status: "pending", done: new Set(), total: 1 },
      ],
    },
    {
      id: "result-analysis",
      label: "result-analyzer · 结果分析",
      status: "pending",
      milestones: [
        { id: "analysis", label: "分析实验或回传数据", status: "pending", done: new Set(), total: 1 },
        { id: "patch", label: "生成 observation patch", status: "pending", done: new Set(), total: 1 },
      ],
    },
    {
      id: "paper-writing",
      label: "paper-writer · 论文写作",
      status: "pending",
      milestones: [
        { id: "outline", label: "写作计划与大纲", status: "pending", done: new Set(), total: 2 },
        { id: "english", label: "英文稿与参考文献", status: "pending", done: new Set(), total: 8 },
        { id: "chinese", label: "中文稿", status: "pending", done: new Set(), total: 7 },
        { id: "compile", label: "编译双语 PDF", status: "pending", done: new Set(), total: 2 },
        { id: "report", label: "形成交付报告", status: "pending", done: new Set(), total: 2 },
      ],
    },
    {
      id: "data-verification",
      label: "data-verifier · 数据与主张核验",
      status: "pending",
      milestones: [
        { id: "claims", label: "核验论文主张与来源", status: "pending", done: new Set(), total: 1 },
        { id: "numbers", label: "核验数据和表述一致性", status: "pending", done: new Set(), total: 1 },
      ],
    },
    {
      id: "paper-review",
      label: "paper-reviewer · 论文评审",
      status: "pending",
      milestones: [
        { id: "rubric", label: "按科学性和可复现性评分", status: "pending", done: new Set(), total: 1 },
        { id: "gaps", label: "输出研究缺口", status: "pending", done: new Set(), total: 1 },
      ],
    },
    {
      id: "review-feedback",
      label: "review-feedback · 反馈迭代",
      status: "pending",
      milestones: [
        { id: "compile", label: "编译评审反馈", status: "pending", done: new Set(), total: 1 },
        { id: "next", label: "生成下一轮 action 与本体补丁", status: "pending", done: new Set(), total: 1 },
      ],
    },
    {
      id: "ontology-reaudit",
      label: "ontology-auditor · 本体再审计",
      status: "pending",
      milestones: [
        { id: "audit", label: "检查反馈后的本体", status: "pending", done: new Set(), total: 1 },
        { id: "decision", label: "决定完成或进入下一轮", status: "pending", done: new Set(), total: 1 },
      ],
    },
    // 兼容旧三阶段 Lite pipeline 事件。
    {
      id: "idea-analysis",
      label: "idea-analyst · 创意分析",
      status: "pending",
      milestones: [
        { id: "plan", label: "制定分析计划", status: "pending", done: new Set(), total: 1 },
        { id: "directions", label: "生成检索方向", status: "pending", done: new Set(), total: 2 },
        { id: "report", label: "形成分析报告", status: "pending", done: new Set(), total: 2 },
      ],
    },
    {
      id: "literature-review",
      label: "lit-searcher · 文献检索",
      status: "pending",
      milestones: [
        { id: "plan", label: "制定检索计划", status: "pending", done: new Set(), total: 1 },
        { id: "search", label: "检索候选文献", status: "pending", done: new Set(), detail: "尚未开始" },
        { id: "select", label: "筛选入选论文", status: "pending", done: new Set(), total: 1 },
        { id: "report", label: "形成文献综述", status: "pending", done: new Set(), total: 2 },
      ],
    },
  ]
  let current: ProgressStage | undefined
  let active: ProgressMilestone | undefined
  const pendingArtifacts = new Map<string, Array<Array<{ milestone: ProgressMilestone; key: string }>>>()
  let searchCalls = 0
  let note = ""
  let activeSince = Date.now()
  let activeAttempt = 0
  let timerRunning = false

  const aliases: Record<string, string> = {
    ontology_seed: "ontology-seed",
    "seed-ontology": "ontology-seed",
    "audit-ontology": "ontology-audit",
    audit_ontology: "ontology-reaudit",
    ontology_audit: "ontology-audit",
    "ontology-enrich": "ontology-enrichment",
    ontology_enrichment: "ontology-enrichment",
    "enrich-evidence": "ontology-enrichment",
    enrich_evidence: "ontology-enrichment",
    search: "ontology-enrichment",
    "build-hypothesis": "hypothesis-build",
    build_hypothesis: "hypothesis-build",
    "hypothesis-builder": "hypothesis-build",
    "plan-build": "research-plan",
    "build-research-plan": "research-plan",
    build_research_plan: "research-plan",
    "action-dag": "research-plan",
    synthesis: "evaluation",
    "falsification-assessment": "evaluation",
    falsification_assessment: "evaluation",
    "experiment-plan": "experiment-design",
    experiment_design: "experiment-design",
    experiment_execution: "experiment-execution",
    result_analysis: "result-analysis",
    paper_writing: "paper-writing",
    data_verification: "data-verification",
    paper_review: "paper-review",
    review_feedback: "review-feedback",
  }

  const genericMilestones = (): ProgressMilestone[] => [
    { id: "run", label: "远端阶段执行中", status: "pending", done: new Set(), total: 1 },
    { id: "receipt", label: "等待 receipt / artifact", status: "pending", done: new Set(), total: 1 },
  ]

  const normalizeStageID = (stageID: string) => aliases[stageID] ?? aliases[stageID.replaceAll("_", "-")] ?? stageID

  const findOrCreateStage = (stageID: string) => {
    const normalized = normalizeStageID(stageID || "remote-stage")
    let stage = stages.find((item) => item.id === normalized)
    if (stage) return stage
    stage = {
      id: normalized,
      label: `${stageID || "remote-stage"} · 远端阶段`,
      status: "pending",
      milestones: genericMilestones(),
    }
    stages.push(stage)
    return stage
  }

  const startMilestone = (milestone: ProgressMilestone) => {
    if (!current) return
    if (milestone.status === "completed") return
    milestone.status = "running"
    if (milestone.detail === "尚未开始") milestone.detail = undefined
    if (active !== milestone) activeSince = Date.now()
    active = milestone
    timerRunning = true
    note = ""
  }

  const resetStage = (stage: ProgressStage) => {
    for (const milestone of stage.milestones) {
      milestone.status = "pending"
      milestone.done?.clear()
      milestone.detail = milestone.id === "search" ? "尚未开始" : undefined
    }
    if (["literature-review", "ontology-enrichment"].includes(stage.id)) searchCalls = 0
    pendingArtifacts.clear()
  }

  const artifact = (filePath: string) => {
    if (!current) return undefined
    const value = filePath.replaceAll("\\", "/").toLowerCase()
    const stage = value
      .split("/")
      .findLastIndex((part) => part === current?.id || new RegExp(`^${current?.id}_a\\d+$`).test(part))
    if (stage < 0) return undefined
    const relative = value
      .split("/")
      .slice(stage + 1)
      .join("/")
    const match = (milestoneID: string, key: string) => {
      const milestone = current?.milestones.find((item) => item.id === milestoneID)
      return milestone ? { milestone, key } : undefined
    }
    if (relative === "work_plan.md") {
      return match(current.id === "paper-writing" ? "outline" : "plan", "work_plan")
    }
    if (current.id === "idea-analysis") {
      if (relative === "artifacts/search_queries.json") return match("directions", "search_queries")
      if (relative === "artifacts/similar_papers.json") return match("directions", "similar_papers")
      if (relative === "report.json") return match("report", "report_json")
      if (relative === "report.md") return match("report", "report_md")
    }
    if (["literature-review", "ontology-enrichment"].includes(current.id)) {
      if (relative === "artifacts/papers_selected.json") return match("select", "papers_selected")
      if (relative === "report.json") return match("report", "report_json")
      if (relative === "report.md") return match("report", "report_md")
      if (relative.endsWith("search-log.jsonl")) return match("search", "search_log")
    }
    if (current.id === "experiment-design") {
      if (relative.endsWith("experiment_protocol.json")) return match("protocol", "experiment_protocol")
      if (relative.endsWith("external_authorization_package.json"))
        return match("authorization", "external_authorization_package")
    }
    if (current.id !== "paper-writing") return undefined
    if (relative === "artifacts/outline.json") return match("outline", "outline")
    const section = relative.match(
      /^artifacts\/sections\/(abstract|introduction|related_work|method|experiments|conclusion)\.tex$/,
    )
    if (section) return match("english", `section:${section[1]}`)
    if (relative === "artifacts/paper.tex") return match("english", "paper_tex")
    if (relative === "artifacts/references.bib") return match("english", "references")
    const sectionZh = relative.match(
      /^artifacts\/sections_zh\/(abstract|introduction|related_work|method|experiments|conclusion)\.tex$/,
    )
    if (sectionZh) return match("chinese", `section:${sectionZh[1]}`)
    if (relative === "artifacts/paper_zh.tex") return match("chinese", "paper_zh_tex")
    if (relative === "report.json") return match("report", "report_json")
    if (relative === "report.md") return match("report", "report_md")
    return undefined
  }

  const invalidatePaperBuild = (items: Array<{ milestone: ProgressMilestone; key: string }>) => {
    if (current?.id !== "paper-writing") return
    if (!items.some((item) => ["english", "chinese"].includes(item.milestone.id))) return
    const compile = current.milestones.find((item) => item.id === "compile")!
    const report = current.milestones.find((item) => item.id === "report")!
    if (compile.status === "pending" && !compile.done?.size && report.status === "pending") return
    compile.status = "pending"
    compile.done?.clear()
    report.status = "pending"
    report.done?.clear()
    if (active === compile || active === report) active = undefined
    activeSince = Date.now()
    timerRunning = true
    note = "论文源文件已更新，等待重新编译"
  }

  const completeArtifact = (item: { milestone: ProgressMilestone; key: string }) => {
    item.milestone.done?.add(item.key)
    if (!item.milestone.total || item.milestone.done?.size < item.milestone.total) return
    item.milestone.status = "completed"
    if (active !== item.milestone || !current) return
    const next = current.milestones[current.milestones.indexOf(item.milestone) + 1]
    if (next) return startMilestone(next)
    active = undefined
    activeSince = Date.now()
    timerRunning = true
    note = "正在进行 Guard / 阶段产物验收"
  }

  const enqueueArtifacts = (toolName: string, items: Array<{ milestone: ProgressMilestone; key: string }>) => {
    const queue = pendingArtifacts.get(toolName) ?? []
    queue.push(items)
    pendingArtifacts.set(toolName, queue)
  }

  const dequeueArtifacts = (toolName: string) => {
    const queue = pendingArtifacts.get(toolName)
    const items = queue?.shift()
    if (!queue?.length) pendingArtifacts.delete(toolName)
    return items
  }

  return {
    startStage(stageID: string, attempt: number) {
      const stage = findOrCreateStage(stageID)
      if (stage === current && attempt <= activeAttempt) return
      if (stage === current) resetStage(stage)
      current = stage
      activeAttempt = attempt
      for (const item of stages) {
        if (item === current) break
        if (item.status === "pending") item.status = "completed"
      }
      current.status = "running"
      startMilestone(current.milestones[0])
    },
    finishStage(stageID: string, status: "completed" | "failed") {
      const stage = findOrCreateStage(stageID)
      stage.status = status
      if (status === "completed") {
        for (const milestone of stage.milestones) milestone.status = "completed"
      } else {
        for (const milestone of stage.milestones) {
          if (milestone.status === "running") milestone.status = "failed"
        }
      }
      current = stage
      active = status === "failed" ? active : undefined
      timerRunning = false
      note = status === "failed" ? "阶段执行失败" : "阶段验收通过"
    },
    finishPipeline() {
      for (const stage of stages) {
        stage.status = "completed"
        for (const milestone of stage.milestones) milestone.status = "completed"
      }
      active = undefined
      timerRunning = false
      note = "全部阶段已完成"
    },
    cancel() {
      if (current?.status === "running") current.status = "failed"
      if (active?.status === "running") active.status = "failed"
      timerRunning = false
      note = "任务已取消"
    },
    note(value: string) {
      note = value
    },
    toolCall(toolName: string, content: string) {
      const lower = toolName.toLowerCase()
      if (
        current?.id === "ontology-enrichment" &&
        ["kbsearch", "websearch", "webfetch", "openalex_search", "crossref_search"].includes(lower)
      ) {
        const milestone = current.milestones.find((item) => item.id === "search")!
        startMilestone(milestone)
        return
      }
      if (current?.id === "literature-review" && ["arxiv_search", "openalex_search"].includes(lower)) {
        const milestone = current.milestones.find((item) => item.id === "search")!
        startMilestone(milestone)
        return
      }
      if (current?.id === "paper-writing" && lower === "compilelatex") {
        const milestone = current.milestones.find((item) => item.id === "compile")!
        const key = content.toLowerCase().includes("paper_zh.tex") ? "paper_zh_pdf" : "paper_pdf"
        milestone.done?.delete(key)
        if (milestone.status === "completed") milestone.status = "pending"
        startMilestone(milestone)
        enqueueArtifacts(lower, [{ milestone, key }])
        return
      }
      if (current?.id === "paper-writing" && lower === "buildresearchpaper") {
        const milestone = current.milestones.find((item) => item.id === "compile")!
        milestone.status = "pending"
        milestone.done?.clear()
        startMilestone(milestone)
        enqueueArtifacts(lower, [
          { milestone, key: "paper_pdf" },
          { milestone, key: "paper_zh_pdf" },
        ])
        return
      }
      if (lower !== "writefile") return
      const item = artifact(getWriteFilePath(content))
      if (!item) {
        enqueueArtifacts(lower, [])
        return
      }
      if (
        ["literature-review", "ontology-enrichment"].includes(current?.id ?? "") &&
        item.milestone.id === "select" &&
        searchCalls
      ) {
        const searchMilestone = current?.milestones.find((milestone) => milestone.id === "search")
        if (searchMilestone) searchMilestone.status = "completed"
      }
      startMilestone(item.milestone)
      enqueueArtifacts(lower, [item])
    },
    toolResult(toolName: string, content: string) {
      const lower = toolName.toLowerCase()
      const failed = /\[(?:tool_)?error\]|mcp_error|"status"\s*:\s*"(?:failed|error)"|"passed"\s*:\s*false/i.test(
        content,
      )
      if (
        current?.id === "ontology-enrichment" &&
        ["kbsearch", "websearch", "webfetch", "openalex_search", "crossref_search"].includes(lower)
      ) {
        if (!failed) searchCalls++
        const milestone = current.milestones.find((item) => item.id === "search")!
        milestone.detail = searchCalls ? `已完成 ${searchCalls} 轮检索/核验` : "等待检索重试"
        return
      }
      if (current?.id === "literature-review" && ["arxiv_search", "openalex_search"].includes(lower)) {
        if (!failed) searchCalls++
        const milestone = current.milestones.find((item) => item.id === "search")!
        milestone.detail = searchCalls ? `已完成 ${searchCalls} 轮检索` : "等待检索重试"
        return
      }
      const items = dequeueArtifacts(lower)
      if (!items) return
      if (!failed) {
        items.forEach(completeArtifact)
        invalidatePaperBuild(items)
      }
      if (failed) {
        for (const item of items) item.milestone.status = "failed"
        note = "当前操作失败，等待重试"
      }
    },
    snapshot() {
      if (!current) return undefined
      const rows = stages.flatMap((stage) => {
        const detail = stage.status === "running" ? active?.label : undefined
        const parent = [{ id: stage.id, label: stage.label, status: stage.status, detail, depth: 0 }]
        if (stage !== current || !["running", "failed"].includes(stage.status)) return parent
        return parent.concat(
          stage.milestones.map((milestone) => ({
            id: `${stage.id}:${milestone.id}`,
            label: milestone.label,
            status: milestone.status,
            detail:
              milestone.detail ??
              (milestone.status !== "completed" && milestone.total && milestone.done?.size
                ? `${milestone.done.size}/${milestone.total}`
                : undefined),
            depth: 1,
          })),
        )
      })
      return {
        kind: "research67",
        title: active ? `Research67 · ${current.label} · ${active.label}` : `Research67 · ${current.label}`,
        rows,
        activeSince: timerRunning ? activeSince : undefined,
        footer: note || (!active ? "进度已更新" : undefined),
      }
    },
  }
}