fix: 防止 HWPOD operation 重复派发

2026-07-17 04:19:31 +02:00
parent ab76290e9d
commit 84dc6618e6
6 changed files with 197 additions and 19 deletions
@@ -445,16 +445,26 @@ export async function prepareCaseRun(context: CaseContext, action = "prepare") {
 }

 export async function buildCaseRun(context: CaseContext, prepared?: PreparedCaseRun) {
+  return buildCaseRunWithAuthorityMode(context, prepared, false);
+}
+
+export async function recoverBuildCaseRun(context: CaseContext, prepared?: PreparedCaseRun) {
+  return buildCaseRunWithAuthorityMode(context, prepared, true);
+}
+
+async function buildCaseRunWithAuthorityMode(context: CaseContext, prepared: PreparedCaseRun | undefined, recoverOnly: boolean) {
  const run = prepared ?? await loadOrPrepareCaseRun(context);
  const validationPlan = run.validationPlan ?? caseValidationPlanFromDefinition(run.definition);
  const buildStep = validationPlan.steps.find((step) => step.kind === "build")!;
  const command = hwpodCommandForValidationStep(buildStep, { executable: process.execPath, cliPath: path.join(context.cwd, "tools/hwpod-cli.ts"), specPath: run.specPath, reason: `case-run ${run.caseId} ${run.runId} ${validationPlan.mode}` });
-  const invoked = await (context.runProcess ?? runProcess)(command, context.cwd, {
-    ...process.env,
-    ...context.env,
-    HWLAB_RUNTIME_API_URL: run.apiUrl,
-    HWLAB_RUNTIME_ENDPOINT_LOCKED: "1"
-  });
+  const invoked = recoverOnly
+    ? await recoverHwpodInvocation(context, run)
+    : await (context.runProcess ?? runProcess)(command, context.cwd, {
+        ...process.env,
+        ...context.env,
+        HWLAB_RUNTIME_API_URL: run.apiUrl,
+        HWLAB_RUNTIME_ENDPOINT_LOCKED: "1"
+      });
  const hwpodPayload = parseJsonMaybe(invoked.stdout);
  const hwpodDocument = await readHwpodSpec(run.specPath);
  const operation = summarizeHwpodOperationForTest(hwpodPayload, hwpodDocument, "debug.build");
@@ -794,6 +804,40 @@ export async function runAgentTaskStage(context: CaseContext, run: PreparedCaseR
  return { run: nextRun, agent };
 }

+export async function recoverAgentTaskStage(context: CaseContext, run: PreparedCaseRun) {
+  const baseUrl = webUrlFrom(context);
+  const providerProfile = text(context.parsed.providerProfile) || run.agentTask.providerProfile;
+  const projectId = text(context.parsed.projectId) || run.agentTask.projectId;
+  const conversationId = text(context.parsed.conversationId) || `cnv_case_${slug(run.caseId)}_${slug(run.runId)}`;
+  const traceId = text(context.parsed.traceId);
+  if (!traceId) throw Object.assign(new Error("AgentRun recovery requires the stable trace identity"), { code: "agentrun_recovery_identity_required" });
+  const existing = run.agent?.traceId === traceId ? run.agent : null;
+  const resultUrl = text(existing?.resultUrl) || `${baseUrl}/v1/agent/chat/result/${encodeURIComponent(traceId)}`;
+  const recoveredAgent = existing ?? agentFailureStage({
+    stageStatus: "recovering",
+    baseUrl,
+    projectId,
+    providerProfile,
+    conversationId,
+    traceId,
+    promptPath: path.join(run.runDir, "agent-prompt.md"),
+    promptSha256: ""
+  });
+  const result = await pollAgentResult(context, { baseUrl, traceId, acceptedBody: { resultUrl }, resultUrl, run, agent: recoveredAgent });
+  const agent = agentWithResultEvidence({
+    ...recoveredAgent,
+    stageStatus: result.stageStatus,
+    resultUrl,
+    polls: result.polls,
+    timedOut: result.timedOut,
+    lastPollAt: result.lastPollAt,
+    lastHttpStatus: result.lastHttpStatus,
+    error: result.error
+  }, result.body, run.agentTask) as AgentRunStage;
+  const nextRun = await updateRun(context, run, { stage: "agent-recovered", agent });
+  return { run: nextRun, agent };
+}
+
 export async function collectAgentTraceEvidence(context: CaseContext, run: PreparedCaseRun) {
  const agent = run.agent;
  const traceId = text(agent?.traceId);
@@ -907,6 +951,22 @@ async function pollAgentResult(context: CaseContext, input: { baseUrl: string; t
  return { stageStatus: "timeout", body: lastBody, polls, timedOut: true, lastPollAt, lastHttpStatus: lastStatus, error: { code: "agent_task_timeout", traceId: input.traceId, timeoutMs, lastHttpStatus: lastStatus, lastBody: compactObject(lastBody) } };
 }

+async function recoverHwpodInvocation(context: CaseContext, run: PreparedCaseRun) {
+  const planId = text(context.env.HWLAB_HWPOD_OPERATION_IDENTITY);
+  if (!planId) throw Object.assign(new Error("HWPOD recovery requires the stable operation identity"), { code: "hwpod_recovery_identity_required" });
+  const timeoutMs = Math.max(numberOption(context.parsed.timeoutMs) ?? DEFAULT_KEIL_COMMAND_TIMEOUT_MS, 1000);
+  const startedAt = Date.now();
+  while (Date.now() - startedAt < timeoutMs) {
+    const response = await context.fetchImpl(`${run.apiUrl}/v1/hwpod-node-ops?planId=${encodeURIComponent(planId)}`, { method: "GET" });
+    const body = parseJsonMaybe(await response.text());
+    if (response.status === 404) throw Object.assign(new Error(`HWPOD operation ${planId} was not found`), { code: "hwpod_operation_not_found", planId });
+    if (!response.ok) throw Object.assign(new Error(`HWPOD operation lookup failed with HTTP ${response.status}`), { code: "hwpod_operation_lookup_failed", planId, status: response.status });
+    if (body?.result) return { stdout: JSON.stringify(body.result), stderr: "", exitCode: body.result.ok === false ? 1 : 0 };
+    await context.sleep(Math.max(numberOption(context.parsed.pollIntervalMs) ?? DEFAULT_POLL_INTERVAL_MS, 250));
+  }
+  throw Object.assign(new Error(`HWPOD operation ${planId} did not complete within ${timeoutMs}ms`), { code: "hwpod_operation_recovery_timeout", planId, timeoutMs });
+}
+
 function effectiveAgentTimeoutMs(context: CaseContext, agentTask?: PreparedAgentTask | null) {
  return numberOption(context.parsed.agentTimeoutMs ?? context.parsed.timeoutMs) ?? agentTask?.timeoutMs ?? DEFAULT_AGENT_TIMEOUT_MS;
 }