mirrors/pikasTech-HWLAB
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: route native caserun service dependencies

Browse Source
This commit is contained in:
root
2026-07-24 16:12:47 +02:00
parent 87d725a2a5
commit fb885ac26b
6 changed files with 198 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/harnessrl/activities.ts
+28 -8
View File
@@ -1,7 +1,8 @@
// SPEC: PJ2026-010306 嵌入式AgentBench.
// SPEC_VERSION: draft-2026-07-24-p0-l0-l1-arm2d.
// SPEC_VERSION: draft-2026-07-24-p2-l0-l1-arm2d.
import { heartbeat } from "@temporalio/activity";
import { ApplicationFailure } from "@temporalio/common";
import { createHash } from "node:crypto";
import { mkdir, readFile, writeFile } from "node:fs/promises";
import path from "node:path";
@@ -150,14 +151,26 @@ export function createHarnessRLActivities(options: {
},
async markCompleted({ runId }: { runId: string }) { await transition(registry, runId, "completed", "completed", {}); },
async markCanceled({ runId }: { runId: string }) { await transition(registry, runId, "canceled", "canceled", {}); },
async markFailed(input: { runId: string; code: string; message: string }) {
await registry.updateRun(input.runId, { status: "failed", stage: "failed", error: { code: input.code, message: input.message } });
await registry.appendEvent(input.runId, "failed", "failed", { code: input.code, message: input.message.slice(0, 500) });
async markFailed(input: { runId: string; code: string; message: string; details?: unknown }) {
const error = { code: input.code, message: input.message, ...(input.details === undefined ? {} : { details: input.details }) };
await registry.updateRun(input.runId, { status: "failed", stage: "failed", error });
await registry.appendEvent(input.runId, "failed", "failed", { ...error, message: input.message.slice(0, 500) });
}
};
}
async function idempotent(registry: HarnessRLRegistry, input: { runId: string; identity: string }, execute: () => Promise<Record<string, unknown>>, recover?: () => Promise<Record<string, unknown>>) {
try {
return await idempotentActivity(registry, input, execute, recover);
} catch (error: any) {
if (error instanceof ApplicationFailure) throw error;
const code = String(error?.code ?? error?.name ?? "caserun_activity_failed");
const details = error?.details === undefined ? [] : [error.details];
throw ApplicationFailure.create({ message: error?.message ?? String(error), type: code, nonRetryable: false, details });
}
}
async function idempotentActivity(registry: HarnessRLRegistry, input: { runId: string; identity: string }, execute: () => Promise<Record<string, unknown>>, recover?: () => Promise<Record<string, unknown>>) {
const activityName = input.identity.split(":").at(-2) ?? "activity";
const existing = await registry.getActivityResult(input.runId, activityName, input.identity);
if ((existing?.output as any)?.state === "completed") return (existing!.output as any).result;
@@ -184,17 +197,24 @@ async function transition(registry: HarnessRLRegistry, runId: string, status: an
}
function contextForRecord(record: any, env: Record<string, string | undefined>, identity: string) {
const apiUrl = String(env.HWLAB_CASERUN_INTERNAL_API_URL ?? env.HWLAB_RUNTIME_INTERNAL_API_URL ?? record.runtimeApiUrl).replace(/\/+$/u, "");
const hwpodApiUrl = requiredServiceUrl(env.HWLAB_CASERUN_HWPOD_API_URL, "HWLAB_CASERUN_HWPOD_API_URL");
const workbenchApiUrl = requiredServiceUrl(env.HWLAB_CASERUN_WORKBENCH_API_URL, "HWLAB_CASERUN_WORKBENCH_API_URL");
const traceId = `trc_harnessrl_${createHash("sha256").update(identity).digest("hex").slice(0, 24)}`;
const caseSnapshotDir = requiredCaseSnapshot(record);
return {
parsed: { _: [], caseId: record.caseId, runId: record.runId, runDir: record.runDir, caseRepo: record.caseRepo, caseDefinitionRepo: caseSnapshotDir, apiUrl, traceId, operationIdentity: identity, noCaseRepoRecord: false, stateDir: path.join(path.dirname(record.runDir), "cli") },
env: { ...process.env, ...env, HWLAB_CASE_REPO: record.caseRepo, HWLAB_RUNTIME_API_URL: apiUrl, HWLAB_CASERUN_PUBLIC_RUNTIME_API_URL: record.runtimeApiUrl, HWLAB_RUNTIME_ENDPOINT_LOCKED: "1", HWLAB_HWPOD_OPERATION_IDENTITY: identity },
fetchImpl: caseRunInternalFetch(globalThis.fetch, apiUrl, env), cwd: record.caseRepo,
parsed: { _: [], caseId: record.caseId, runId: record.runId, runDir: record.runDir, caseRepo: record.caseRepo, caseDefinitionRepo: caseSnapshotDir, apiUrl: hwpodApiUrl, webUrl: workbenchApiUrl, traceId, operationIdentity: identity, noCaseRepoRecord: false, stateDir: path.join(path.dirname(record.runDir), "cli") },
env: { ...process.env, ...env, HWLAB_CASE_REPO: record.caseRepo, HWLAB_RUNTIME_API_URL: hwpodApiUrl, HWLAB_RUNTIME_WEB_URL: workbenchApiUrl, HWLAB_CASERUN_PUBLIC_RUNTIME_API_URL: record.runtimeApiUrl, HWLAB_RUNTIME_ENDPOINT_LOCKED: "1", HWLAB_HWPOD_OPERATION_IDENTITY: identity },
fetchImpl: caseRunInternalFetch(globalThis.fetch, hwpodApiUrl, env), cwd: record.caseRepo,
now: () => new Date().toISOString(), sleep: (ms: number) => new Promise((resolve) => setTimeout(resolve, ms)), argv: [], rest: ["worker", record.caseId]
};
}
function requiredServiceUrl(value: string | undefined, name: string) {
const normalized = String(value ?? "").trim().replace(/\/+$/u, "");
if (!normalized) throw Object.assign(new Error(`${name} is required from the owning YAML native service dependencies`), { code: "caserun_service_dependency_required", details: { name } });
return normalized;
}
function requiredCaseSnapshot(record: any) {
const value = String(record?.caseSnapshotDir ?? "").trim();
if (!value || !path.isAbsolute(value)) {
internal/harnessrl/contracts.ts
+1 -1
View File
@@ -1,5 +1,5 @@
// SPEC: PJ2026-010306 嵌入式AgentBench.
// SPEC_VERSION: draft-2026-07-24-p0-l0-l1-arm2d.
// SPEC_VERSION: draft-2026-07-24-p2-l0-l1-arm2d.
export type CaseRunStatus = "queued" | "running" | "completed" | "failed" | "blocked" | "cancel_requested" | "canceled";
internal/harnessrl/harnessrl.test.ts
+46 -3
View File
@@ -5,6 +5,7 @@ import path from "node:path";
import { test } from "node:test";
import { createHarnessRLActivities } from "./activities.ts";
import { caseRunWorkflowFailure } from "./workflows.ts";
import type { ActivityResult, CaseRunEvent, CaseRunRecord, CaseRunStatus, HarnessRLRegistry, HarnessRLTemporalGateway, StartCaseRunInput } from "./contracts.ts";
import { terminalStatus } from "./contracts.ts";
import { createHarnessRLHttpApp } from "./http.ts";
@@ -155,9 +156,12 @@ test("worker uses configured cloud-api authority and resubmits the same identity
let recoverCount = 0;
const activities = createHarnessRLActivities({
registry,
env: { HWLAB_CASERUN_INTERNAL_API_URL: "http://hwlab-cloud-api.hwlab-v03.svc.cluster.local:6667" },
env: {
HWLAB_CASERUN_HWPOD_API_URL: "http://127.0.0.1:6681",
HWLAB_CASERUN_WORKBENCH_API_URL: "http://127.0.0.1:6677",
},
stages: {
agent: async (context: any, run: any) => { submitCount += 1; assert.equal(context.parsed.apiUrl, "http://hwlab-cloud-api.hwlab-v03.svc.cluster.local:6667"); assert.equal(context.parsed.operationIdentity, identity); return { run, agent: { traceId: "trace-authority-retry" } }; },
agent: async (context: any, run: any) => { submitCount += 1; assert.equal(context.parsed.apiUrl, "http://127.0.0.1:6681"); assert.equal(context.parsed.webUrl, "http://127.0.0.1:6677"); assert.equal(context.parsed.operationIdentity, identity); return { run, agent: { traceId: "trace-authority-retry" } }; },
recoverAgent: async () => { recoverCount += 1; throw Object.assign(new Error("not found"), { code: "agentrun_task_not_found" }); }
}
});
@@ -169,6 +173,45 @@ test("worker uses configured cloud-api authority and resubmits the same identity
assert.match(deploy, /HWLAB_CASERUN_INTERNAL_API_URL: http:\/\/hwlab-cloud-api\.hwlab-v03\.svc\.cluster\.local:6667/u);
});
test("Temporal activity wrappers preserve the underlying typed CaseRun failure", () => {
const failure = caseRunWorkflowFailure({
name: "ActivityFailure",
message: "Activity task failed",
activityType: "prepare",
cause: {
name: "ApplicationFailure",
type: "subject_worktree_prepare_failed",
message: "failed to prepare isolated subject worktree on the HWPOD node",
details: [{ httpStatus: 404, blocker: { code: "hwpod_route_not_found" } }],
},
});
assert.deepEqual(failure, {
code: "subject_worktree_prepare_failed",
message: "failed to prepare isolated subject worktree on the HWPOD node",
details: { httpStatus: 404, blocker: { code: "hwpod_route_not_found" } },
});
});
test("activity errors become retryable ApplicationFailure values with typed details", async () => {
const fixture = await hardwareFixture();
const registry = new MemoryRegistry();
const runId = "typed-activity-failure";
await registry.createRun({ runId, caseId: fixture.caseId, workflowId: `workflow-${runId}`, caseAuthority: fixture.authority, launchContext: null, caseSnapshotDir: fixture.root, caseRepo: fixture.root, runDir: path.join(fixture.stateRoot, "runs", runId), runtimeApiUrl: "http://127.0.0.1:4316" });
const worker = createHarnessRLActivities({
registry,
env: { HWLAB_CASERUN_HWPOD_API_URL: "http://127.0.0.1:6681", HWLAB_CASERUN_WORKBENCH_API_URL: "http://127.0.0.1:6677" },
stages: {
prepare: async () => { throw Object.assign(new Error("HWPOD route failed"), { code: "subject_worktree_prepare_failed", details: { httpStatus: 404 } }); },
},
});
await assert.rejects(worker.prepare({ runId, identity: `${runId}:prepare:v1` }), (error: any) => {
assert.equal(error.type, "subject_worktree_prepare_failed");
assert.equal(error.nonRetryable, false);
assert.deepEqual(error.details, [{ httpStatus: 404 }]);
return true;
});
});
test("AgentRun and HWPOD crash gaps reuse stable authoritative operation identities", async () => {
const fixture = await hardwareFixture();
const registry = new MemoryRegistry();
@@ -208,7 +251,7 @@ test("AgentRun and HWPOD crash gaps reuse stable authoritative operation identit
return { run, status: "completed", summary: { artifactManifestPath: manifestPath }, evidence };
}
};
const worker = createHarnessRLActivities({ registry, env: {}, stages });
const worker = createHarnessRLActivities({ registry, env: { HWLAB_CASERUN_HWPOD_API_URL: "http://127.0.0.1:6681", HWLAB_CASERUN_WORKBENCH_API_URL: "http://127.0.0.1:6677" }, stages });
await worker.prepare({ runId: "authority-gap", identity: "authority-gap:prepare:v1" });
await assert.rejects(worker.agent({ runId: "authority-gap", identity: "authority-gap:agent:v1" }), /AgentRun accepted/);
await Promise.all([
internal/harnessrl/workflows.ts
+18 -2
View File
@@ -10,7 +10,7 @@ type Activities = {
collect(input: { runId: string; identity: string }): Promise<Record<string, unknown>>;
markCompleted(input: { runId: string }): Promise<void>;
markCanceled(input: { runId: string }): Promise<void>;
markFailed(input: { runId: string; code: string; message: string }): Promise<void>;
markFailed(input: { runId: string; code: string; message: string; details?: unknown }): Promise<void>;
};
export const cancelCaseRun = defineSignal("cancelCaseRun");
@@ -46,7 +46,23 @@ export async function caseRunWorkflow(input: { runId: string; activity: { startT
await activities.markCompleted({ runId: input.runId });
return { status: "completed", runId: input.runId };
} catch (error: any) {
await activities.markFailed({ runId: input.runId, code: error?.code ?? error?.name ?? "caserun_failed", message: error?.message ?? String(error) });
await activities.markFailed({ runId: input.runId, ...caseRunWorkflowFailure(error) });
throw error;
}
}
export function caseRunWorkflowFailure(error: any) {
let current = error;
let fallback = { code: "caserun_failed", message: String(error), details: undefined as unknown };
for (let depth = 0; current && depth < 8; depth += 1) {
const code = String(current.code ?? current.type ?? current.name ?? "").trim();
const message = String(current.message ?? "").trim();
const details = Array.isArray(current.details) && current.details.length === 1 ? current.details[0] : current.details;
if (code || message) fallback = { code: code || fallback.code, message: message || fallback.message, details };
if (code && !["ActivityFailure", "ApplicationFailure", "Error"].includes(code) && message !== "Activity task failed") {
return { code, message: message || code, ...(details === undefined ? {} : { details }) };
}
current = current.cause;
}
return { code: fallback.code, message: fallback.message, ...(fallback.details === undefined ? {} : { details: fallback.details }) };
}
package-lock.json
Generated
+104 -1
View File
@@ -12,6 +12,7 @@
"@swc/wasm": "^1.15.43",
"@temporalio/activity": "^1.20.3",
"@temporalio/client": "^1.20.3",
"@temporalio/common": "^1.21.1",
"@temporalio/worker": "^1.20.3",
"@temporalio/workflow": "^1.20.3",
"fzstd": "0.1.1",
@@ -914,6 +915,21 @@
"node": ">= 20.3.0"
}
},
"node_modules/@temporalio/activity/node_modules/@temporalio/common": {
"version": "1.20.3",
"resolved": "https://registry.npmjs.org/@temporalio/common/-/common-1.20.3.tgz",
"integrity": "sha512-VM9GyUAa7tl413nM24mcBhUfUeqcGazmDkx/GfDA0c+KuEVusclMg9eqa63EDv/rguC/YhWaXu6q5Kl8+99GSw==",
"dependencies": {
"@temporalio/proto": "1.20.3",
"long": "^5.2.3",
"ms": "3.0.0-canary.1",
"nexus-rpc": "^0.0.2",
"proto3-json-serializer": "^2.0.0"
},
"engines": {
"node": ">= 20.3.0"
}
},
"node_modules/@temporalio/client": {
"version": "1.20.3",
"resolved": "https://registry.npmjs.org/@temporalio/client/-/client-1.20.3.tgz",
@@ -931,7 +947,7 @@
"node": ">= 20.3.0"
}
},
"node_modules/@temporalio/common": {
"node_modules/@temporalio/client/node_modules/@temporalio/common": {
"version": "1.20.3",
"resolved": "https://registry.npmjs.org/@temporalio/common/-/common-1.20.3.tgz",
"integrity": "sha512-VM9GyUAa7tl413nM24mcBhUfUeqcGazmDkx/GfDA0c+KuEVusclMg9eqa63EDv/rguC/YhWaXu6q5Kl8+99GSw==",
@@ -946,6 +962,33 @@
"node": ">= 20.3.0"
}
},
"node_modules/@temporalio/common": {
"version": "1.21.1",
"resolved": "https://registry.npmjs.org/@temporalio/common/-/common-1.21.1.tgz",
"integrity": "sha512-8Pis59xYLrGu6GfkkWvrYWkSPEvo8lBBILsR6gBHGbymNlEc0ynylRXqPmRjwuLfYS7jHzxqjjO7cvXJQ/z2Fg==",
"dependencies": {
"@temporalio/proto": "1.21.1",
"long": "^5.2.3",
"ms": "3.0.0-canary.1",
"nexus-rpc": "^0.0.2",
"proto3-json-serializer": "^2.0.0"
},
"engines": {
"node": ">= 20.3.0"
}
},
"node_modules/@temporalio/common/node_modules/@temporalio/proto": {
"version": "1.21.1",
"resolved": "https://registry.npmjs.org/@temporalio/proto/-/proto-1.21.1.tgz",
"integrity": "sha512-eSHGrZ6CxbtjrAzxiMgKrWeDiBlWk6/JkIqsB1hrkPB6TQXC67Az8v0BL0Fj3ur8ktQZbaacON/xCDjTsvJyGw==",
"dependencies": {
"long": "^5.2.3",
"protobufjs": "^7.6.4"
},
"engines": {
"node": ">= 20.3.0"
}
},
"node_modules/@temporalio/core-bridge": {
"version": "1.20.3",
"resolved": "https://registry.npmjs.org/@temporalio/core-bridge/-/core-bridge-1.20.3.tgz",
@@ -958,6 +1001,21 @@
"node": ">= 20.3.0"
}
},
"node_modules/@temporalio/core-bridge/node_modules/@temporalio/common": {
"version": "1.20.3",
"resolved": "https://registry.npmjs.org/@temporalio/common/-/common-1.20.3.tgz",
"integrity": "sha512-VM9GyUAa7tl413nM24mcBhUfUeqcGazmDkx/GfDA0c+KuEVusclMg9eqa63EDv/rguC/YhWaXu6q5Kl8+99GSw==",
"dependencies": {
"@temporalio/proto": "1.20.3",
"long": "^5.2.3",
"ms": "3.0.0-canary.1",
"nexus-rpc": "^0.0.2",
"proto3-json-serializer": "^2.0.0"
},
"engines": {
"node": ">= 20.3.0"
}
},
"node_modules/@temporalio/nexus": {
"version": "1.20.3",
"resolved": "https://registry.npmjs.org/@temporalio/nexus/-/nexus-1.20.3.tgz",
@@ -973,6 +1031,21 @@
"node": ">= 20.3.0"
}
},
"node_modules/@temporalio/nexus/node_modules/@temporalio/common": {
"version": "1.20.3",
"resolved": "https://registry.npmjs.org/@temporalio/common/-/common-1.20.3.tgz",
"integrity": "sha512-VM9GyUAa7tl413nM24mcBhUfUeqcGazmDkx/GfDA0c+KuEVusclMg9eqa63EDv/rguC/YhWaXu6q5Kl8+99GSw==",
"dependencies": {
"@temporalio/proto": "1.20.3",
"long": "^5.2.3",
"ms": "3.0.0-canary.1",
"nexus-rpc": "^0.0.2",
"proto3-json-serializer": "^2.0.0"
},
"engines": {
"node": ">= 20.3.0"
}
},
"node_modules/@temporalio/proto": {
"version": "1.20.3",
"resolved": "https://registry.npmjs.org/@temporalio/proto/-/proto-1.20.3.tgz",
@@ -1015,6 +1088,21 @@
"node": ">= 20.3.0"
}
},
"node_modules/@temporalio/worker/node_modules/@temporalio/common": {
"version": "1.20.3",
"resolved": "https://registry.npmjs.org/@temporalio/common/-/common-1.20.3.tgz",
"integrity": "sha512-VM9GyUAa7tl413nM24mcBhUfUeqcGazmDkx/GfDA0c+KuEVusclMg9eqa63EDv/rguC/YhWaXu6q5Kl8+99GSw==",
"dependencies": {
"@temporalio/proto": "1.20.3",
"long": "^5.2.3",
"ms": "3.0.0-canary.1",
"nexus-rpc": "^0.0.2",
"proto3-json-serializer": "^2.0.0"
},
"engines": {
"node": ">= 20.3.0"
}
},
"node_modules/@temporalio/workflow": {
"version": "1.20.3",
"resolved": "https://registry.npmjs.org/@temporalio/workflow/-/workflow-1.20.3.tgz",
@@ -1028,6 +1116,21 @@
"node": ">= 20.3.0"
}
},
"node_modules/@temporalio/workflow/node_modules/@temporalio/common": {
"version": "1.20.3",
"resolved": "https://registry.npmjs.org/@temporalio/common/-/common-1.20.3.tgz",
"integrity": "sha512-VM9GyUAa7tl413nM24mcBhUfUeqcGazmDkx/GfDA0c+KuEVusclMg9eqa63EDv/rguC/YhWaXu6q5Kl8+99GSw==",
"dependencies": {
"@temporalio/proto": "1.20.3",
"long": "^5.2.3",
"ms": "3.0.0-canary.1",
"nexus-rpc": "^0.0.2",
"proto3-json-serializer": "^2.0.0"
},
"engines": {
"node": ">= 20.3.0"
}
},
"node_modules/@types/estree": {
"version": "1.0.9",
"resolved": "https://registry.npmjs.org/@types/estree/-/estree-1.0.9.tgz",
package.json
+1
View File
@@ -67,6 +67,7 @@
"@swc/wasm": "^1.15.43",
"@temporalio/activity": "^1.20.3",
"@temporalio/client": "^1.20.3",
"@temporalio/common": "^1.21.1",
"@temporalio/worker": "^1.20.3",
"@temporalio/workflow": "^1.20.3",
"fzstd": "0.1.1",