opencode/.opencode/lib/agentpaas.ts

import { link, mkdir, readdir, rename, rm } from "node:fs/promises"
import { homedir } from "node:os"
import path from "node:path"
import { setTimeout } from "node:timers/promises"

const DEFAULT_AGENTPAAS_URL = "http://127.0.0.1:8000"
export type Connection = { url: string; apiKey: string }
export type ConversationState = { threadID: string; runID: string }
type AgentListItem = { id: string; name: string }
const revisions = new Map<string, string>()
const probes = new Map<
  string,
  { at: number; report: Record<string, unknown> & { blocked: boolean; summary: string } }
>()

export async function requestJSON(config: Connection, route: string, signal: AbortSignal, init: RequestInit = {}) {
  const response = await fetch(`${config.url}/api/v1${route}`, {
    ...init,
    headers: { Authorization: `Bearer ${config.apiKey}`, Accept: "application/json", ...init.headers },
    redirect: "error",
    signal: AbortSignal.any([signal, AbortSignal.timeout(30000)]),
  }).catch((error) => {
    if (signal.aborted) throw error
    throw new Error(
      `无法取得 PaaS 响应（${config.url}）；请检查服务、地址与网络。已有任务请保留原 run_id 查询，不要重复提交。${getString(asRecord(error)?.code)}`,
    )
  })
  if (!response.ok) {
    const detail = (await response.text()).replaceAll(config.apiKey, "[redacted]").slice(0, 500)
    throw new Error(`AgentPaaS HTTP ${response.status}: ${detail}`)
  }
  const result = asRecord(await response.json())
  if (!result) throw new Error("AgentPaaS 返回了无效 JSON 对象")
  return result
}

export async function verifyAgent(
  config: Connection,
  agentID: string,
  signal: AbortSignal,
  options = { check_model: false, knowledge_query: "" },
  cached = false,
): Promise<Record<string, unknown> & { blocked: boolean; summary: string }> {
  const scope = digest(canonicalJSON([config.url, config.apiKey, agentID]))
  const revision = revisions.get(scope)
  const key = `${scope}:${revision}`
  const useCache = cached && !options.check_model && !options.knowledge_query
  if (!useCache) probes.delete(key)
  const previous = probes.get(key)
  if (useCache && revision && previous && Date.now() - previous.at < 5000)
    return { ...previous.report, cache_hit: true }
  const report = await requestJSON(config, `/agents/${encodeURIComponent(agentID)}/verify`, signal, {
    method: "POST",
    headers: { "Content-Type": "application/json" },
    body: JSON.stringify(options),
  })
  if (
    !["verified", "unverified", "degraded", "blocked"].includes(getString(report.status)) ||
    !Array.isArray(report.checks)
  )
    throw new Error("PaaS 能力检查返回无效报告，任务未提交；请更新 PaaS 后重试检查。")
  const checks = report.checks.map(asRecord)
  if (
    checks.some(
      (item) =>
        !item ||
        typeof item.required !== "boolean" ||
        !["passed", "failed", "not_checked"].includes(getString(item.status)),
    )
  )
    throw new Error("PaaS 能力检查条目无效，任务未提交。")
  const blocked = report.status === "blocked" || checks.some((item) => item?.required && item.status === "failed")
  const lines = checks.flatMap((item) =>
    item && item.status !== "passed"
      ? [
          `${item.required ? "必需" : "可选"} ${getString(item.kind)}/${getString(item.name)}：${item.status === "failed" ? "失败" : "未验证"}${item.error_code ? ` (${getString(item.error_code)})` : ""}`,
        ]
      : [],
  )
  const result = {
    ...report,
    blocked,
    summary: [blocked ? "必需能力不可用，任务未提交。" : "未发现已确认失效的必需能力。", ...lines].join("\n"),
  }
  if (useCache && revision && !blocked) {
    if (probes.size > 100) probes.clear()
    probes.set(key, { at: Date.now(), report: result })
  }
  return result
}

type TrackedRun = { agent_id: string; agent_name: string; run_id: string; thread_id: string; recorded_at: string }

function trackedRunDirectory(config: Connection) {
  return path.join(agentPaaSConversationDir(), "runs", digest(canonicalJSON([config.url, config.apiKey])))
}

export async function trackRun(config: Connection, run: Omit<TrackedRun, "recorded_at">) {
  if (!isSupportedAgentID(run.agent_id) || !run.run_id) throw new Error("无法保存无效任务编号")
  const directory = trackedRunDirectory(config)
  await mkdir(directory, { recursive: true, mode: 0o700 })
  const target = path.join(directory, `${digest(canonicalJSON([run.agent_id, run.run_id]))}.json`)
  const temporary = `${target}.${crypto.randomUUID()}.tmp`
  try {
    await Bun.write(temporary, JSON.stringify({ ...run, recorded_at: new Date().toISOString() }), { mode: 0o600 })
    await rename(temporary, target)
  } finally {
    await rm(temporary, { force: true })
  }
}

