From da03fb7e0cb62744cc83ca1388e7e05465e65906 Mon Sep 17 00:00:00 2001 From: root Date: Fri, 17 Jul 2026 05:06:27 +0200 Subject: [PATCH] =?UTF-8?q?fix:=20=E4=BF=AE=E5=A4=8D=20HarnessRL=20?= =?UTF-8?q?=E7=9C=9F=E5=AE=9E=E9=87=8D=E8=AF=95=E7=8A=B6=E6=80=81=E6=9C=BA?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- internal/cloud/hwpod-node-ops.test.ts | 79 ++++++++++++++++++---- internal/cloud/hwpod-node-ws-registry.ts | 84 +++++++++++++++++------- internal/harnessrl/harnessrl.test.ts | 15 +++++ tools/src/hwlab-caserun-runtime.ts | 9 ++- 4 files changed, 150 insertions(+), 37 deletions(-) diff --git a/internal/cloud/hwpod-node-ops.test.ts b/internal/cloud/hwpod-node-ops.test.ts index 56185612..348bf997 100644 --- a/internal/cloud/hwpod-node-ops.test.ts +++ b/internal/cloud/hwpod-node-ops.test.ts @@ -487,23 +487,80 @@ test("WebSocket registry rebuild recovers one durable plan result without redisp test("WebSocket registry retries the same plan identity after pre-accept send failure", async () => { const records = new Map(); const plan = { ...samplePlan(), nodeId: "pc-host-1", planId: "retry-same-plan" }; - const firstRegistry = createHwpodNodeWsRegistry({ operationLedger: createMemoryHwpodOperationLedger({ records }) }); - const failedSocket = { send(value: string) { if (JSON.parse(value).type === "hwpod-node-ops") throw new Error("connection failed before acceptance"); }, close() {} }; - firstRegistry.openBunSocket(failedSocket); - firstRegistry.handleBunMessage(failedSocket, JSON.stringify({ type: "register", nodeId: "pc-host-1", maxInFlight: 1 })); - assert.equal((await firstRegistry.dispatch(plan, { requestId: "req_failed" }, { timeoutMs: 1000 })).status, "blocked"); - + const registry = createHwpodNodeWsRegistry({ operationLedger: createMemoryHwpodOperationLedger({ records }) }); + let failSend = true; let acceptanceCount = 0; - const retryRegistry = createHwpodNodeWsRegistry({ operationLedger: createMemoryHwpodOperationLedger({ records }) }); - const retrySocket = { send(value: string) { const message = JSON.parse(value); if (message.type !== "hwpod-node-ops") return; acceptanceCount += 1; queueMicrotask(() => { retryRegistry.handleBunMessage(retrySocket, JSON.stringify({ type: "hwpod-node-ops-accepted", requestId: message.requestId, nodeId: "pc-host-1" })); retryRegistry.handleBunMessage(retrySocket, JSON.stringify({ type: "hwpod-node-ops-result", requestId: message.requestId, nodeId: "pc-host-1", result: { ok: true, status: "completed", planId: message.plan.planId, results: [] } })); }); }, close() {} }; - retryRegistry.openBunSocket(retrySocket); - retryRegistry.handleBunMessage(retrySocket, JSON.stringify({ type: "register", nodeId: "pc-host-1", maxInFlight: 1 })); - const result = await retryRegistry.dispatch(plan, { requestId: "req_retry" }, { timeoutMs: 1000 }); + let dispatchCount = 0; + const socket = { send(value: string) { const message = JSON.parse(value); if (message.type !== "hwpod-node-ops") return; dispatchCount += 1; if (failSend) { failSend = false; throw new Error("connection failed before acceptance"); } queueMicrotask(() => { acceptanceCount += 1; registry.handleBunMessage(socket, JSON.stringify({ type: "hwpod-node-ops-accepted", requestId: message.requestId, nodeId: "pc-host-1" })); registry.handleBunMessage(socket, JSON.stringify({ type: "hwpod-node-ops-result", requestId: message.requestId, nodeId: "pc-host-1", result: { ok: true, status: "completed", planId: message.plan.planId, results: [] } })); }); }, close() {} }; + registry.openBunSocket(socket); + registry.handleBunMessage(socket, JSON.stringify({ type: "register", nodeId: "pc-host-1", maxInFlight: 1 })); + assert.equal((await registry.dispatch(plan, { requestId: "req_failed" }, { timeoutMs: 1000 })).status, "blocked"); + const result = await registry.dispatch(plan, { requestId: "req_retry" }, { timeoutMs: 1000 }); assert.equal(result.status, "completed"); assert.equal(result.planId, plan.planId); + assert.equal(dispatchCount, 2); assert.equal(acceptanceCount, 1); }); +test("WebSocket registry clears pre-accept timeout and disconnect for same-process retry", async () => { + for (const failureMode of ["timeout", "disconnect"] as const) { + const registry = createHwpodNodeWsRegistry({ operationLedger: createMemoryHwpodOperationLedger() }); + const plan = { ...samplePlan(), nodeId: "pc-host-1", planId: `retry-${failureMode}-plan` }; + const failedSocket = { send() {}, close() {} }; + registry.openBunSocket(failedSocket); + registry.handleBunMessage(failedSocket, JSON.stringify({ type: "register", nodeId: "pc-host-1", maxInFlight: 1 })); + const failedPromise = registry.dispatch(plan, { requestId: `req_${failureMode}_failed` }, { timeoutMs: 10 }); + if (failureMode === "disconnect") registry.closeBunSocket(failedSocket); + assert.equal((await failedPromise).status, "blocked"); + + let acceptanceCount = 0; + const retrySocket = { send(value: string) { const message = JSON.parse(value); if (message.type !== "hwpod-node-ops") return; queueMicrotask(() => { acceptanceCount += 1; registry.handleBunMessage(retrySocket, JSON.stringify({ type: "hwpod-node-ops-accepted", requestId: message.requestId, nodeId: "pc-host-1" })); registry.handleBunMessage(retrySocket, JSON.stringify({ type: "hwpod-node-ops-result", requestId: message.requestId, nodeId: "pc-host-1", result: { ok: true, status: "completed", planId: message.plan.planId, results: [] } })); }); }, close() {} }; + registry.openBunSocket(retrySocket); + registry.handleBunMessage(retrySocket, JSON.stringify({ type: "register", nodeId: "pc-host-1", maxInFlight: 1 })); + assert.equal((await registry.dispatch(plan, { requestId: `req_${failureMode}_retry` }, { timeoutMs: 1000 })).status, "completed"); + assert.equal(acceptanceCount, 1); + } +}); + +test("WebSocket registry bounds ledger write failures and permits node replay recovery", async () => { + const records = new Map(); + const memoryLedger = createMemoryHwpodOperationLedger({ records }); + let failComplete = true; + const operationLedger = { ...memoryLedger, async complete(planId: string, result: any) { if (failComplete) { failComplete = false; throw new Error("database unavailable"); } return memoryLedger.complete(planId, result); } }; + const registry = createHwpodNodeWsRegistry({ operationLedger }); + const plan = { ...samplePlan(), nodeId: "pc-host-1", planId: "ledger-replay-plan" }; + let dispatchCount = 0; + const socket = { send(value: string) { const message = JSON.parse(value); if (message.type !== "hwpod-node-ops") return; dispatchCount += 1; queueMicrotask(() => { registry.handleBunMessage(socket, JSON.stringify({ type: "hwpod-node-ops-accepted", requestId: message.requestId, nodeId: "pc-host-1" })); registry.handleBunMessage(socket, JSON.stringify({ type: "hwpod-node-ops-result", requestId: message.requestId, nodeId: "pc-host-1", result: { ok: true, status: "completed", planId: message.plan.planId, replayed: dispatchCount > 1, results: [] } })); }); }, close() {} }; + registry.openBunSocket(socket); + registry.handleBunMessage(socket, JSON.stringify({ type: "register", nodeId: "pc-host-1", maxInFlight: 1 })); + const first = await registry.dispatch(plan, { requestId: "req_ledger_failed" }, { timeoutMs: 1000 }); + assert.equal(first.status, "blocked"); + assert.equal(first.blocker.code, "hwpod_operation_ledger_write_failed"); + const recovered = await registry.dispatch(plan, { requestId: "req_ledger_replay" }, { timeoutMs: 1000 }); + assert.equal(recovered.status, "completed"); + assert.equal(recovered.replayed, true); + assert.equal(dispatchCount, 2); +}); + +test("WebSocket registry bounds accepted-ledger failure without returning hardware success", async () => { + const memoryLedger = createMemoryHwpodOperationLedger(); + let failAccept = true; + const operationLedger = { ...memoryLedger, async accept(planId: string) { if (failAccept) { failAccept = false; throw new Error("database unavailable"); } return memoryLedger.accept(planId); } }; + const registry = createHwpodNodeWsRegistry({ operationLedger }); + const plan = { ...samplePlan(), nodeId: "pc-host-1", planId: "ledger-accept-replay-plan" }; + let dispatchCount = 0; + const socket = { send(value: string) { const message = JSON.parse(value); if (message.type !== "hwpod-node-ops") return; dispatchCount += 1; queueMicrotask(() => { registry.handleBunMessage(socket, JSON.stringify({ type: "hwpod-node-ops-accepted", requestId: message.requestId, nodeId: "pc-host-1" })); registry.handleBunMessage(socket, JSON.stringify({ type: "hwpod-node-ops-result", requestId: message.requestId, nodeId: "pc-host-1", result: { ok: true, status: "completed", planId: message.plan.planId, replayed: dispatchCount > 1, results: [] } })); }); }, close() {} }; + registry.openBunSocket(socket); + registry.handleBunMessage(socket, JSON.stringify({ type: "register", nodeId: "pc-host-1", maxInFlight: 1 })); + const first = await registry.dispatch(plan, { requestId: "req_accept_failed" }, { timeoutMs: 1000 }); + assert.equal(first.status, "blocked"); + assert.equal(first.blocker.code, "hwpod_operation_ledger_write_failed"); + const recovered = await registry.dispatch(plan, { requestId: "req_accept_replay" }, { timeoutMs: 1000 }); + assert.equal(recovered.status, "completed"); + assert.equal(recovered.replayed, true); + assert.equal(dispatchCount, 2); +}); + test("WebSocket registry keeps readiness probes separate and rejects dispatch above maxInFlight", async () => { const sent: any[] = []; const socket = { send(value: string) { sent.push(JSON.parse(value)); }, close() {} }; diff --git a/internal/cloud/hwpod-node-ws-registry.ts b/internal/cloud/hwpod-node-ws-registry.ts index b293e269..2c073221 100644 --- a/internal/cloud/hwpod-node-ws-registry.ts +++ b/internal/cloud/hwpod-node-ws-registry.ts @@ -36,6 +36,8 @@ type PendingDispatch = { operationKind: "user-operation" | "readiness-probe"; operation: Record; timer: ReturnType; + accepted: boolean; + ledgerWrite: Promise; resolve: (value: any) => void; }; @@ -145,24 +147,30 @@ export function createHwpodNodeWsRegistry(options: any = {}) { if (planId) await operationLedger.start({ planId, nodeId, operation }); if (operationKind === "readiness-probe") connection.latestReadinessProbe = operation; else connection.latestOperation = operation; - const promise = new Promise((resolve) => { - const timer = setTimeout(() => { - pending.delete(requestId); - finishPendingOperation({ nodeId, requestId, connection, operationKind, operation, timer, resolve }, "timed-out", now(), "hwpod_node_unavailable"); - resolve(blockedDispatch(plan, requestMeta, `hwpod-node ${nodeId} WebSocket dispatch timed out after ${timeoutMs}ms`)); - }, timeoutMs); - const waiter = { planId, nodeId, requestId, connection, operationKind, operation, timer, resolve } satisfies PendingDispatch; - pending.set(requestId, waiter); - try { - sendJson(connection, { type: "hwpod-node-ops", requestId, plan, requestMeta }); - } catch (error) { - clearTimeout(timer); - pending.delete(requestId); - finishPendingOperation(waiter, "failed", now(), "hwpod_node_unavailable"); - resolve(blockedDispatch(plan, requestMeta, error instanceof Error ? error.message : String(error))); - } - }); + let resolveDispatch: (value: any) => void = () => {}; + const promise = new Promise((resolve) => { resolveDispatch = resolve; }); + const waiter = { + planId, + nodeId, + requestId, + connection, + operationKind, + operation, + timer: null as unknown as ReturnType, + accepted: false, + ledgerWrite: Promise.resolve(true), + resolve: resolveDispatch + } satisfies PendingDispatch; if (planId) operations.set(planId, { planId, nodeId, operation, promise, result: null }); + waiter.timer = setTimeout(() => { + finishTransportFailure(waiter, blockedDispatch(plan, requestMeta, `hwpod-node ${nodeId} WebSocket dispatch timed out after ${timeoutMs}ms`), "timed-out"); + }, timeoutMs); + pending.set(requestId, waiter); + try { + sendJson(connection, { type: "hwpod-node-ops", requestId, plan, requestMeta }); + } catch (error) { + finishTransportFailure(waiter, blockedDispatch(plan, requestMeta, error instanceof Error ? error.message : String(error)), "failed"); + } void promise.then((result) => { const authoritativeOperation = planId ? operations.get(planId) : null; if (authoritativeOperation) { @@ -265,7 +273,16 @@ export function createHwpodNodeWsRegistry(options: any = {}) { } if (messageType === "hwpod-node-ops-accepted") { const waiter = pending.get(safeText(message.requestId)); - if (waiter?.planId) void operationLedger.accept(waiter.planId); + if (waiter?.planId && !waiter.accepted) { + waiter.accepted = true; + waiter.ledgerWrite = waiter.ledgerWrite.then(async () => { + await operationLedger.accept(waiter.planId); + return true; + }).catch((error) => { + finishLedgerFailure(waiter, "accept", error); + return false; + }); + } return; } if (messageType === "hwpod-node-ops-result") void completeDispatch(message); @@ -276,7 +293,6 @@ export function createHwpodNodeWsRegistry(options: any = {}) { const waiter = requestId ? pending.get(requestId) : null; if (!waiter) return; clearTimeout(waiter.timer); - pending.delete(requestId); const result = message.result && typeof message.result === "object" && !Array.isArray(message.result) ? message.result : { ok: false, status: "failed", results: [], blocker: { code: "hwpod_node_result_invalid", layer: "hwpod-node", retryable: true, summary: "hwpod-node WebSocket result payload is invalid" } }; @@ -286,7 +302,14 @@ export function createHwpodNodeWsRegistry(options: any = {}) { now(), safeText(result.blocker?.code) || null ); - if (waiter.planId) await operationLedger.complete(waiter.planId, result); + if (waiter.planId && !(await waiter.ledgerWrite)) return; + try { + if (waiter.planId) await operationLedger.complete(waiter.planId, result); + } catch (error) { + finishLedgerFailure(waiter, "complete", error); + return; + } + pending.delete(requestId); if (waiter.planId) operations.delete(waiter.planId); waiter.resolve(result); } @@ -296,12 +319,27 @@ export function createHwpodNodeWsRegistry(options: any = {}) { for (const [requestId, waiter] of pending) { if (waiter.nodeId !== connection.nodeId) continue; clearTimeout(waiter.timer); - pending.delete(requestId); - finishPendingOperation(waiter, "disconnected", now(), "hwpod_node_unavailable"); - waiter.resolve(blockedDispatch({ nodeId: waiter.nodeId, ops: [] }, { requestId }, `hwpod-node ${waiter.nodeId} WebSocket disconnected before returning result`)); + finishTransportFailure(waiter, blockedDispatch({ nodeId: waiter.nodeId, ops: [] }, { requestId }, `hwpod-node ${waiter.nodeId} WebSocket disconnected before returning result`), "disconnected"); } } + function finishTransportFailure(waiter: PendingDispatch, result: any, status: string) { + clearTimeout(waiter.timer); + pending.delete(waiter.requestId); + finishPendingOperation(waiter, status, now(), "hwpod_node_unavailable"); + if (waiter.planId && !waiter.accepted) operations.delete(waiter.planId); + waiter.resolve(result); + } + + function finishLedgerFailure(waiter: PendingDispatch, phase: "accept" | "complete", error: unknown) { + clearTimeout(waiter.timer); + pending.delete(waiter.requestId); + finishPendingOperation(waiter, "failed", now(), "hwpod_operation_ledger_write_failed"); + if (waiter.planId) operations.delete(waiter.planId); + const summary = `hwpod operation ledger ${phase} failed: ${error instanceof Error ? error.message : String(error)}`; + waiter.resolve(blockedDispatch({ nodeId: waiter.nodeId, ops: [] }, { requestId: waiter.requestId }, summary, "hwpod_operation_ledger_write_failed")); + } + return { openBunSocket, handleBunMessage, closeBunSocket, dispatch, lookup, describe, hasNode }; } diff --git a/internal/harnessrl/harnessrl.test.ts b/internal/harnessrl/harnessrl.test.ts index 1f5c1b63..8ba07cf7 100644 --- a/internal/harnessrl/harnessrl.test.ts +++ b/internal/harnessrl/harnessrl.test.ts @@ -9,6 +9,21 @@ import type { ActivityResult, CaseRunEvent, CaseRunRecord, CaseRunStatus, Harnes import { terminalStatus } from "./contracts.ts"; import { createHarnessRLHttpApp } from "./http.ts"; import { createHarnessRLService } from "./service.ts"; +import { recoverAgentTaskStage } from "../../tools/src/hwlab-caserun-runtime.ts"; + +test("real AgentRun recovery helper maps only result 404 to typed not-found", async () => { + const run: any = { caseId: "case-1", runId: "run-1", runDir: "/tmp/run-1", agentTask: { providerProfile: "default", projectId: "project-1", timeoutMs: 1000, pollIntervalMs: 250 } }; + const context = (status: number): any => ({ + parsed: { _: [], baseUrl: "https://hwlab.example.test", traceId: "trace-authority" }, + env: { HWLAB_API_KEY: "hwl_live_test" }, + fetchImpl: async () => new Response(JSON.stringify({ ok: false, status }), { status, headers: { "content-type": "application/json" } }), + cwd: "/tmp", now: () => "2026-07-17T03:00:00.000Z", sleep: async () => {}, rest: [] + }); + await assert.rejects(recoverAgentTaskStage(context(404), run), (error: any) => error?.code === "agentrun_task_not_found" && error?.status === 404); + const failed = await recoverAgentTaskStage(context(503), run); + assert.equal(failed.agent.stageStatus, "result_request_failed"); + assert.equal(failed.agent.lastHttpStatus, 503); +}); test("API restart preserves PostgreSQL-shaped registry read model", async () => { const fixture = await softwareSmokeFixture(); diff --git a/tools/src/hwlab-caserun-runtime.ts b/tools/src/hwlab-caserun-runtime.ts index 4b494d3e..820ad213 100644 --- a/tools/src/hwlab-caserun-runtime.ts +++ b/tools/src/hwlab-caserun-runtime.ts @@ -823,7 +823,7 @@ export async function recoverAgentTaskStage(context: CaseContext, run: PreparedC promptPath: path.join(run.runDir, "agent-prompt.md"), promptSha256: "" }); - const result = await pollAgentResult(context, { baseUrl, traceId, acceptedBody: { resultUrl }, resultUrl, run, agent: recoveredAgent }); + const result = await pollAgentResult(context, { baseUrl, traceId, acceptedBody: { resultUrl }, resultUrl, run, agent: recoveredAgent, notFoundCode: "agentrun_task_not_found" }); const agent = agentWithResultEvidence({ ...recoveredAgent, stageStatus: result.stageStatus, @@ -919,7 +919,7 @@ async function submitAgentTask(context: CaseContext, input: { baseUrl: string; p }, "agent_task_submit"); } -async function pollAgentResult(context: CaseContext, input: { baseUrl: string; traceId: string; acceptedBody: any; resultUrl?: string; run?: PreparedCaseRun; agent?: AgentRunStage }) { +async function pollAgentResult(context: CaseContext, input: { baseUrl: string; traceId: string; acceptedBody: any; resultUrl?: string; run?: PreparedCaseRun; agent?: AgentRunStage; notFoundCode?: string }) { const requestedTimeoutMs = effectiveAgentTimeoutMs(context, input.run?.agentTask); const timeoutMs = Math.max(Math.min(requestedTimeoutMs, MAX_AGENT_TIMEOUT_MS), 1000); const pollIntervalMs = effectiveAgentPollIntervalMs(context, input.run?.agentTask); @@ -936,11 +936,14 @@ async function pollAgentResult(context: CaseContext, input: { baseUrl: string; t const response = await caseFetchJson(context, resultUrl, { method: "GET", headers: caseAgentHeaders(context) }, "agent_task_result"); lastBody = response.body; lastStatus = response.status; + if (!response.ok && response.status === 404 && input.notFoundCode) { + throw Object.assign(new Error(`AgentRun task ${input.traceId} was not found`), { code: input.notFoundCode, traceId: input.traceId, status: response.status }); + } + if (!response.ok) return { stageStatus: "result_request_failed", body: lastBody, polls, timedOut: false, lastPollAt, lastHttpStatus: lastStatus, error: response.error }; if (input.run && input.agent) { const runningAgent = agentWithResultEvidence({ ...input.agent, stageStatus: "running", polls, timeoutMs, pollIntervalMs, resultUrl, lastPollAt, lastHttpStatus: lastStatus }, lastBody, input.run.agentTask); await updateRun(context, input.run, { stage: "agent-running", agent: runningAgent }); } - if (!response.ok) return { stageStatus: "result_request_failed", body: lastBody, polls, timedOut: false, lastPollAt, lastHttpStatus: lastStatus, error: response.error }; const terminal = agentTerminalEvidence(lastBody); const status = text(lastBody?.status); if ((status && status !== "running") || terminal.terminal) {