fix: 修复 HarnessRL 真实重试状态机
This commit is contained in:
root
@@ -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() {} };
|
||||
|
||||
@@ -36,6 +36,8 @@ type PendingDispatch = {
|
||||
operationKind: "user-operation" | "readiness-probe";
|
||||
operation: Record<string, unknown>;
|
||||
timer: ReturnType<typeof setTimeout>;
|
||||
accepted: boolean;
|
||||
ledgerWrite: Promise<boolean>;
|
||||
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<typeof setTimeout>,
|
||||
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 };
|
||||
}
|
||||
|
||||
|
||||
@@ -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();
|
||||
|
||||
Reference in New Issue
Block a user