export async function trackedRuns(config: Connection, limit: number) {
  const directory = trackedRunDirectory(config)
  const files = await readdir(directory).catch((error: NodeJS.ErrnoException) => {
    if (error.code === "ENOENT") return []
    throw error
  })
  const rows = await Promise.all(
    files
      .filter((file) => /^[a-f0-9]{64}\.json$/.test(file))
      .map(async (file) => {
        const row = asRecord(await Bun.file(path.join(directory, file)).json())
        if (
          !row ||
          !isSupportedAgentID(getString(row.agent_id)) ||
          !getString(row.run_id) ||
          !getString(row.recorded_at)
        )
          throw new Error("任务索引损坏，请用 agentpaas_task history 查询服务端历史。")
        return row as TrackedRun
      }),
  )
  return rows.sort((a, b) => b.recorded_at.localeCompare(a.recorded_at)).slice(0, limit)
}

export async function listAgents(config: Connection, signal: AbortSignal, wanted?: string) {
  for (let attempt = 0; ; attempt++) {
    const payload = await requestJSON(config, "/agents", signal)
    if (!Array.isArray(payload.agents)) throw new Error("AgentPaaS Agent 列表响应格式无效")
    const agents = payload.agents.flatMap((item): Array<Record<string, unknown> & AgentListItem> => {
      const record = asRecord(item)
      return record && typeof record.id === "string" && typeof record.name === "string"
        ? [{ ...record, id: record.id, name: record.name }]
        : []
    })
    const status = getString(asRecord(payload.discovery)?.status)
    const found = wanted && agents.some((agent) => agent.id === wanted || agent.name === wanted)
    if (found || !["pending", "refreshing"].includes(status) || attempt >= 7)
      return { agents, discovery: asRecord(payload.discovery) ?? { status: "unknown" } }
    // Only read-only discovery is retried. Never replay a control mutation here.
    await setTimeout(Math.min(250 * (attempt + 1), 1000), undefined, { signal })
  }
}

export async function resolveAgentID(config: Connection, selector: string, signal: AbortSignal) {
  const name = selector.trim()
  if (!name) throw new Error("必须提供 agent_name 或 agent_id")
  const catalog = await listAgents(config, signal, name)
  const byID = catalog.agents.filter((item) => item.id === name)
  const matches = byID.length ? byID : catalog.agents.filter((item) => item.name === name)
  if (!matches.length) {
    const status = getString(catalog.discovery.status)
    if (["pending", "refreshing", "unavailable"].includes(status))
      throw new Error(`AgentPaaS 尚未找到“${name}”，远端发现状态：${status}；请稍后刷新目录，任务未提交。`)
    throw new Error(`AgentPaaS 中没有名为“${name}”的 active Agent`)
  }
  if (matches.length > 1)
    throw new Error(
      `AgentPaaS 中存在 ${matches.length} 个名为“${name}”的 active Agent，请使用目录中的 agent_id：${matches.map((item) => item.id).join(", ")}`,
    )
  if (!isSupportedAgentID(matches[0].id)) throw new Error(`AgentPaaS 为“${name}”返回了无效 ID: ${matches[0].id}`)
  const revision = matches[0].current_version ?? matches[0].agent_version
  const scope = digest(canonicalJSON([config.url, config.apiKey, matches[0].id]))
  if (revision !== undefined)
    revisions.set(scope, canonicalJSON([revision, matches[0].runtime_mode, matches[0].runtime_binding]))
  else revisions.delete(scope)
  return matches[0].id
}

// Parse the two supported ID namespaces; never accept URL/path syntax or decoded input.
// Every route still encodes the complete ID as one path segment.
export function isSupportedAgentID(id: string): boolean {
  if (id.length > 512) return false
  if (id.startsWith("ag_")) return /^ag_[A-Za-z0-9_-]+$/.test(id)
  const parts = id.split(":")
  if (parts.length !== 3 || parts[0] !== "locostack") return false
  const namespace = parts[1]
  const name = parts[2]
  const label = /^[a-z0-9](?:[a-z0-9-]*[a-z0-9])?$/
  return (
    namespace.length <= 63 &&
    label.test(namespace) &&
    name.length <= 253 &&
    name.split(".").every((part) => part.length <= 63 && label.test(part))
  )
}

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

export function digest(value: string) {
  return new Bun.CryptoHasher("sha256").update(value).digest("hex")
}

