opencode/.opencode/tool/agentpaas.ts

import { tool } from "@opencode-ai/plugin"
import {
  recoverPendingRequests,
  recoverRequest,
  observation,
  saveObservation,
  cancelAndObserve,
  cancellationStatus,
} from "../lib/recovery"
import { mkdir, open, realpath, stat } from "node:fs/promises"
import path from "node:path"
import { setTimeout } from "node:timers/promises"
import { pathToFileURL } from "node:url"
import {
  asRecord,
  digest,
  formatRunResult,
  getString,
  isSupportedAgentID,
  listAgents,
  loadAgentPaaSConfig,
  loadConversationState,
  saveConversationState,
  requestJSON,
  resolveAgentID,
  verifyAgent,
  trackRun,
  trackedRuns,
} from "../lib/agentpaas"

const selector = {
  agent_id: tool.schema.string().optional().describe("目录返回的智能体 ID，优先使用"),
  agent_name: tool.schema.string().optional().describe("智能体名称；同名时必须用 ID"),
}

export const agents = tool({
  description: "检查 PaaS 连接并列出当前账号可用的全部智能体、用途和本地/远端运行位置。不执行任务。",
  args: {},
  async execute(_args, context) {
    const config = await loadAgentPaaSConfig()
    const catalog = await listAgents(config, context.abort)
    return {
      title: "PaaS 智能体目录",
      output: [
        `已连接 ${config.url}；发现状态：${getString(catalog.discovery.status)}`,
        ...catalog.agents.map(
          (agent) =>
            `${agent.name} | ${agent.id} | ${agent.runtime_mode ?? "unknown"} | ${agent.status ?? "unknown"}\n${getString(agent.description)}`,
        ),
        ...(!catalog.agents.length ? ["当前列表为空；远端发现未完成或不可用时，不能据此判断平台没有智能体。"] : []),
      ].join("\n"),
      metadata: { server: config.url, ...catalog },
    }
  },
})

export const check = tool({
  description:
    "检查智能体的执行服务、skill、MCP 与知识绑定。不创建研究任务。默认不调用模型；check_model=true 会产生一次或多次小额模型请求，knowledge_query 可验证检索。未验证不等于不可用。",
  args: {
    ...selector,
    check_model: tool.schema.boolean().optional(),
    knowledge_query: tool.schema.string().max(500).optional(),
  },
  async execute(args, context) {
    const config = await loadAgentPaaSConfig()
    const agentID = await resolveAgentID(config, args.agent_id || args.agent_name || "", context.abort)
    const report = await verifyAgent(config, agentID, context.abort, {
      check_model: args.check_model ?? false,
      knowledge_query: args.knowledge_query ?? "",
    })
    return { title: "PaaS 能力检查", output: report.summary, metadata: { agent_id: agentID, preflight: report } }
  },
})

export const tasks = tool({
  description:
    "找回本机此 PaaS 账号登记过的最近任务，跨 OpenCode 会话刷新服务端真实状态。只观察，不重跑、不取消、不自动批准。未登记的更早任务用 task/history 查询。",
  args: { limit: tool.schema.number().int().min(1).max(50).optional() },
  async execute(args, context) {
    const config = await loadAgentPaaSConfig()
    const warnings = await recoverPendingRequests(config, context.abort)
    const saved = await trackedRuns(config, args.limit ?? 20)
    const results: Array<ReturnType<typeof formatRunResult>> = []
    // Bound concurrency so reopening a client cannot flood the control plane.
    for (let offset = 0; offset < saved.length; offset += 5) {
      results.push(
        ...(await Promise.all(
          saved.slice(offset, offset + 5).map(async (row) => {
            const snapshot = await requestJSON(
              config,
              `/agents/${encodeURIComponent(row.agent_id)}/runs/${encodeURIComponent(row.run_id)}`,
              context.abort,
            ).catch((error): Record<string, unknown> => {
              if (context.abort.aborted) throw error
              return {
                status: "unknown",
                lookup_warning: `状态读取失败：${String(error)}；已保留任务编号，不会重新执行。`,
              }
            })
            // A task list should not copy every research report or approval payload
            // into the model context. Read full materials through task/status.
            const confirmation = asRecord(snapshot.pending_confirmation)
            const local = await observation(config, row.agent_id, row.run_id)
            const count = Array.isArray(snapshot.artifacts) ? snapshot.artifacts.length : 0
            return formatRunResult(row.agent_name, {
              agent_id: row.agent_id,
              run_id: row.run_id,
              status: snapshot.status,
              output:
                getString(snapshot.lookup_warning) ||
                `已登记产物：${count}。完整结果和审批材料请用 agentpaas_task status 查看。`,
              pending_confirmation: confirmation
                ? {
                    status: confirmation.status,
                    approval_id: confirmation.approval_id,
                    expires_at: confirmation.expires_at,
                  }
                : undefined,
              pending_approval: snapshot.pending_approval
                ? {
                    approval_id: asRecord(snapshot.pending_approval)?.approval_id,
                    expires_at: asRecord(snapshot.pending_approval)?.expires_at,
                  }
                : undefined,
              cancellation_status: cancellationStatus(local, snapshot.status),
              usage: snapshot.usage,
              checkpoint_id: snapshot.checkpoint_id,
              artifact_count: count,
              total_tokens: snapshot.total_tokens ?? asRecord(snapshot.usage)?.total_tokens,
              cost_usd: snapshot.cost_usd ?? asRecord(snapshot.usage)?.cost_usd,
            })
          }),
        )),
      )
    }
    return {
      title: "PaaS 最近任务",
      output: results.length
        ? [results.map((result) => `${result.title}\n${result.output}`).join("\n\n"), ...warnings].join("\n")
        : ["此账号尚无本机任务索引；可先用 agentpaas_task history 查询指定智能体的服务端历史。", ...warnings].join(
            "\n",
          ),
      metadata: {
        runs: results.map((result) => result.metadata),
        scope: "locally_tracked_runs",
        limit: args.limit ?? 20,
        warnings,
      },
    }
  },
})

