|
|
@@ -1,6 +1,6 @@
|
|
|
/// <reference path="../env.d.ts" />
|
|
|
import { tool } from "@opencode-ai/plugin"
|
|
|
-import { mkdir } from "node:fs/promises"
|
|
|
+import { link, mkdir, rm } from "node:fs/promises"
|
|
|
import { homedir } from "node:os"
|
|
|
import path from "node:path"
|
|
|
|
|
|
@@ -66,7 +66,7 @@ export default tool({
|
|
|
}
|
|
|
const agentID = await resolveAgentID(config, args.agent_name, context.abort)
|
|
|
const conversation = args.conversation ?? "continue"
|
|
|
- const conversationKey = `${config.url}\n${agentID}`
|
|
|
+ const conversationKey = `${config.url}\n${agentID}\n${digest(config.apiKey)}`
|
|
|
const savedConversation =
|
|
|
conversation === "new" ? undefined : await loadConversationState(context.sessionID, conversationKey)
|
|
|
const threadID = args.thread_id?.trim() || savedConversation?.threadID || ""
|
|
|
@@ -74,6 +74,13 @@ export default tool({
|
|
|
conversation === "continue" && (!args.thread_id?.trim() || args.thread_id.trim() === savedConversation?.threadID)
|
|
|
? savedConversation?.runID || ""
|
|
|
: ""
|
|
|
+ const intent = digest(canonicalJSON({ ...args, idempotency_key: undefined, conversation }))
|
|
|
+ const idempotencyKey =
|
|
|
+ args.idempotency_key?.trim() ||
|
|
|
+ `opencode-${digest(canonicalJSON([context.sessionID, context.messageID, agentID, intent]))}`
|
|
|
+ // Freeze the parent before submitting. Once `started` advances the session,
|
|
|
+ // retrying this same request must not turn it into a new research revision.
|
|
|
+ const admission = await requestContext(config, agentID, idempotencyKey, { intent, threadID, runID: continueRunID })
|
|
|
// OpenCode 的宿主机目录不会自动挂载进 AgentPaaS Docker。只有调用方
|
|
|
// 明确给出服务端可见路径时才覆盖,否则让 PaaS 创建并隔离 run workspace。
|
|
|
const workDir = args.work_dir
|
|
|
@@ -86,17 +93,17 @@ export default tool({
|
|
|
Authorization: `Bearer ${config.apiKey}`,
|
|
|
"Content-Type": "application/json",
|
|
|
Accept: "text/event-stream",
|
|
|
- ...(args.idempotency_key ? { "Idempotency-Key": args.idempotency_key } : {}),
|
|
|
+ "Idempotency-Key": idempotencyKey,
|
|
|
},
|
|
|
body: JSON.stringify({
|
|
|
input: args.input,
|
|
|
mode: args.mode,
|
|
|
target_subagent: args.target_subagent ?? "",
|
|
|
- thread_id: threadID,
|
|
|
+ thread_id: admission.threadID,
|
|
|
parameters: args.parameters ?? {},
|
|
|
context: {
|
|
|
...(workDir ? { work_dir: workDir } : {}),
|
|
|
- ...(continueRunID ? { run_id: continueRunID } : {}),
|
|
|
+ ...(admission.runID ? { run_id: admission.runID } : {}),
|
|
|
...(args.question_id ? { question_id: args.question_id } : {}),
|
|
|
...(args.feedback_kind ? { feedback_kind: args.feedback_kind } : {}),
|
|
|
...(args.input_artifact_ids ? { input_artifact_ids: args.input_artifact_ids } : {}),
|
|
|
@@ -117,6 +124,7 @@ export default tool({
|
|
|
const liveMetadata: Record<string, unknown> = {
|
|
|
agent_name: args.agent_name,
|
|
|
agent_id: agentID,
|
|
|
+ idempotency_key: idempotencyKey,
|
|
|
status: "running",
|
|
|
}
|
|
|
let done: Record<string, unknown> | null = null
|
|
|
@@ -145,10 +153,13 @@ export default tool({
|
|
|
startedRunID = getString(data?.run_id)
|
|
|
const remoteThreadID = getString(data?.thread_id) || getString(data?.session_id)
|
|
|
if (remoteThreadID && startedRunID) {
|
|
|
- await saveConversationState(context.sessionID, conversationKey, {
|
|
|
- threadID: remoteThreadID,
|
|
|
- runID: startedRunID,
|
|
|
- })
|
|
|
+ const latest = await loadConversationState(context.sessionID, conversationKey)
|
|
|
+ if (admission.created || !latest || latest.runID === admission.runID || latest.runID === startedRunID) {
|
|
|
+ await saveConversationState(context.sessionID, conversationKey, {
|
|
|
+ threadID: remoteThreadID,
|
|
|
+ runID: startedRunID,
|
|
|
+ })
|
|
|
+ }
|
|
|
}
|
|
|
if (context.abort.aborted) cancelOnAbort()
|
|
|
updateMetadata("Research67 正在运行", {
|
|
|
@@ -304,7 +315,9 @@ export default tool({
|
|
|
}
|
|
|
if (!done) {
|
|
|
const hint = sawCancelled ? "(收到 cancelled 事件)" : "(未收到 done 事件,连接可能中断)"
|
|
|
- throw new Error(`AgentPaaS 任务未正常结束${hint}${errors.length ? ":" + errors.join("; ") : ""}`)
|
|
|
+ throw new Error(
|
|
|
+ `AgentPaaS 任务未正常结束${hint}${errors.length ? ":" + errors.join("; ") : ""}。run_id=${startedRunID || "尚未收到"};idempotency_key=${idempotencyKey}。请先查询原运行状态,不要换键重复执行。`,
|
|
|
+ )
|
|
|
}
|
|
|
const status = typeof done.status === "string" ? done.status : "unknown"
|
|
|
if (status === "completed" || status === "failed") progress.stopUnverified()
|
|
|
@@ -314,6 +327,7 @@ export default tool({
|
|
|
const metadata = {
|
|
|
agent_name: args.agent_name,
|
|
|
agent_id: agentID,
|
|
|
+ idempotency_key: idempotencyKey,
|
|
|
run_id: done.run_id ?? liveMetadata.run_id,
|
|
|
thread_id: done.thread_id ?? done.session_id ?? liveMetadata.thread_id,
|
|
|
status,
|
|
|
@@ -429,6 +443,50 @@ function isSupportedAgentID(id: string): boolean {
|
|
|
|
|
|
const conversationCache = new Map<string, ConversationState>()
|
|
|
|
|
|
+function digest(value: string) {
|
|
|
+ return new Bun.CryptoHasher("sha256").update(value).digest("hex")
|
|
|
+}
|
|
|
+
|
|
|
+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"
|
|
|
+}
|
|
|
+
|
|
|
+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 }
|
|
|
+}
|
|
|
+
|
|
|
async function loadConversationState(sessionID: string, key: string): Promise<ConversationState | undefined> {
|
|
|
const cacheKey = `${sessionID}\n${key}`
|
|
|
const cached = conversationCache.get(cacheKey)
|