fix(tool): 保留 Research67 失败现场并联动取消

本地会话中断或远端阶段失败时，桥接层此前可能丢失工作区和结构化错误，也不会主动取消仍在运行的远端任务。本提交补齐失败传播与停止边界。

- 在 SSE started 后记录 run_id，并在本地 abort 时请求取消对应远端 Run
- 将 error 事件的错误码、消息和工作区路径持续写入工具 metadata
- 失败结果保留已有产物位置，并明确禁止自动重跑或改调其他 subagent
- 修正失败阶段里程碑状态与 TOOL_ERROR 识别
- 新增失败保留和远端取消回归测试；验证：bun test test/tool/agentpaas-run.test.ts（3 pass），bun typecheck（通过）

.opencode/agent/research-67-pipeline.md

@@ -12,6 +12,7 @@ tools:
 你是 Research67 Lite 流水线入口（research-67-pipeline）。
 
 使用 agentpaas_run 工具把用户任务转发给远端 AgentPaaS 的三阶段研究草稿流水线：
+
 - agent_name: Research67 三阶段研究草稿流水线
 - mode: iterate（多轮迭代）
 - input: 用户任务原文，可补充必要的科研背景
@@ -19,4 +20,10 @@ tools:
 只允许调用一次 agentpaas_run。工具返回后，无论成功或失败，都必须立即忠实返回结果，
 不要自己编造或补充科研内容，不得再次调用工具。
 
+若 agentpaas_run 返回失败、取消或中断：
+
+- 立即原样返回真实错误、run_id、workspace_path 和已有产物信息。
+- 失败严禁重跑。不得再次调用 agentpaas_run，不得拆分任务，不得改调其他 subagent 自行补做。
+- 只能等待用户检查并给出明确的新指令；不得把“建议重试”解释为已获准重试。
+
 > 注：Research67 Lite 当前使用 pipeline，不调用 orchestrator；agentpaas_run 会按 agent_name 动态解析当前 Agent ID。

.opencode/tool/agentpaas_run.ts