export const task = tool({
  description:
    "查询、跟踪或控制已有 PaaS 任务，不创建新任务。history 查看历史；select 选中历史任务作为当前对话的后续上下文；status 查询；watch 恢复进度；pause/resume/cancel 控制原任务；approve/reject 处理审批。",
  args: {
    ...selector,
    action: tool.schema.enum([
      "history",
      "recover",
      "select",
      "status",
      "watch",
      "pause",
      "resume",
      "cancel",
      "approve",
      "reject",
    ]),
    run_id: tool.schema.string().optional().describe("省略时使用本 OpenCode 会话中该智能体的最近任务"),
    idempotency_key: tool.schema.string().optional().describe("recover 按原提交凭据找回任务，不重新提交"),
    checkpoint_id: tool.schema.string().optional(),
    approval_id: tool.schema.string().optional(),
    note: tool.schema.string().optional().describe("用户的审批意见，原样记录"),
    after: tool.schema.number().int().min(0).optional().describe("事件游标；省略时使用本机已保存进度，显式 0 重放"),
    wait_seconds: tool.schema
      .number()
      .int()
      .min(0)
      .max(120)
      .optional()
      .describe("watch 最长观察时间，默认 30 秒；到期不取消任务"),
  },
  async execute(args, context) {
    const config = await loadAgentPaaSConfig()
    const agentID = args.agent_id || (await resolveAgentID(config, args.agent_name || "", context.abort))
    if (!isSupportedAgentID(agentID)) throw new Error("无效 agent_id")
    const base = `/agents/${encodeURIComponent(agentID)}/runs`
    if (args.action === "recover") {
      if (!args.idempotency_key) throw new Error("recover 需要原 idempotency_key")
      const found = await recoverRequest(config, agentID, args.idempotency_key, context.abort)
      if (!found)
        return {
          title: "受理结果尚未确认",
          output: "尚未查到此提交；不能据此断定未受理。稍后再次 recover，不要换键重跑。",
          metadata: { agent_id: agentID, admission_status: "unknown" },
        }
      return formatRunResult(args.agent_name || agentID, found, { agent_id: agentID, admission_recovered: true })
    }
    if (args.action === "history") {
      const history = await requestJSON(config, base, context.abort)
      return {
        title: "PaaS 任务历史",
        output: JSON.stringify(history, null, 2),
        metadata: { agent_id: agentID, ...history },
      }
    }
    const conversationKey = `${config.url}\n${agentID}\n${digest(config.apiKey)}`
    const saved = await loadConversationState(context.sessionID, conversationKey)
    if (args.action === "select" && !args.run_id) throw new Error("select 必须明确指定要继续的 run_id")
    const runID = args.run_id || saved?.runID
    if (!runID) throw new Error("没有最近任务，请先查询 history 并选择 run_id；不会自动提交新任务。")
    const route = `${base}/${encodeURIComponent(runID)}`
    if (args.action === "cancel") {
      const result = await cancelAndObserve(config, agentID, runID)
      return formatRunResult(args.agent_name || agentID, result, { agent_id: agentID })
    }
    let snapshot = await requestJSON(config, route, context.abort)
    const local = await observation(config, agentID, runID)
    snapshot.cancellation_status = cancellationStatus(local, snapshot.status)
    if (args.action === "select") {
      const threadID = getString(snapshot.thread_id) || getString(snapshot.session_id)
      if (!threadID) throw new Error("此历史任务没有可恢复的会话 ID，未改变当前对话")
      await saveConversationState(context.sessionID, conversationKey, { threadID, runID }, true)
      const trackingWarning = await trackRun(config, {
        agent_id: agentID,
        agent_name: args.agent_name || agentID,
        run_id: runID,
        thread_id: threadID,
      }).then(
        () => "",
        () => "最近任务索引保存失败，请保留此 run_id；当前对话选择已保存。\n",
      )
      const result = formatRunResult(args.agent_name || agentID, snapshot, {
        agent_id: agentID,
        run_id: runID,
        selected: true,
      })
      return {
        ...result,
        output: `已选中此任务，后续修订将沿用会话和父版本。运行中先 watch，暂停用 resume，等待确认先处理审批。\n${trackingWarning}${result.output}`,
      }
    }
    if (["pause", "resume", "cancel", "approve", "reject"].includes(args.action)) {
      const approval = asRecord(snapshot.pending_approval)
      const confirmation = asRecord(snapshot.pending_confirmation)
      const checkpoint = args.checkpoint_id || getString(snapshot.checkpoint_id)
      if (args.action === "resume" && (snapshot.status !== "paused" || !checkpoint))
        throw new Error("原任务未暂停或缺少 checkpoint_id，不能通过新建任务代替恢复。")
      if (args.action === "resume" && snapshot.checkpoint_id && checkpoint !== snapshot.checkpoint_id)
        throw new Error("checkpoint_id 已过期，请查询原任务。")
      const confirming = args.action === "approve" || args.action === "reject"
      const currentApprovalID = getString(approval?.approval_id) || getString(confirmation?.approval_id)
      const approvalID = args.approval_id || currentApprovalID
      if (confirming) {
        if (
          !["waiting_human", "waiting_confirmation"].includes(getString(snapshot.status)) &&
          confirmation?.status !== "pending"
        )
          throw new Error("原任务当前没有等待确认；请检查状态，不会自动批准或重跑。")
        if (!currentApprovalID) throw new Error("缺少 approval_id，请升级执行服务并查询最新审批材料。")
        if (approvalID !== currentApprovalID) throw new Error("approval_id 已过期，请查询最新审批材料。")
        if (args.action === "approve")
          await context.ask({
            permission: "agentpaas_confirm",
            patterns: [`${agentID}:${runID}:${approvalID}`],
            always: [],
            metadata: {
              agent_id: agentID,
              run_id: runID,
              approval_id: approvalID,
              pending_approval: approval,
              pending_confirmation: confirmation,
              note: args.note ?? "",
            },
          })
      }
      const body = confirming
        ? {
            approved: args.action === "approve",
            approval_id: approvalID,
            note: args.note ?? "",
            idempotency_key: `opencode-control-${digest(JSON.stringify([config.url, digest(config.apiKey), agentID, runID, approvalID, args.action, args.note ?? ""]))}`,
          }
        : args.action === "resume"
          ? { checkpoint_id: checkpoint }
          : {}
      // Mutations are sent once. An ambiguous response must be resolved by status.
      await requestJSON(config, `${route}/${confirming ? "confirm" : args.action}`, context.abort, {
        method: "POST",
        headers: { "Content-Type": "application/json" },
        body: JSON.stringify(body),
      })
      snapshot = await requestJSON(config, route, context.abort)
    }
    if (args.action !== "watch")
      return formatRunResult(args.agent_name || agentID, snapshot, { agent_id: agentID, run_id: runID })
    let cursor = args.after ?? (typeof local.event_cursor === "number" ? local.event_cursor : 0)
    const warnings: string[] = []
    const stages = new Map<string, Record<string, unknown>>()
    if (args.after === undefined && Array.isArray(local.stages))
      for (const item of local.stages) {
        const row = asRecord(item)
        if (row && getString(row.stage_id)) stages.set(getString(row.stage_id), row)
      }
    let idle = 0
    const deadline = Date.now() + (args.action === "watch" ? (args.wait_seconds ?? 30) * 1000 : 0)
    do {
      const before = cursor
      const page = await requestJSON(config, `${route}/events?after=${cursor}&source=service`, context.abort).catch(
        (error) => {
          if (context.abort.aborted) throw error
          warnings.push(`事件读取暂不可用，已保留任务快照与产物：${String(error)}`)
          return { events: [] }
        },
      )
      for (const item of Array.isArray(page.events) ? page.events : []) {
        const event = asRecord(item)
        if (!event || typeof event.sequence !== "number" || event.sequence <= cursor) continue
        cursor = event.sequence
        const data = asRecord(event.data)
        const stageID = getString(data?.stage_id)
        const kind = getString(event.event_type) || getString(event.event)
        if (stageID && ["stage_started", "stage_passed", "stage_completed", "stage_failed"].includes(kind))
          stages.set(stageID, {
            ...data,
            status: kind === "stage_started" ? "running" : kind === "stage_failed" ? "failed" : "completed",
          })
      }
      snapshot = await requestJSON(config, route, context.abort).catch((error) => {
        if (context.abort.aborted) throw error
        warnings.push(`状态刷新暂不可用，以下为最近一次快照：${String(error)}`)
        return snapshot
      })
      idle = cursor === before ? idle + 1 : 0
      snapshot.cancellation_status = cancellationStatus(local, snapshot.status)
      await saveObservation(config, agentID, runID, { event_cursor: cursor, stages: [...stages.values()] }).catch(
        () => {
          warnings.push("进度游标保存失败，重新打开时可能重放旧事件；任务不会重跑。")
        },
      )
      context.metadata({
        title: `PaaS ${agentID} · ${snapshot.status}`,
        metadata: {
          agent_id: agentID,
          run_id: runID,
          status: snapshot.status,
          event_cursor: cursor,
          stages: [...stages.values()],
        },
      })
      if (
        warnings.length ||
        asRecord(snapshot.pending_confirmation)?.status === "pending" ||
        !["queued", "running", "resuming"].includes(getString(snapshot.status)) ||
        Date.now() >= deadline
      )
        break
      // Aborting observation never cancels the worker. cancel is a separate action.
      await setTimeout(
        Math.min(Math.min(500 * 2 ** Math.min(idle, 4), 5000), Math.max(0, deadline - Date.now())),
        undefined,
        { signal: context.abort },
      )
    } while (Date.now() <= deadline)
    const result = formatRunResult(args.agent_name || agentID, snapshot, {
      agent_id: agentID,
      run_id: runID,
      event_cursor: cursor,
      stages: [...stages.values()],
      warnings,
    })
    return { ...result, output: [result.output, ...warnings].join("\n") }
  },
})

