opencode/.opencode/lib/recovery.ts

import { mkdir, readdir, rename, rm } from "node:fs/promises"
import path from "node:path"
import { setTimeout } from "node:timers/promises"
import {
  agentPaaSConversationDir,
  asRecord,
  canonicalJSON,
  digest,
  getString,
  requestJSON,
  trackRun,
  type Connection,
} from "./agentpaas"

function directory(config: Connection) {
  return path.join(agentPaaSConversationDir(), "recovery", digest(canonicalJSON([config.url, config.apiKey])))
}
function filename(config: Connection, kind: string, agentID: string, id: string) {
  return path.join(directory(config), `${kind}-${digest(canonicalJSON([agentID, id]))}.json`)
}
async function write(file: string, value: unknown) {
  await mkdir(path.dirname(file), { recursive: true, mode: 0o700 })
  const temporary = `${file}.${crypto.randomUUID()}.tmp`
  try {
    await Bun.write(temporary, JSON.stringify(value), { mode: 0o600 })
    await rename(temporary, file)
  } finally {
    await rm(temporary, { force: true })
  }
}

export async function pendingRequest(config: Connection, agentID: string, key: string) {
  await write(filename(config, "request", agentID, key), { agent_id: agentID, idempotency_key: key })
}
export async function clearPendingRequest(config: Connection, agentID: string, key: string) {
  await rm(filename(config, "request", agentID, key), { force: true })
}
export async function recoverRequest(config: Connection, agentID: string, key: string, signal: AbortSignal) {
  const receipt = await requestJSON(
    config,
    `/agents/${encodeURIComponent(agentID)}/requests/${encodeURIComponent(key)}`,
    signal,
  )
  const run = asRecord(receipt.run)
  if (receipt.found !== true || !getString(run?.run_id)) return undefined
  await trackRun(config, {
    agent_id: agentID,
    agent_name: agentID,
    run_id: getString(run!.run_id),
    thread_id: getString(run!.session_id) || getString(run!.thread_id),
  })
  await clearPendingRequest(config, agentID, key)
  return run
}
export async function recoverPendingRequests(config: Connection, signal: AbortSignal) {
  const files = await readdir(directory(config)).catch((error: NodeJS.ErrnoException) => {
    if (error.code === "ENOENT") return []
    throw error
  })
  const warnings: string[] = []
  // Bound each refresh; keep unresolved receipts for a later read-only retry.
  for (const file of files.filter((file) => /^request-[a-f0-9]{64}\.json$/.test(file)).slice(0, 20)) {
    const row = asRecord(
      await Bun.file(path.join(directory(config), file))
        .json()
        .catch(() => undefined),
    )
    if (!row) {
      warnings.push("提交凭据损坏，请查询服务端历史。")
      continue
    }
    const agentID = getString(row.agent_id),
      key = getString(row.idempotency_key)
    try {
      if (!(await recoverRequest(config, agentID, key, signal)))
        warnings.push(`${agentID} 的提交尚未查到，仍保留凭据；不要换键重跑。idempotency_key=${key}`)
    } catch (error) {
      if (signal.aborted) throw error
      warnings.push(`提交结果暂不可核查：${agentID}；${String(error)}`)
    }
  }
  return warnings
}

export async function observation(config: Connection, agentID: string, runID: string) {
  const file = Bun.file(filename(config, "run", agentID, runID))
  return (await file.exists()) ? (asRecord(await file.json()) ?? {}) : {}
}
export async function saveObservation(
  config: Connection,
  agentID: string,
  runID: string,
  update: Record<string, unknown>,
) {
  await write(filename(config, "run", agentID, runID), { ...(await observation(config, agentID, runID)), ...update })
}

export async function cancelAndObserve(config: Connection, agentID: string, runID: string) {
  const base = `/agents/${encodeURIComponent(agentID)}/runs/${encodeURIComponent(runID)}`
  let persistenceWarning = ""
  const persist = async (state: string) => {
    await saveObservation(config, agentID, runID, { cancellation_status: state }).catch(() => {
      persistenceWarning = "本地取消记录未保存；请使用 run_id 查询服务端状态。"
    })
  }
  await persist("requested")
  const signal = AbortSignal.timeout(6000)
  let error = ""
  await requestJSON(config, `${base}/cancel`, signal, {
    method: "POST",
    headers: { "Content-Type": "application/json" },
    body: "{}",
  }).catch((cause) => {
    error = String(cause)
  })
  let snapshot: Record<string, unknown> = {}
  for (let attempt = 0; attempt < 3 && !signal.aborted; attempt++) {
    snapshot = await requestJSON(config, base, signal).catch(() => snapshot)
    if (["cancelled", "completed", "partial", "failed"].includes(getString(snapshot.status))) break
    await setTimeout(300 * (attempt + 1), undefined, { signal }).catch(() => undefined)
  }
  const state =
    snapshot.status === "cancelled"
      ? "confirmed"
      : ["completed", "partial", "failed"].includes(getString(snapshot.status))
        ? "already_terminal"
        : error
          ? "unconfirmed"
          : "requested"
  await persist(state)
  return {
    ...snapshot,
    run_id: runID,
    status: snapshot.status || "unknown",
    cancellation_status: state,
    ...(persistenceWarning ? { persistence_warning: persistenceWarning } : {}),
  }
}

export function cancellationStatus(saved: Record<string, unknown>, status: unknown) {
  if (!saved.cancellation_status) return undefined
  return status === "cancelled"
    ? "confirmed"
    : ["completed", "partial", "failed"].includes(getString(status))
      ? "already_terminal"
      : saved.cancellation_status
}