export function canonicalJSON(value: unknown): string {
  if (Array.isArray(value)) return `[${value.map(canonicalJSON).join(",")}]`
  const record = asRecord(value)
  if (record)
    return `{${Object.keys(record)
      .sort()
      .filter((key) => record[key] !== undefined)
      .map((key) => `${JSON.stringify(key)}:${canonicalJSON(record[key])}`)
      .join(",")}}`
  return JSON.stringify(value) ?? "null"
}

export async function requestContext(
  config: { url: string; apiKey: string },
  agentID: string,
  key: string,
  current: ConversationState & { intent: string },
) {
  const directory = path.join(agentPaaSConversationDir(), "requests")
  const filePath = path.join(directory, `${digest(canonicalJSON([config.url, config.apiKey, agentID, key]))}.json`)
  await mkdir(directory, { recursive: true, mode: 0o700 })
  const temporary = `${filePath}.${crypto.randomUUID()}.tmp`
  await Bun.write(temporary, JSON.stringify(current), { mode: 0o600 })
  // link is atomic and never replaces the first submission's saved context.
  const created = await link(temporary, filePath)
    .then(
      () => true,
      (error: NodeJS.ErrnoException) => {
        if (error.code === "EEXIST") return false
        throw error
      },
    )
    .finally(() => rm(temporary, { force: true }))
  const saved = asRecord(await Bun.file(filePath).json())
  if (!saved || typeof saved.threadID !== "string" || typeof saved.runID !== "string")
    throw new Error("AgentPaaS 提交记录损坏，停止提交；请先检查原运行状态。")
  if (saved.intent !== current.intent) throw new Error("AgentPaaS 幂等键已用于不同请求；修改后的任务必须使用新键。")
  return { threadID: saved.threadID, runID: saved.runID, created }
}

export 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
  }
}

const conversationWrites = new Map<string, Promise<void>>()

export async function saveConversationState(
  sessionID: string,
  key: string,
  state: ConversationState,
  required = false,
) {
  const filePath = conversationStatePath(sessionID)
  const pending = (conversationWrites.get(filePath) ?? Promise.resolve())
    .catch(() => undefined)
    .then(async () => {
      const file = Bun.file(filePath)
      const saved = (await file.exists()) ? asRecord(await file.json()) : {}
      if (!saved) throw new Error("当前会话记录损坏，未覆盖已有记录")
      saved[key] = { thread_id: state.threadID, run_id: state.runID }
      await mkdir(agentPaaSConversationDir(), { recursive: true, mode: 0o700 })
      const temporary = `${filePath}.${crypto.randomUUID()}.tmp`
      try {
        await Bun.write(temporary, `${JSON.stringify(saved, null, 2)}\n`, { mode: 0o600 })
        await rename(temporary, filePath)
      } finally {
        await rm(temporary, { force: true })
      }
    })
  conversationWrites.set(filePath, pending)
  try {
    await pending
  } catch (error) {
    if (required) throw error
  } finally {
    if (conversationWrites.get(filePath) === pending) conversationWrites.delete(filePath)
  }
  conversationCache.set(`${sessionID}\n${key}`, state)
}

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

export 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
  const parsedURL = new URL(url)
  if (
    !["http:", "https:"].includes(parsedURL.protocol) ||
    parsedURL.username ||
    parsedURL.password ||
    parsedURL.search ||
    parsedURL.hash
  )
    throw new Error("AgentPaaS 地址必须是无凭证、查询参数和片段的 HTTP(S) 地址")
  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")
}

export function agentPaaSConversationDir() {
  return (
    process.env.AGENTPAAS_BRIDGE_STATE_DIR?.trim() || path.join(agentPaaSHome(), ".agentpaas", "opencode-conversations")
  )
}

export function asRecord(value: unknown): Record<string, unknown> | undefined {
  if (value && typeof value === "object" && !Array.isArray(value)) return value as Record<string, unknown>
}
export function getString(value: unknown): string {
  return typeof value === "string" ? value : ""
}

export function scientificLabel(status: unknown): string {
  if (status === "needs_evidence") return "证据不足"
  if (status === "needs_revision") return "方案待修订"
  if (status === "review_pending") return "审阅未完成，已保留方案"
  if (status === "needs_review") return "方案检查通过，待人工审阅"
  return "研究状态待核查"
}