const MAX_FILE_BYTES = 256 * 1024 * 1024

export const upload = tool({
  description:
    "把用户指定的本机文件上传到 PaaS，返回 file_id。随后将这些 ID 传入 agentpaas_run.input_file_ids。仅上传用户明确选择的文件。",
  args: { paths: tool.schema.array(tool.schema.string()).min(1).max(20) },
  async execute(args, context) {
    const config = await loadAgentPaaSConfig()
    const capabilities = await requestJSON(config, "/files/capabilities", context.abort)
    const serverLimit = capabilities.direct_upload_max_bytes
    if (typeof serverLimit !== "number" || !Number.isSafeInteger(serverLimit) || serverLimit < 0)
      throw new Error("PaaS 未返回有效上传限制，请更新 PaaS 后再上传")
    const limit = Math.min(MAX_FILE_BYTES, serverLimit)
    const paths = await Promise.all(args.paths.map((name) => realpath(path.resolve(context.directory, name))))
    for (const filename of paths) {
      const info = await stat(filename)
      if (!info.isFile() || info.size > limit) throw new Error(`文件不可上传或超过当前上限 ${limit} 字节：${filename}`)
    }
    await context.ask({
      permission: "agentpaas_upload",
      patterns: paths,
      always: [],
      metadata: { server: config.url, paths },
    })
    const files: Record<string, unknown>[] = []
    for (const filename of paths) {
      const file = Bun.file(filename)
      const bytes = await file.arrayBuffer()
      if (bytes.byteLength > limit) throw new Error(`文件超过当前上限 ${limit} 字节：${filename}`)
      const query = new URLSearchParams({
        name: path.basename(filename),
        mime_type: file.type || "application/octet-stream",
      })
      const result = await requestJSON(config, `/files?${query}`, context.abort, {
        method: "POST",
        headers: {
          "Content-Type": file.type || "application/octet-stream",
          "X-File-SHA256": new Bun.CryptoHasher("sha256").update(bytes).digest("hex"),
        },
        body: bytes,
      })
      if (!getString(result.file_id) || result.status !== "available") throw new Error("PaaS 未返回可用的 file_id")
      files.push({ file_id: result.file_id, name: result.name, size_bytes: result.size_bytes })
      context.metadata({ title: `已上传 ${files.length}/${paths.length} 个文件`, metadata: { files: [...files] } })
    }
    return { title: "输入文件已上传", output: JSON.stringify(files, null, 2), metadata: { files } }
  },
})