@@ -92,6 +92,7 @@ export default tool({
       status: "running",
     }
     let done: Record<string, unknown> | null = null
+    let startedRunID = ""
     let sawCancelled = false
     let lastThinkUpdate = 0
     const progress = createResearch67Progress()
@@ -103,9 +104,17 @@ export default tool({
         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)
+        if (context.abort.aborted) cancelOnAbort()
         updateMetadata("Research67 正在运行", {
           run_id: data?.run_id,
           thread_id: data?.thread_id,
@@ -220,8 +229,17 @@ export default tool({
       } else if (ev.event === "heartbeat") {
         progress.note("连接正常")
         updateMetadata(withStage("Research67 仍在运行", liveMetadata), { latest_event: ev.event })
-      } else if (ev.event === "error") errors.push(extractMessage(ev.data))
-      else if (ev.event === "cancelled") {
+      } 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" })
@@ -230,6 +248,7 @@ export default tool({
         break
       }
     }
+    context.abort.removeEventListener("abort", cancelOnAbort)
     if (!done) {
       const hint = sawCancelled ? "（收到 cancelled 事件）" : "（未收到 done 事件，连接可能中断）"
       throw new Error(`AgentPaaS 任务未正常结束${hint}${errors.length ? "：" + errors.join("; ") : ""}`)
@@ -239,13 +258,15 @@ export default tool({
     const metadata = {
       agent_name: args.agent_name,
       agent_id: agentID,
-      run_id: done.run_id,
-      thread_id: done.thread_id,
+      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,
+      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")
@@ -253,7 +274,13 @@ export default tool({
     if (status === "cancelled") return { title: `AgentPaaS ${args.agent_name}`, output: "任务已被取消。", metadata }
     return {
       title: `AgentPaaS ${args.agent_name}（失败）`,
-      output: `[AgentPaaS 执行失败] ${errors.join("; ") || String(done.output ?? "") || "未知错误"}`,
+      output: [
+        `[AgentPaaS 执行失败] ${errors.join("; ") || String(done.output ?? "") || "未知错误"}`,
+        metadata.workspace_path
+          ? `已有产物保留在：${metadata.workspace_path}`
+          : "已有产物已保留；请根据 run_id 查询工作区。",
+        "失败严禁自动重跑、拆分任务或调用其他 subagent；必须先向用户报告真实错误和已有产物，等待用户明确指令。",
+      ].join("\n"),
       metadata,
     }
   },
@@ -343,6 +370,21 @@ async function resolveConfirmation(
   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()
@@ -521,7 +563,13 @@ function createResearch67Progress() {
       const stage = stages.find((item) => item.id === stageID)
       if (!stage) return
       stage.status = status
-      for (const milestone of stage.milestones) milestone.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
       note = status === "failed" ? "阶段执行失败" : "阶段验收通过"
@@ -566,7 +614,7 @@ function createResearch67Progress() {
     },
     toolResult(toolName: string, content: string) {
       const lower = toolName.toLowerCase()
-      const failed = /\[error\]|mcp_error|"status"\s*:\s*"(?:failed|error)"/i.test(content)
+      const failed = /\[(?:tool_)?error\]|mcp_error|"status"\s*:\s*"(?:failed|error)"/i.test(content)
       if (current?.id === "literature-review" && ["arxiv_search", "openalex_search"].includes(lower)) {
         if (!failed) searchCalls++
         const milestone = current.milestones.find((item) => item.id === "search")!

packages/opencode/test/tool/agentpaas-run.test.ts

@@ -260,3 +260,190 @@ test("shows pipeline stages and resolves remote confirmations", async () => {
     await rm(home, { recursive: true, force: true })
   }
 })
+
+test("preserves completed artifacts and forbids automatic reruns after failure", async () => {
+  const home = await mkdtemp(path.join(tmpdir(), "agentpaas-bridge-failure-"))
+  const originalHome = process.env.HOME
+  const originalFetch = globalThis.fetch
+  const updates: Array<{ title?: string; metadata?: Record<string, unknown> }> = []
+  const workspacePath = "/workspace/runs/run_failure"
+  const events: Array<[string, unknown]> = [
+    ["started", { run_id: "run_failure", thread_id: "thread_failure" }],
+    [
+      "stage_started",
+      {
+        pipeline_id: "research67-one-round",
+        stage_id: "paper-writing",
+        stage_name: "实验前研究草稿撰写",
+        attempt: 1,
+        status: "running",
+      },
+    ],
+    ["tool_call", { tool: "WriteFile", content: '{"file_path":"/run/paper-writing/artifacts/outline.json"}' }],
+    ["tool_result", { tool: "WriteFile", content: "[OK] Created outline.json" }],
+    ["tool_call", { tool: "CompileLatex", content: '{"tex_path":"/run/paper-writing/artifacts/paper.tex"}' }],
+    ["tool_result", { tool: "CompileLatex", content: "[TOOL_ERROR] UnicodeDecodeError" }],
+    [
+      "stage_failed",
+      {
+        pipeline_id: "research67-one-round",
+        stage_id: "paper-writing",
+        stage_name: "实验前研究草稿撰写",
+        attempt: 1,
+        status: "failed",
+        error: "Native function-calling loop produced no final output",
+      },
+    ],
+    [
+      "error",
+      {
+        code: "AGENT_EXECUTION_FAILED",
+        message: "Native function-calling loop produced no final output",
+        workspace_path: workspacePath,
+      },
+    ],
+    ["done", { run_id: "run_failure", thread_id: "thread_failure", status: "failed" }],
+  ]
+
+  try {
+    process.env.HOME = home
+    globalThis.fetch = Object.assign(
+      async (input: URL | RequestInfo) => {
+        const pathname = new URL(input instanceof Request ? input.url : input.toString()).pathname
+        if (pathname === "/api/v1/agents") {
+          return Response.json({
+            agents: [{ id: "ag_pipeline", name: "Research67 三阶段研究草稿流水线" }],
+          })
+        }
+        if (pathname === "/api/v1/agents/ag_pipeline/run/stream") {
+          const body =
+            events.map(([event, data]) => `event: ${event}\ndata: ${JSON.stringify(data)}`).join("\n\n") + "\n\n"
+          return new Response(body, { headers: { "Content-Type": "text/event-stream" } })
+        }
+        return new Response("not found", { status: 404 })
+      },
+      { preconnect: originalFetch.preconnect },
+    )
+    const configDir = path.join(home, ".agentpaas")
+    await mkdir(configDir)
+    await writeFile(
+      path.join(configDir, "config.json"),
+      JSON.stringify({ server: "http://agentpaas.test", api_key: "test-key" }),
+    )
+    const bridge = (await import(`../../../../.opencode/tool/agentpaas_run.ts?failure=${Date.now()}`)).default
+    const result = await bridge.execute(
+      { input: "调查课题", agent_name: "Research67 三阶段研究草稿流水线", mode: "iterate" },
+      {
+        sessionID: "session_failure",
+        messageID: "message_failure",
+        agent: "research-67-pipeline",
+        directory: "/workspace",
+        worktree: "/workspace",
+        abort: new AbortController().signal,
+        metadata: (update: (typeof updates)[number]) => updates.push(update),
+        ask: async () => undefined,
+      },
+    )
+
+    expect(result.output).toContain("Native function-calling loop produced no final output")
+    expect(result.output).toContain(workspacePath)
+    expect(result.output).toContain("失败严禁自动重跑、拆分任务或调用其他 subagent")
+    expect(result.metadata).toMatchObject({
+      run_id: "run_failure",
+      status: "failed",
+      error_code: "AGENT_EXECUTION_FAILED",
+      workspace_path: workspacePath,
+      research67_progress: {
+        rows: expect.arrayContaining([
+          expect.objectContaining({ id: "paper-writing:outline", status: "completed" }),
+          expect.objectContaining({ id: "paper-writing:compile", status: "failed" }),
+          expect.objectContaining({ id: "paper-writing:report", status: "pending" }),
+        ]),
+      },
+    })
+  } finally {
+    globalThis.fetch = originalFetch
+    process.env.HOME = originalHome
+    await rm(home, { recursive: true, force: true })
+  }
+})
+
+test("cancels the remote run when the local request is aborted", async () => {
+  const home = await mkdtemp(path.join(tmpdir(), "agentpaas-bridge-cancel-"))
+  const originalHome = process.env.HOME
+  const originalFetch = globalThis.fetch
+  const abort = new AbortController()
+  const cancelledRuns: string[] = []
+  let markStarted: (() => void) | undefined
+  const started = new Promise<void>((resolve) => {
+    markStarted = resolve
+  })
+
+  try {
+    process.env.HOME = home
+    globalThis.fetch = Object.assign(
+      async (input: URL | RequestInfo, init?: RequestInit) => {
+        const pathname = new URL(input instanceof Request ? input.url : input.toString()).pathname
+        if (pathname === "/api/v1/agents") {
+          return Response.json({
+            agents: [{ id: "ag_pipeline", name: "Research67 三阶段研究草稿流水线" }],
+          })
+        }
+        if (pathname === "/api/v1/agents/ag_pipeline/run/stream") {
+          const body = new ReadableStream<Uint8Array>({
+            start(controller) {
+              controller.enqueue(
+                new TextEncoder().encode('event: started\ndata: {"run_id":"run_abort","thread_id":"thread_abort"}\n\n'),
+              )
+              init?.signal?.addEventListener(
+                "abort",
+                () => controller.error(new DOMException("Aborted", "AbortError")),
+                {
+                  once: true,
+                },
+              )
+            },
+          })
+          return new Response(body, { headers: { "Content-Type": "text/event-stream" } })
+        }
+        if (pathname === "/api/v1/agents/ag_pipeline/runs/run_abort/cancel") {
+          cancelledRuns.push("run_abort")
+          return Response.json({ run_id: "run_abort", status: "cancelled" })
+        }
+        return new Response("not found", { status: 404 })
+      },
+      { preconnect: originalFetch.preconnect },
+    )
+    const configDir = path.join(home, ".agentpaas")
+    await mkdir(configDir)
+    await writeFile(
+      path.join(configDir, "config.json"),
+      JSON.stringify({ server: "http://agentpaas.test", api_key: "test-key" }),
+    )
+    const bridge = (await import(`../../../../.opencode/tool/agentpaas_run.ts?cancel=${Date.now()}`)).default
+    const running = bridge.execute(
+      { input: "调查课题", agent_name: "Research67 三阶段研究草稿流水线", mode: "iterate" },
+      {
+        sessionID: "session_abort",
+        messageID: "message_abort",
+        agent: "research-67-pipeline",
+        directory: "/workspace",
+        worktree: "/workspace",
+        abort: abort.signal,
+        metadata: (update: { metadata?: Record<string, unknown> }) => {
+          if (update.metadata?.run_id === "run_abort") markStarted?.()
+        },
+        ask: async () => undefined,
+      },
+    )
+
+    await started
+    abort.abort()
+    await expect(running).rejects.toThrow("Aborted")
+    expect(cancelledRuns).toEqual(["run_abort"])
+  } finally {
+    globalThis.fetch = originalFetch
+    process.env.HOME = originalHome
+    await rm(home, { recursive: true, force: true })
+  }
+})