export function formatRunResult(agentName: string, run: Record<string, unknown>, extra: Record<string, unknown> = {}) {
  const status =
    asRecord(run.pending_confirmation)?.status === "pending" && run.status === "running"
      ? "waiting_confirmation"
      : getString(run.status) || "unknown"
  const labels: Record<string, string> = {
    completed: "已完成",
    partial: "部分完成",
    needs_clarification: "需要补充信息",
    failed: "失败",
    cancelled: "已取消",
    paused: "已暂停",
    waiting_human: "等待确认",
    waiting_confirmation: "等待确认",
    queued: "排队中",
    running: "运行中",
    resuming: "恢复中",
    unknown: "状态待核查",
  }
  const artifacts = Array.isArray(run.artifacts) ? run.artifacts : []
  const metadata: Record<string, unknown> = { ...run, ...extra, status, artifacts }
  const science = asRecord(asRecord(run.research)?.science)
  const hints: Record<string, string> = {
    partial: "任务已部分完成，保留已有产物；请查看缺失项后继续补充。",
    needs_clarification: "需要补充信息后才能继续；请回答智能体提出的问题，不要重新执行同一请求。",
    paused: `任务已暂停。checkpoint_id=${getString(run.checkpoint_id) || "未提供"}；使用 agentpaas_task 的 resume 恢复原任务。`,
    waiting_human: "任务正在等待人工确认；使用 agentpaas_task 查看审批材料并提交决定，这不是任务失败。",
    waiting_confirmation: "任务正在等待工具确认；使用 agentpaas_task 查看并确认原任务。",
    cancelled: "任务已被取消。",
    queued: "任务已排队；使用 agentpaas_task 的 watch 查看进度。",
    running: "任务仍在运行；使用 agentpaas_task 的 watch 恢复查看，不要重新提交。",
    resuming: "任务正在恢复；使用 agentpaas_task 的 watch 查看进度。",
    failed: "失败严禁自动重跑、拆分任务或调用其他 subagent；必须先向用户报告真实错误和已有产物，等待用户明确指令。",
  }
  const artifactLines = artifacts.flatMap((item) => {
    const artifact = asRecord(item)
    if (!artifact || !getString(artifact.artifact_id)) return []
    return [`- ${getString(artifact.name).replace(/[\r\n]/g, " ") || "文件"}（artifact_id=${artifact.artifact_id}）`]
  })
  return {
    title:
      status === "completed"
        ? science
          ? scientificLabel(science.status)
          : `AgentPaaS ${agentName}`
        : `AgentPaaS ${agentName}（${labels[status] ?? "状态待核查"}）`,
    output: [
      ...(science ? [`研究状态：${scientificLabel(science.status)}。执行结束不代表科学结论成立。`] : []),
      ...(science?.execution_status === "conditional_diagnostic_completed"
        ? ["模拟状态：已完成设定模型下的诊断；证据充分性与现实适用性须另行判断。"]
        : []),
      ...(status === "failed"
        ? [`[AgentPaaS 执行失败] ${getString(run.error_message) || getString(asRecord(run.error)?.message)}`]
        : []),
      getString(run.output),
      ...(run.cancellation_status
        ? [
            {
              requested: "取消请求已发送，仍待服务端确认。",
              confirmed: "服务端已确认取消。",
              unconfirmed: "取消未确认，任务可能仍在执行；请查询原任务。",
              already_terminal: "任务已结束，不能将其标记为由本次取消操作停止。",
            }[getString(run.cancellation_status)] || "取消状态待核查。",
          ]
        : []),
      hints[status] ?? "",
      getString(run.persistence_warning),
      ...(metadata.run_id ? [`agent_id=${getString(metadata.agent_id)} · run_id=${metadata.run_id}`] : []),
      ...(asRecord(run.research)?.revision !== undefined ? [`研究版本：${asRecord(run.research)?.revision}`] : []),
      `用量：tokens=${run.total_tokens ?? asRecord(run.usage)?.total_tokens ?? "未报告"}（${asRecord(run.usage)?.usage_complete === true ? "计量完整" : "完整性未确认"}）；费用 USD=${run.cost_usd ?? asRecord(run.usage)?.cost_usd ?? "未报告"}（${asRecord(run.usage)?.cost_complete === true ? "计量完整" : "完整性未确认，不能据 0 判断免费"}）`,
      ...(run.workspace_path ? [`服务端工作目录：${run.workspace_path}`] : []),
      ...(artifactLines.length ? ["产物（使用 agentpaas_download 下载到本机）：", ...artifactLines] : []),
      ...(run.pending_approval ? [`待确认材料：${JSON.stringify(run.pending_approval)}`] : []),
      ...(run.pending_confirmation ? [`待确认工具：${JSON.stringify(run.pending_confirmation)}`] : []),
      ...(asRecord(run.pending_confirmation)?.expires_at
        ? [`工具确认截止：${asRecord(run.pending_confirmation)?.expires_at}；超时不会自动批准。`]
        : []),
      ...(run.pending_approval
        ? [
            `人工审阅截止：${getString(asRecord(run.pending_approval)?.expires_at) || "服务端未指定"}；与普通工具确认分开处理。`,
          ]
        : []),
    ]
      .filter(Boolean)
      .join("\n"),
    metadata,
  }
}