export const download = tool({
  description:
    "通过 PaaS 身份验证下载已有任务的产物到本机，校验大小/摘要并返回可打开的文件。不会读取服务端文件路径，也不会覆盖已有文件。",
  args: {
    ...selector,
    run_id: tool.schema.string(),
    artifact_id: tool.schema.string(),
    output_path: tool.schema.string().optional().describe("本机保存路径；省略时放入当前项目 .agentpaas-artifacts 目录"),
  },
  async execute(args, context) {
    const config = await loadAgentPaaSConfig()
    const agentID = args.agent_id || (await resolveAgentID(config, args.agent_name || "", context.abort))
    if (!isSupportedAgentID(agentID)) throw new Error("无效 agent_id")
    const route = `/agents/${encodeURIComponent(agentID)}/runs/${encodeURIComponent(args.run_id)}/artifacts`
    const manifest = await requestJSON(config, route, context.abort)
    const artifact = (Array.isArray(manifest.artifacts) ? manifest.artifacts : [])
      .map(asRecord)
      .find((item) => item?.artifact_id === args.artifact_id)
    if (!artifact) throw new Error("当前任务没有该产物")
    const name =
      path.basename(getString(artifact.name).replaceAll("\\", "/")).replace(/[<>:\x00-\x1f]/g, "_") || "artifact"
    const base = await realpath(context.directory)
    const destination = args.output_path
      ? path.resolve(base, args.output_path)
      : path.join(
          base,
          ".agentpaas-artifacts",
          digest(`${config.url}:${agentID}:${args.run_id}`).slice(0, 24),
          `${digest(args.artifact_id).slice(0, 12)}-${name}`,
        )
    await context.ask({
      permission: "agentpaas_download",
      patterns: [destination],
      always: [],
      metadata: {
        server: config.url,
        agent_id: agentID,
        run_id: args.run_id,
        artifact_id: args.artifact_id,
        destination,
      },
    })
    const response = await fetch(`${config.url}/api/v1${route}/${encodeURIComponent(args.artifact_id)}/download`, {
      headers: { Authorization: `Bearer ${config.apiKey}` },
      redirect: "error",
      signal: AbortSignal.any([context.abort, AbortSignal.timeout(60000)]),
    })
    if (!response.ok) throw new Error(`产物下载失败：HTTP ${response.status}`)
    if (Number(response.headers.get("Content-Length")) > MAX_FILE_BYTES) {
      await response.body?.cancel()
      throw new Error("产物超过 256 MiB，请通过平台文件传输接口获取")
    }
    const chunks: Uint8Array[] = []
    let size = 0
    const reader = response.body?.getReader()
    if (!reader) throw new Error("产物下载没有响应内容")
    try {
      while (true) {
        const chunk = await reader.read()
        if (chunk.done) break
        size += chunk.value.length
        if (size > MAX_FILE_BYTES) throw new Error("产物超过 256 MiB")
        chunks.push(chunk.value)
      }
    } finally {
      await reader.cancel().catch(() => undefined)
      reader.releaseLock()
    }
    const content = Buffer.concat(chunks)
    const hash = new Bun.CryptoHasher("sha256").update(content).digest("hex")
    if (artifact.sha256 && artifact.sha256 !== hash) throw new Error("产物摘要不匹配，未写入本机")
    if (typeof artifact.size_bytes === "number" && artifact.size_bytes > 0 && artifact.size_bytes !== size)
      throw new Error("产物大小不匹配，未写入本机")
    await mkdir(path.dirname(destination), { recursive: true })
    const parent = await realpath(path.dirname(destination))
    if (!args.output_path && !parent.startsWith(base + path.sep)) throw new Error("默认产物目录指向项目之外，拒绝写入")
    const handle = await open(path.join(parent, path.basename(destination)), "wx", 0o600)
    try {
      await handle.writeFile(content)
    } finally {
      await handle.close()
    }
    return {
      title: "产物已下载",
      output: `[${name.replace(/[\[\]]/g, "_")}](<${destination}>)`,
      metadata: {
        agent_id: agentID,
        run_id: args.run_id,
        artifact_id: args.artifact_id,
        local_path: destination,
        size_bytes: size,
        sha256: hash,
      },
      attachments: [
        {
          type: "file" as const,
          filename: name,
          mime: getString(artifact.mime_type) || "application/octet-stream",
          url: pathToFileURL(destination).href,
        },
      ],
    }
  },
})