diff --git a/.tekton/hwlab-nc01-v03-pac.yaml b/.tekton/hwlab-nc01-v03-pac.yaml index 9e61d484..eefc0f59 100644 --- a/.tekton/hwlab-nc01-v03-pac.yaml +++ b/.tekton/hwlab-nc01-v03-pac.yaml @@ -77,7 +77,7 @@ spec: - name: registry-prefix value: 127.0.0.1:5000/hwlab - name: services - value: "hwlab-cloud-api,hwlab-workbench-runtime,hwlab-user-billing,hwlab-project-management,hwlab-tasktree-api,hwlab-tasktree-worker,hwlab-cloud-web,hwlab-gateway,hwlab-edge-proxy,hwlab-agent-skills" + value: "hwlab-cloud-api,hwlab-workbench-runtime,hwlab-user-billing,hwlab-project-management,hwlab-tasktree-api,hwlab-tasktree-worker,hwlab-harnessrl-api,hwlab-harnessrl-worker,hwlab-cloud-web,hwlab-gateway,hwlab-edge-proxy,hwlab-agent-skills" - name: base-image value: 127.0.0.1:5000/hwlab/hwlab-node20-base:20-bookworm-slim - name: build-cache-mode diff --git a/cmd/hwlab-harnessrl-api/main.ts b/cmd/hwlab-harnessrl-api/main.ts new file mode 100755 index 00000000..cdc17c42 --- /dev/null +++ b/cmd/hwlab-harnessrl-api/main.ts @@ -0,0 +1,12 @@ +#!/usr/bin/env bun +import { createHarnessRLHttpApp } from "../../internal/harnessrl/http.ts"; +import { harnessRLRuntime } from "../../internal/harnessrl/runtime.ts"; + +const runtime = harnessRLRuntime(); +await runtime.registry.ensureSchema(); +const app = createHarnessRLHttpApp({ service: runtime.service }); +const host = process.env.HARNESSRL_API_HOST || "0.0.0.0"; +const port = Number.parseInt(process.env.HARNESSRL_API_PORT || process.env.PORT || "6675", 10); +const server = Bun.serve({ hostname: host, port, fetch: (request) => app.fetch(request) }); +process.stdout.write(`${JSON.stringify({ serviceId: "hwlab-harnessrl-api", status: "listening", host, port: server.port })}\n`); +for (const signal of ["SIGINT", "SIGTERM"] as const) process.once(signal, async () => { server.stop(true); await app.close(); process.exit(0); }); diff --git a/cmd/hwlab-harnessrl-worker/main.ts b/cmd/hwlab-harnessrl-worker/main.ts new file mode 100755 index 00000000..f744f83c --- /dev/null +++ b/cmd/hwlab-harnessrl-worker/main.ts @@ -0,0 +1,18 @@ +#!/usr/bin/env bun +import { NativeConnection, Worker } from "@temporalio/worker"; +import path from "node:path"; + +import { createHarnessRLActivities } from "../../internal/harnessrl/activities.ts"; +import { harnessRLRuntime } from "../../internal/harnessrl/runtime.ts"; + +const runtime = harnessRLRuntime(); +await runtime.registry.ensureSchema(); +const connection = await NativeConnection.connect({ address: runtime.temporalAddress }); +const worker = await Worker.create({ connection, namespace: runtime.temporalNamespace, taskQueue: runtime.taskQueue, + workflowsPath: path.resolve(import.meta.dir, "../../internal/harnessrl/workflows.ts"), activities: createHarnessRLActivities({ registry: runtime.registry }) }); +const healthPort = Number.parseInt(process.env.HARNESSRL_WORKER_HEALTH_PORT || "6676", 10); +const health = Bun.serve({ hostname: process.env.HARNESSRL_WORKER_HEALTH_HOST || "0.0.0.0", port: healthPort, + fetch(request) { const pathname = new URL(request.url).pathname; return pathname === "/healthz" || pathname === "/readyz" || pathname === "/health/live" || pathname === "/health/ready" ? Response.json({ ok: true, serviceId: "hwlab-harnessrl-worker", namespace: runtime.temporalNamespace, taskQueue: runtime.taskQueue }) : new Response("not found", { status: 404 }); } }); +process.stdout.write(`${JSON.stringify({ serviceId: "hwlab-harnessrl-worker", status: "started", namespace: runtime.temporalNamespace, taskQueue: runtime.taskQueue, healthPort: health.port })}\n`); +for (const signal of ["SIGINT", "SIGTERM"] as const) process.once(signal, () => worker.shutdown()); +try { await worker.run(); } finally { health.stop(true); await connection.close(); await runtime.registry.close(); } diff --git a/config/hwlab-v03/harnessrl.private.yaml b/config/hwlab-v03/harnessrl.private.yaml new file mode 100644 index 00000000..fcecff52 --- /dev/null +++ b/config/hwlab-v03/harnessrl.private.yaml @@ -0,0 +1,20 @@ +apiVersion: hwlab.pikastech.local/v1alpha1 +kind: HarnessRLConsumerConfig +metadata: + name: harnessrl-v03 +spec: + api: + serviceId: hwlab-harnessrl-api + port: 6675 + worker: + serviceId: hwlab-harnessrl-worker + healthPort: 6676 + registry: + databaseUrl: secretRef:hwlab-harnessrl-db/database-url + temporal: + address: temporal-frontend.temporal.svc.cluster.local:7233 + namespace: unidesk + taskQueue: hwlab-v03-harnessrl + consumer: + cloudApiEnv: + HARNESSRL_API_URL: http://hwlab-harnessrl-api.hwlab-v03.svc.cluster.local:6675 diff --git a/deploy/deploy.yaml b/deploy/deploy.yaml index 223922d7..50b9f462 100644 --- a/deploy/deploy.yaml +++ b/deploy/deploy.yaml @@ -508,6 +508,8 @@ lanes: - hwlab-project-management - hwlab-tasktree-api - hwlab-tasktree-worker + - hwlab-harnessrl-api + - hwlab-harnessrl-worker - hwlab-cloud-web - hwlab-gateway - hwlab-edge-proxy @@ -519,6 +521,8 @@ lanes: hwlab-project-management: deploy/runtime/boot/hwlab-project-management.sh hwlab-tasktree-api: deploy/runtime/boot/hwlab-tasktree-api.sh hwlab-tasktree-worker: deploy/runtime/boot/hwlab-tasktree-worker.sh + hwlab-harnessrl-api: deploy/runtime/boot/hwlab-harnessrl-api.sh + hwlab-harnessrl-worker: deploy/runtime/boot/hwlab-harnessrl-worker.sh hwlab-cloud-web: deploy/runtime/boot/hwlab-cloud-web.sh hwlab-gateway: deploy/runtime/boot/hwlab-gateway.sh hwlab-edge-proxy: deploy/runtime/boot/hwlab-edge-proxy.sh @@ -665,6 +669,38 @@ lanes: env: TZ: Asia/Shanghai observable: true + hwlab-harnessrl-api: + runtimeKind: bun-command + entrypoint: cmd/hwlab-harnessrl-api/main.ts + artifactKind: bun-command + healthPath: /readyz + healthPort: 6675 + componentPaths: + - cmd/hwlab-harnessrl-api/ + - internal/harnessrl/ + - internal/cloud/server-caserun-http.ts + - package.json + - package-lock.json + - deploy/runtime/boot/hwlab-harnessrl-api.sh + env: + TZ: Asia/Shanghai + observable: true + hwlab-harnessrl-worker: + runtimeKind: bun-command + entrypoint: cmd/hwlab-harnessrl-worker/main.ts + artifactKind: bun-command + healthPath: /readyz + healthPort: 6676 + componentPaths: + - cmd/hwlab-harnessrl-worker/ + - internal/harnessrl/ + - tools/src/hwlab-caserun-lib.ts + - package.json + - package-lock.json + - deploy/runtime/boot/hwlab-harnessrl-worker.sh + env: + TZ: Asia/Shanghai + observable: true hwlab-cloud-web: runtimeKind: cloud-web entrypoint: web/hwlab-cloud-web/index.html @@ -893,6 +929,7 @@ lanes: HWLAB_WORKBENCH_RUNTIME_URL: http://hwlab-workbench-runtime.hwlab-v03.svc.cluster.local:6671 HWLAB_PROJECT_MANAGEMENT_URL: http://hwlab-project-management.hwlab-v03.svc.cluster.local:6672 HWLAB_TASKTREE_URL: http://hwlab-tasktree-api.hwlab-v03.svc.cluster.local:6673 + HARNESSRL_API_URL: http://hwlab-harnessrl-api.hwlab-v03.svc.cluster.local:6675 HWLAB_HWPOD_SPEC_REGISTRY_DIRS: /etc/hwlab/hwpod-specs HWLAB_HWPOD_NODE_WS_TOKEN: secretRef:hwlab-v03-hwpod-node-auth/token HWLAB_USER_BILLING_LOGIN_RETRY_ATTEMPTS: "2" @@ -1018,7 +1055,31 @@ lanes: TASKTREE_TEMPORAL_TASK_QUEUE: hwlab-v03-tasktree TASKTREE_WORKER_HEALTH_PORT: "6674" OTEL_SERVICE_NAME: hwlab-tasktree-worker + - serviceId: hwlab-harnessrl-api + replicas: 1 + env: + HARNESSRL_DATABASE_URL: secretRef:hwlab-harnessrl-db/database-url + HARNESSRL_TEMPORAL_ADDRESS: temporal-frontend.temporal.svc.cluster.local:7233 + HARNESSRL_TEMPORAL_NAMESPACE: unidesk + HARNESSRL_TEMPORAL_TASK_QUEUE: hwlab-v03-harnessrl + HARNESSRL_API_PORT: "6675" + OTEL_SERVICE_NAME: hwlab-harnessrl-api + - serviceId: hwlab-harnessrl-worker + replicas: 1 + env: + HARNESSRL_DATABASE_URL: secretRef:hwlab-harnessrl-db/database-url + HARNESSRL_TEMPORAL_ADDRESS: temporal-frontend.temporal.svc.cluster.local:7233 + HARNESSRL_TEMPORAL_NAMESPACE: unidesk + HARNESSRL_TEMPORAL_TASK_QUEUE: hwlab-v03-harnessrl + HARNESSRL_WORKER_HEALTH_PORT: "6676" + OTEL_SERVICE_NAME: hwlab-harnessrl-worker services: + - serviceId: hwlab-harnessrl-api + namespace: hwlab-dev + healthPath: /health/live + - serviceId: hwlab-harnessrl-worker + namespace: hwlab-dev + healthPath: /health/live - serviceId: hwlab-tasktree-api namespace: hwlab-dev healthPath: /health/live diff --git a/deploy/runtime/boot/hwlab-harnessrl-api.sh b/deploy/runtime/boot/hwlab-harnessrl-api.sh new file mode 100755 index 00000000..7b1b23f7 --- /dev/null +++ b/deploy/runtime/boot/hwlab-harnessrl-api.sh @@ -0,0 +1,9 @@ +#!/bin/sh +set -eu + +export HWLAB_SERVICE_ID=hwlab-harnessrl-api +export HWLAB_SERVICE_ENTRYPOINT=cmd/hwlab-harnessrl-api/main.ts +export HARNESSRL_API_PORT="${HARNESSRL_API_PORT:-6675}" +export PORT="$HARNESSRL_API_PORT" + +exec "${HWLAB_BUN_COMMAND:-bun}" run cmd/hwlab-harnessrl-api/main.ts diff --git a/deploy/runtime/boot/hwlab-harnessrl-worker.sh b/deploy/runtime/boot/hwlab-harnessrl-worker.sh new file mode 100755 index 00000000..9075587d --- /dev/null +++ b/deploy/runtime/boot/hwlab-harnessrl-worker.sh @@ -0,0 +1,9 @@ +#!/bin/sh +set -eu + +export HWLAB_SERVICE_ID=hwlab-harnessrl-worker +export HWLAB_SERVICE_ENTRYPOINT=cmd/hwlab-harnessrl-worker/main.ts +export HARNESSRL_WORKER_HEALTH_PORT="${HARNESSRL_WORKER_HEALTH_PORT:-6676}" +export PORT="$HARNESSRL_WORKER_HEALTH_PORT" + +exec "${HWLAB_BUN_COMMAND:-bun}" run cmd/hwlab-harnessrl-worker/main.ts diff --git a/internal/cloud/server-caserun-http.test.ts b/internal/cloud/server-caserun-http.test.ts index 3af95938..649c9209 100644 --- a/internal/cloud/server-caserun-http.test.ts +++ b/internal/cloud/server-caserun-http.test.ts @@ -1,14 +1,10 @@ import assert from "node:assert/strict"; -import { mkdir, mkdtemp, readFile, rm, writeFile } from "node:fs/promises"; -import os from "node:os"; -import path from "node:path"; -import { test } from "bun:test"; +import { createServer } from "node:http"; +import { test } from "node:test"; -import { createCloudApiServer } from "./server.ts"; -import { caseRunInternalFetch } from "./server-caserun-http.ts"; -import { caseValidationPlanFromDefinition, hwpodCommandForValidationStep } from "../../tools/src/hwlab-caserun-validation.ts"; +import { caseRunInternalFetch, handleCaseRunHttp } from "./server-caserun-http.ts"; -test("CaseRun internal requests reuse the existing bootstrap admin API key only for the internal origin", async () => { +test("CaseRun internal requests reuse bootstrap key only for the internal origin", async () => { const calls: Array<{ url: string; authorization: string | null }> = []; const fetchImpl = async (input: URL | RequestInfo, init: RequestInit = {}) => { const headers = new Headers(init.headers); @@ -16,321 +12,52 @@ test("CaseRun internal requests reuse the existing bootstrap admin API key only return new Response("{}", { status: 200, headers: { "content-type": "application/json" } }); }; const internalFetch = caseRunInternalFetch(fetchImpl as typeof fetch, "http://hwlab-cloud-api.hwlab-v03.svc.cluster.local:6667", { HWLAB_BOOTSTRAP_ADMIN_API_KEY: "hwl_live_test" }); - await internalFetch("http://hwlab-cloud-api.hwlab-v03.svc.cluster.local:6667/v1/hwpod-node-ops", { method: "POST" }); await internalFetch("https://lab-dev.hwpod.com/v1/hwpod-node-ops", { method: "POST" }); - await internalFetch(new Request("http://hwlab-cloud-api.hwlab-v03.svc.cluster.local:6667/v1/hwpod-node-ops", { headers: { authorization: "Bearer caller-token" } }), { method: "POST" }); - assert.deepEqual(calls, [ { url: "http://hwlab-cloud-api.hwlab-v03.svc.cluster.local:6667/v1/hwpod-node-ops", authorization: "Bearer hwl_live_test" }, - { url: "https://lab-dev.hwpod.com/v1/hwpod-node-ops", authorization: null }, - { url: "http://hwlab-cloud-api.hwlab-v03.svc.cluster.local:6667/v1/hwpod-node-ops", authorization: "Bearer caller-token" } + { url: "https://lab-dev.hwpod.com/v1/hwpod-node-ops", authorization: null } ]); }); -test("cloud-api exposes web CaseRun cases, run status, events and aggregate", async () => { - const root = await mkdtemp(path.join(os.tmpdir(), "hwlab-caserun-api-")); - const stateRoot = path.join(root, "state"); - const caseDir = path.join(root, "cases", "d601-f103-v2-compile"); - await mkdir(caseDir, { recursive: true }); - await writeFile(path.join(caseDir, "case.json"), JSON.stringify({ - contractVersion: "hwpod-case-v1", - caseId: "d601-f103-v2-compile", - title: "D601 compile", - mode: "compile-only", - hwpodSpec: "hwpod-spec.yaml", - subject: { repoLocalPath: "F:\\Work\\HWLAB-CASE-F103", commitId: "abc", subdir: "projects/01_baseline" } - }), "utf8"); - await writeFile(path.join(caseDir, "hwpod-spec.yaml"), sampleSpecYaml(), "utf8"); - const seen: any[] = []; - const accessController = { - async ensureBootstrap() {}, - async requireNavAccess() { return true; } +test("cloud-api preserves same-origin CaseRun routes as a pure HarnessRL proxy", async () => { + const calls: Array<{ url: string; method: string; body: unknown }> = []; + const fetchImpl = async (input: URL | RequestInfo, init: RequestInit = {}) => { + const body = init.body ? JSON.parse(String(init.body)) : null; + calls.push({ url: String(input), method: String(init.method), body }); + return Response.json({ ok: true, runId: "proxy-run", status: "queued", authority: "harnessrl" }, { status: 202 }); }; - const server = createCloudApiServer({ - accessController, - env: { PATH: process.env.PATH, HWLAB_CASERUN_CASE_REPO: root, HWLAB_CASERUN_STATE_DIR: stateRoot, HWLAB_ENVIRONMENT: "v03", HWLAB_CLOUD_API_PORT: "6667" }, - caseRunExecutor: async (context: any, emit: any) => { - seen.push(context.parsed); - await emit("prepared", { runDir: context.parsed.runDir }); - await emit("build-completed", { jobId: "job-test" }); - return { - summary: { status: "recorded", jobId: "job-test", artifactCount: 2, artifacts: [{ path: "atk_f103.hex" }, { path: "atk_f103.axf" }] }, - evidence: { status: "recorded", keilJob: { jobId: "job-test", status: "completed" }, artifacts: [{ path: "atk_f103.hex" }, { path: "atk_f103.axf" }] }, - run: { runId: context.parsed.runId, caseId: context.parsed.caseId } - }; - } + const server = createServer((request, response) => { + const url = new URL(request.url ?? "/", "http://cloud-api.test"); + void handleCaseRunHttp(request, response, url, { env: { HARNESSRL_API_URL: "http://hwlab-harnessrl-api:6675" }, fetchImpl }); }); - await listen(server); + await new Promise((resolve) => server.listen(0, "127.0.0.1", resolve)); try { - const indexResponse = await fetch(`${serverUrl(server)}/v1`); - const indexPayload = await indexResponse.json(); - assert.equal(indexPayload.caserun.route, "/v1/caserun"); - - const casesResponse = await fetch(`${serverUrl(server)}/v1/caserun/cases`); - const casesPayload = await casesResponse.json(); - assert.equal(casesResponse.status, 200); - assert.equal(casesPayload.count, 1); - assert.equal(casesPayload.cases[0].caseId, "d601-f103-v2-compile"); - - const startResponse = await fetch(`${serverUrl(server)}/v1/caserun/runs`, { - method: "POST", - headers: { "content-type": "application/json" }, - body: JSON.stringify({ caseId: "d601-f103-v2-compile" }) - }); - const startPayload = await startResponse.json(); - assert.equal(startResponse.status, 202); - assert.ok(["queued", "running"].includes(startPayload.status)); - assert.match(startPayload.runId, /^web-d601-f103-v2-compile-/u); - assert.equal(startPayload.statusUrl, `/v1/caserun/runs/${startPayload.runId}`); - - const runPayload = await waitForRun(server, startPayload.runId, "completed"); - assert.equal(runPayload.terminal, true); - assert.equal(runPayload.facts.jobId, "job-test"); - assert.equal(runPayload.facts.artifactCount, 2); - assert.equal(runPayload.references.events.count >= 3, true); - assert.match(runPayload.references.aggregate.sha256, /^[a-f0-9]{64}$/u); - assert.equal(runPayload.sourceAuthority.authority, "caserun-archived-facts"); - assert.equal(seen[0].noCaseRepoRecord, true); - assert.equal(seen[0].caseRepo, root); - assert.equal(seen[0].apiUrl, "http://hwlab-cloud-api.hwlab-v03.svc.cluster.local:6667"); - - const eventsResponse = await fetch(`${serverUrl(server)}/v1/caserun/runs/${startPayload.runId}/events`); - const eventsPayload = await eventsResponse.json(); - assert.equal(eventsResponse.status, 200); - assert.equal(eventsPayload.runId, runPayload.runId); - assert.equal(eventsPayload.caseId, runPayload.caseId); - assert.equal(eventsPayload.sourceAuthority.authority, runPayload.sourceAuthority.authority); - assert.ok(eventsPayload.events.some((event: any) => event.stage === "build-completed")); - - const aggregateResponse = await fetch(`${serverUrl(server)}/v1/caserun/runs/${startPayload.runId}/aggregate`); - const aggregatePayload = await aggregateResponse.json(); - assert.equal(aggregateResponse.status, 200); - assert.equal(aggregatePayload.summary.jobId, "job-test"); - assert.match(aggregatePayload.sha256, /^[a-f0-9]{64}$/u); - assert.equal(aggregatePayload.sourceAuthority.authority, runPayload.sourceAuthority.authority); + const address = server.address(); + if (!address || typeof address === "string") throw new Error("server address unavailable"); + const response = await fetch(`http://127.0.0.1:${address.port}/v1/caserun/runs`, { method: "POST", headers: { "content-type": "application/json" }, body: JSON.stringify({ caseId: "software-smoke" }) }); + assert.equal(response.status, 202); + assert.deepEqual(await response.json(), { ok: true, runId: "proxy-run", status: "queued", authority: "harnessrl" }); + assert.deepEqual(calls, [{ url: "http://hwlab-harnessrl-api:6675/v1/caserun/runs", method: "POST", body: { caseId: "software-smoke" } }]); } finally { - await close(server); - await rm(root, { recursive: true, force: true }); + await new Promise((resolve, reject) => server.close((error) => error ? reject(error) : resolve())); } }); -test("builtin CaseRun exposes and preflights the D601 download and UART case", async () => { - const stateRoot = await mkdtemp(path.join(os.tmpdir(), "hwlab-caserun-download-uart-")); - const caseId = "d601-f103-v2-main-marker-download-uart"; - const caseFile = path.join(process.cwd(), "cases", caseId, "case.json"); - const definition = JSON.parse(await readFile(caseFile, "utf8")); - const validationPlan = caseValidationPlanFromDefinition(definition); - assert.equal(validationPlan.mode, "download-uart"); - assert.equal(validationPlan.compileOnly, false); - assert.deepEqual(validationPlan.steps.map((step) => step.kind), ["build", "download", "uart"]); - assert.deepEqual(validationPlan.steps.map((step) => step.expectedOp), ["debug.build", "debug.download", "io.uart.read"]); - assert.deepEqual(validationPlan.steps.map((step) => hwpodCommandForValidationStep(step, { - executable: "bun", - cliPath: "tools/hwpod-cli.ts", - specPath: ".hwlab/hwpod-spec.yaml", - reason: "focused preflight" - }).slice(2, 4)), [["build", "--spec"], ["download", "--spec"], ["uart", "read"]]); - - const accessController = { - async ensureBootstrap() {}, - async requireNavAccess() { return true; } - }; - const server = createCloudApiServer({ - accessController, - env: { PATH: process.env.PATH, HWLAB_CASERUN_REPO_ROOT: process.cwd(), HWLAB_CASERUN_STATE_DIR: stateRoot, HWLAB_ENVIRONMENT: "v03", HWLAB_CLOUD_API_PORT: "6667" }, - caseRunExecutor: async (context: any, emit: any) => { - await emit("prepared", { runDir: context.parsed.runDir }); - throw Object.assign(new Error("No HWPOD node is available for focused preflight."), { - code: "hwpod_node_unavailable", - details: { blocker: { code: "hwpod_node_unavailable", summary: "No HWPOD node is available.", retryable: true } } - }); - } +test("cloud-api refuses to recreate an in-process CaseRun fallback", async () => { + const server = createServer((request, response) => { + const url = new URL(request.url ?? "/", "http://cloud-api.test"); + void handleCaseRunHttp(request, response, url, { env: {} }); }); - await listen(server); + await new Promise((resolve) => server.listen(0, "127.0.0.1", resolve)); try { - const casesResponse = await fetch(`${serverUrl(server)}/v1/caserun/cases`); - const casesPayload = await casesResponse.json(); - const matches = casesPayload.cases.filter((item: any) => item.caseId === caseId); - assert.equal(matches.length, 1); - assert.deepEqual(matches[0], { - caseId, - title: "D601-F103-V2 main marker download and UART CaseRun", - mode: "download-uart", - hwpodSpec: "hwpod-spec.yaml", - hardwareRequired: true, - available: true, - subject: definition.subject, - expected: definition.expected, - runtime: definition.runtime - }); - - const startResponse = await fetch(`${serverUrl(server)}/v1/caserun/runs`, { - method: "POST", - headers: { "content-type": "application/json" }, - body: JSON.stringify({ caseId }) - }); - const started = await startResponse.json(); - assert.equal(startResponse.status, 202); - assert.match(started.runId, /^web-d601-f103-v2-main-marker-download-uart-/u); - const failed = await waitForRun(server, started.runId, "failed"); - assert.equal(failed.caseId, caseId); - assert.equal(failed.terminal, true); - assert.equal(failed.blocker.code, "hwpod_node_unavailable"); + const address = server.address(); + if (!address || typeof address === "string") throw new Error("server address unavailable"); + const response = await fetch(`http://127.0.0.1:${address.port}/v1/caserun/cases`); + const payload = await response.json(); + assert.equal(response.status, 503); + assert.equal(payload.error.code, "harnessrl_api_url_required"); } finally { - await close(server); - await rm(stateRoot, { recursive: true, force: true }); + await new Promise((resolve, reject) => server.close((error) => error ? reject(error) : resolve())); } }); - -test("cloud-api completes a hardware-free software smoke CaseRun", async () => { - const root = await mkdtemp(path.join(os.tmpdir(), "hwlab-caserun-software-smoke-")); - const stateRoot = path.join(root, "state"); - const caseDir = path.join(root, "cases", "software-smoke"); - await mkdir(caseDir, { recursive: true }); - await writeFile(path.join(caseDir, "case.json"), JSON.stringify({ - contractVersion: "hwlab-caserun-software-smoke-v1", - caseId: "software-smoke", - title: "software smoke", - mode: "software-smoke" - }), "utf8"); - const accessController = { - async ensureBootstrap() {}, - async requireNavAccess() { return true; } - }; - const server = createCloudApiServer({ - accessController, - env: { PATH: process.env.PATH, HWLAB_CASERUN_CASE_REPO: root, HWLAB_CASERUN_STATE_DIR: stateRoot, HWLAB_ENVIRONMENT: "v03", HWLAB_CLOUD_API_PORT: "6667" } - }); - await listen(server); - try { - const casesResponse = await fetch(`${serverUrl(server)}/v1/caserun/cases`); - const casesPayload = await casesResponse.json(); - assert.equal(casesPayload.cases[0].caseId, "software-smoke"); - assert.equal(casesPayload.cases[0].hardwareRequired, false); - assert.equal(casesPayload.cases[0].available, true); - - const startResponse = await fetch(`${serverUrl(server)}/v1/caserun/runs`, { - method: "POST", - headers: { "content-type": "application/json" }, - body: JSON.stringify({ caseId: "software-smoke" }) - }); - const started = await startResponse.json(); - assert.equal(startResponse.status, 202); - const completed = await waitForRun(server, started.runId, "completed"); - assert.equal(completed.terminal, true); - assert.equal(completed.facts.artifactCount, 1); - - const eventsResponse = await fetch(`${serverUrl(server)}/v1/caserun/runs/${started.runId}/events`); - const eventsPayload = await eventsResponse.json(); - assert.ok(eventsPayload.events.some((event: any) => event.stage === "software-smoke-completed")); - - const aggregateResponse = await fetch(`${serverUrl(server)}/v1/caserun/runs/${started.runId}/aggregate`); - const aggregate = await aggregateResponse.json(); - assert.equal(aggregate.status, "completed"); - assert.equal(aggregate.evidence.hardwareRequired, false); - assert.match(aggregate.sha256, /^[a-f0-9]{64}$/u); - const artifact = JSON.parse(await readFile(path.join(stateRoot, "runs", started.runId, "software-smoke.json"), "utf8")); - assert.deepEqual(artifact.checks, { caseDefinitionLoaded: true, runDirectoryWritable: true, backgroundWorkerRunning: true }); - } finally { - await close(server); - await rm(root, { recursive: true, force: true }); - } -}); - -test("cloud-api projects queued, running, completed and failed records with one read-model shape", async () => { - const root = await mkdtemp(path.join(os.tmpdir(), "hwlab-caserun-read-model-")); - const stateRoot = path.join(root, "state"); - const accessController = { - async ensureBootstrap() {}, - async requireNavAccess() { return true; } - }; - const server = createCloudApiServer({ - accessController, - env: { PATH: process.env.PATH, HWLAB_CASERUN_CASE_REPO: root, HWLAB_CASERUN_STATE_DIR: stateRoot } - }); - await mkdir(path.join(stateRoot, "records"), { recursive: true }); - const statuses = ["queued", "running", "completed", "failed"]; - for (const status of statuses) { - const runId = `read-model-${status}`; - const record = { - runId, - caseId: "read-model-case", - status, - stage: status === "running" ? "build" : status, - runDir: path.join(stateRoot, "runs", runId), - createdAt: "2026-07-16T00:00:00.000Z", - updatedAt: "2026-07-16T00:00:01.000Z", - caseAuthority: { kind: "case-repo" }, - events: [], - result: status === "completed" ? { - summary: { jobId: "job-read-model", artifactCount: 1, artifactManifestPath: "runs/read-model/artifact-manifest.json", artifactManifestSha256: "a".repeat(64) }, - evidence: { status: "recorded", replay: { contractVersion: "hwpod-case-validation-replay-v1" }, validation: { archivePath: "validation-observations.json" }, hwpod: { operation: { rawArtifactRef: { path: "raw.json" } } } }, - run: { runId, caseId: "read-model-case" } - } : null, - blocker: status === "failed" ? { code: "hwpod_build_failed", summary: "HWPOD build failed", layer: "hwpod" } : null, - error: status === "failed" ? { code: "caserun_failed", message: "CaseRun failed." } : null - }; - await mkdir(record.runDir, { recursive: true }); - await writeFile(path.join(stateRoot, "records", `${runId}.json`), `${JSON.stringify(record)}\n`, "utf8"); - } - await listen(server); - try { - for (const status of statuses) { - const response = await fetch(`${serverUrl(server)}/v1/caserun/runs/read-model-${status}`); - const payload = await response.json(); - assert.equal(response.status, 200); - assert.deepEqual(Object.keys(payload).sort(), ["blocker", "caseId", "contractVersion", "createdAt", "error", "facts", "ok", "references", "runId", "sourceAuthority", "stage", "status", "terminal", "updatedAt"].sort()); - assert.equal(payload.status, status); - assert.equal(payload.terminal, ["completed", "failed"].includes(status)); - assert.equal(payload.references.status.href, `/v1/caserun/runs/read-model-${status}`); - } - const completed = await (await fetch(`${serverUrl(server)}/v1/caserun/runs/read-model-completed`)).json(); - assert.equal(completed.references.manifest.sha256, "a".repeat(64)); - assert.equal(completed.references.replay.ref, "validation-observations.json"); - assert.equal(completed.references.hwpod.path, "raw.json"); - const failed = await (await fetch(`${serverUrl(server)}/v1/caserun/runs/read-model-failed`)).json(); - assert.equal(failed.blocker.code, "hwpod_build_failed"); - } finally { - await close(server); - await rm(root, { recursive: true, force: true }); - } -}); - -async function waitForRun(server: any, runId: string, status: string) { - const deadline = Date.now() + 3000; - let last: any = null; - while (Date.now() < deadline) { - const response = await fetch(`${serverUrl(server)}/v1/caserun/runs/${runId}`); - last = await response.json(); - if (last.status === status) return last; - await new Promise((resolve) => setTimeout(resolve, 50)); - } - assert.fail(`run did not reach ${status}: ${JSON.stringify(last)}`); -} - -async function listen(server: any) { - await new Promise((resolve) => server.listen(0, "127.0.0.1", resolve)); -} - -async function close(server: any) { - await new Promise((resolve, reject) => server.close((error: Error | undefined) => (error ? reject(error) : resolve(undefined)))); -} - -function serverUrl(server: any) { - return `http://127.0.0.1:${server.address().port}`; -} - -function sampleSpecYaml() { - return `apiVersion: hwlab.dev/v0alpha1 -kind: Hwpod -metadata: - uid: D601-F103-V2 - name: d601-f103-v2 -spec: - workspace: - path: "F:\\\\Work\\\\HWLAB-CASE-F103" - nodeBinding: - nodeId: node-d601-f103-v2 -`; -} diff --git a/internal/cloud/server-caserun-http.ts b/internal/cloud/server-caserun-http.ts index 4a9d1f3d..160fde8b 100644 --- a/internal/cloud/server-caserun-http.ts +++ b/internal/cloud/server-caserun-http.ts @@ -1,219 +1,28 @@ -import { createHash, randomUUID } from "node:crypto"; import type { IncomingMessage, ServerResponse } from "node:http"; -import { mkdir, readdir, readFile, stat, writeFile } from "node:fs/promises"; -import path from "node:path"; -import { buildCaseRun, collectCaseRun, prepareCaseRun } from "../../tools/src/hwlab-caserun-lib.ts"; - -const CONTRACT_VERSION = "hwlab-caserun-web-v1"; -const CASES_CONTRACT_VERSION = "hwlab-caserun-cases-v1"; -const AGGREGATE_CONTRACT_VERSION = "hwlab-caserun-aggregate-v1"; -const SOFTWARE_SMOKE_CONTRACT_VERSION = "hwlab-caserun-software-smoke-v1"; -const SOFTWARE_SMOKE_MODE = "software-smoke"; const DEFAULT_BODY_LIMIT_BYTES = 64 * 1024; -const RUN_ID_PATTERN = /^[A-Za-z0-9_.:-]{3,180}$/u; - -const memoryRuns = new Map(); export async function handleCaseRunHttp(request: IncomingMessage, response: ServerResponse, url: URL, options: any = {}) { const env = options.env ?? process.env; - const cwd = resolveCwd(env); - const stateRoot = resolveStateRoot(cwd, env); - const caseRepo = resolveCaseRepo(cwd, env); - - if (request.method === "GET" && url.pathname === "/v1/caserun") { - sendJson(response, 200, { - ok: true, - contractVersion: CONTRACT_VERSION, - casesRoute: "/v1/caserun/cases", - runsRoute: "/v1/caserun/runs", - caseAuthority: describeCaseAuthority(caseRepo, cwd), - compileOnly: false - }); - return; - } - - if (request.method === "GET" && url.pathname === "/v1/caserun/cases") { - const cases = await listCases(caseRepo); - sendJson(response, 200, { - ok: true, - contractVersion: CASES_CONTRACT_VERSION, - caseAuthority: describeCaseAuthority(caseRepo, cwd), - count: cases.length, - cases - }); - return; - } - - if (url.pathname === "/v1/caserun/runs") { - if (request.method !== "POST") return sendError(response, 405, "method_not_allowed", "POST required."); - const parsed = await readJsonBody(request, DEFAULT_BODY_LIMIT_BYTES); - if (!parsed.ok) return sendJson(response, 400, parsed.error); - const body = parsed.body && typeof parsed.body === "object" && !Array.isArray(parsed.body) ? parsed.body as Record : {}; - const caseId = safeOpaqueId(body.caseId); - if (!caseId) return sendError(response, 400, "invalid_case_id", "caseId is required."); - const caseDefinition = await loadCase(caseRepo, caseId).catch(() => null); - if (!caseDefinition) return sendError(response, 404, "case_not_found", "CaseRun case was not found.", { caseId }); - const requestedRunId = safeOpaqueId(body.runId); - const runId = requestedRunId || nextRunId(caseId); - const runDir = path.join(stateRoot, "runs", runId); - const record = { - ok: true, - contractVersion: CONTRACT_VERSION, - runId, - caseId, - status: "queued", - stage: "queued", - runDir, - createdAt: new Date().toISOString(), - updatedAt: new Date().toISOString(), - caseAuthority: describeCaseAuthority(caseRepo, cwd), - runtimeApiUrl: runtimeApiUrlFromRequest(request, env), - events: [] as any[], - result: null as any - }; - await saveRecord(stateRoot, record); - remember(record); - appendEvent(record, "queued", "queued", { caseId }); - startBackgroundCaseRun(record, { ...options, env, cwd, stateRoot, caseRepo, caseDefinition }); - sendJson(response, 202, acceptedRecord(record)); - return; - } - - const runMatch = url.pathname.match(/^\/v1\/caserun\/runs\/([^/]+)(?:\/(events|aggregate|cancel))?$/u); - if (runMatch) { - const runId = decodeURIComponent(runMatch[1] ?? ""); - if (!RUN_ID_PATTERN.test(runId)) return sendError(response, 400, "invalid_run_id", "runId is invalid."); - const suffix = runMatch[2] ?? ""; - const record = await findRecord(stateRoot, runId); - if (!record) return sendError(response, 404, "caserun_run_not_found", "CaseRun run was not found.", { runId }); - if (!suffix && request.method === "GET") return sendJson(response, 200, await projectRunReadModel(record)); - if (suffix === "events" && request.method === "GET") return sendJson(response, 200, { - ok: true, - contractVersion: CONTRACT_VERSION, - runId, - caseId: record.caseId, - status: record.status, - stage: record.stage, - terminal: terminalStatus(record.status), - sourceAuthority: sourceAuthorityForRecord(record), - events: record.events ?? [] - }); - if (suffix === "aggregate" && request.method === "GET") return sendJson(response, 200, await aggregateForRecord(record)); - if (suffix === "cancel" && request.method === "POST") return sendError(response, 409, "caserun_cancel_not_supported", "This CaseRun worker cannot be cancelled after dispatch.", { runId, status: record.status }); - return sendError(response, 405, "method_not_allowed", suffix === "cancel" ? "POST required." : "GET required."); - } - - sendError(response, 404, "caserun_route_not_found", "CaseRun REST route is not implemented.", { route: url.pathname }); -} - -function startBackgroundCaseRun(record: any, input: any) { - void runBackgroundCaseRun(record, input).catch(async (error) => { - appendEvent(record, "failed", "failed", { code: error?.code ?? error?.name ?? "caserun_failed", message: String(error?.message ?? error).slice(0, 500), details: error?.details ?? null }); - record.status = "failed"; - record.stage = "failed"; - record.error = { code: error?.code ?? error?.name ?? "caserun_failed", message: String(error?.message ?? error), details: error?.details ?? null }; - record.blocker = typedBlocker(error?.blocker ?? error?.details?.blocker ?? error?.details ?? null); - record.updatedAt = new Date().toISOString(); - await saveRecord(input.stateRoot, record).catch(() => undefined); - }); -} - -async function runBackgroundCaseRun(record: any, input: any) { - appendEvent(record, "running", "prepare", { caseId: record.caseId }); - record.status = "running"; - record.stage = "prepare"; - record.updatedAt = new Date().toISOString(); - await saveRecord(input.stateRoot, record); - const context = caseContextForRecord(record, input); - const emit = async (stage: string, payload: Record = {}) => { - appendEvent(record, "running", stage, payload); - record.stage = stage; - record.updatedAt = new Date().toISOString(); - await saveRecord(input.stateRoot, record); - }; - const executor = typeof input.caseRunExecutor === "function" ? input.caseRunExecutor : defaultCaseRunExecutor; - const result = await executor(context, emit, { record, caseDefinition: input.caseDefinition }); - record.status = "completed"; - record.stage = "completed"; - record.result = compactResult(result); - record.updatedAt = new Date().toISOString(); - appendEvent(record, "completed", "completed", { jobId: record.result?.summary?.jobId ?? record.result?.evidence?.keilJob?.jobId ?? null }); - await saveRecord(input.stateRoot, record); -} - -async function defaultCaseRunExecutor(context: any, emit: (stage: string, payload?: Record) => Promise, input: any = {}) { - if (input.caseDefinition?.mode === SOFTWARE_SMOKE_MODE) return runSoftwareSmokeCase(context, emit, input.caseDefinition); - const prepared = await prepareCaseRun(context, "web"); - await emit("prepared", { runDir: prepared.runDir, specPath: prepared.specPath }); - const run = prepared.run; - const build = await buildCaseRun(context, run); - await emit("build-completed", { jobId: build.summary?.jobId ?? build.evidence?.keilJob?.jobId ?? null, hwpodExitCode: build.summary?.hwpodExitCode ?? null }); - const collected = await collectCaseRun(context, run, build.evidence); - await emit("collect-completed", { status: collected.status, artifactCount: Array.isArray(build.evidence?.artifacts) ? build.evidence.artifacts.length : 0 }); - return { prepared, build, collected, evidence: build.evidence, summary: build.summary, run: collected.run ?? build.run ?? run }; -} - -async function runSoftwareSmokeCase(context: any, emit: (stage: string, payload?: Record) => Promise, caseDefinition: any) { - const runDir = path.resolve(context.parsed.runDir); - await mkdir(runDir, { recursive: true }); - await emit("software-smoke-running", { mode: SOFTWARE_SMOKE_MODE, hardwareRequired: false }); - const artifact = { - contractVersion: SOFTWARE_SMOKE_CONTRACT_VERSION, - caseId: context.parsed.caseId, - runId: context.parsed.runId, - mode: SOFTWARE_SMOKE_MODE, - status: "completed", - hardwareRequired: false, - checks: { - caseDefinitionLoaded: caseDefinition.available === true, - runDirectoryWritable: true, - backgroundWorkerRunning: true + const baseUrl = String(env.HARNESSRL_API_URL ?? "").trim().replace(/\/+$/u, ""); + if (!baseUrl) return sendJson(response, 503, { ok: false, error: { code: "harnessrl_api_url_required", message: "HARNESSRL_API_URL is required for CaseRun routes" } }); + try { + const target = new URL(`${url.pathname}${url.search}`, `${baseUrl}/`); + const headers = new Headers(); + for (const [name, value] of Object.entries(request.headers)) { + if (value === undefined || name.toLowerCase() === "host" || name.toLowerCase() === "content-length") continue; + headers.set(name, Array.isArray(value) ? value.join(",") : value); } - }; - const artifactPath = path.join(runDir, "software-smoke.json"); - const artifactText = `${JSON.stringify(artifact, null, 2)}\n`; - await writeFile(artifactPath, artifactText, "utf8"); - const artifacts = [{ path: "software-smoke.json", bytes: Buffer.byteLength(artifactText), sha256: createHash("sha256").update(artifactText).digest("hex") }]; - const evidence = { status: "recorded", mode: SOFTWARE_SMOKE_MODE, hardwareRequired: false, checks: artifact.checks, artifacts }; - const summary = { status: "completed", mode: SOFTWARE_SMOKE_MODE, hardwareRequired: false, artifactCount: artifacts.length, artifacts }; - const run = { runId: context.parsed.runId, caseId: context.parsed.caseId, mode: SOFTWARE_SMOKE_MODE, hardwareRequired: false }; - await emit("software-smoke-completed", { artifactCount: artifacts.length, artifactSha256: artifacts[0].sha256 }); - return { summary, evidence, run }; -} - -function caseContextForRecord(record: any, input: any) { - const internalApiUrl = internalRuntimeApiUrlForCaseRun(record, input.env ?? {}); - const fetchImpl = caseRunInternalFetch(input.fetchImpl ?? globalThis.fetch, internalApiUrl, input.env ?? {}); - const parsed = { - _: [], - caseId: record.caseId, - runId: record.runId, - runDir: record.runDir, - caseRepo: input.caseRepo, - apiUrl: internalApiUrl, - noCaseRepoRecord: true, - stateDir: path.join(input.stateRoot, "cli") - }; - return { - parsed, - env: { - ...process.env, - ...input.env, - HWLAB_CASE_REPO: input.caseRepo, - HWLAB_RUNTIME_API_URL: internalApiUrl, - HWLAB_CASERUN_PUBLIC_RUNTIME_API_URL: record.runtimeApiUrl, - HWLAB_RUNTIME_ENDPOINT_LOCKED: "1" - }, - fetchImpl, - cwd: input.cwd, - now: () => new Date().toISOString(), - sleep: (ms: number) => new Promise((resolve) => setTimeout(resolve, ms)), - runProcess: input.caseRunProcess, - startProcess: input.caseRunStartProcess, - argv: [], - rest: ["web", record.caseId] - }; + headers.set("x-forwarded-host", String(request.headers["x-forwarded-host"] ?? request.headers.host ?? "")); + headers.set("x-forwarded-proto", String(request.headers["x-forwarded-proto"] ?? "http")); + const body = request.method === "GET" || request.method === "HEAD" ? undefined : await readBody(request, DEFAULT_BODY_LIMIT_BYTES); + const upstream = await (options.fetchImpl ?? globalThis.fetch)(target, { method: request.method, headers, body }); + response.statusCode = upstream.status; + for (const [name, value] of upstream.headers) if (name.toLowerCase() !== "content-length" && name.toLowerCase() !== "transfer-encoding") response.setHeader(name, value); + response.end(Buffer.from(await upstream.arrayBuffer())); + } catch (error: any) { + sendJson(response, 502, { ok: false, error: { code: error?.code ?? "harnessrl_api_unavailable", message: error?.message ?? String(error) } }); + } } export function caseRunInternalFetch(fetchImpl: typeof fetch, internalApiUrl: string, env: Record) { @@ -230,305 +39,21 @@ export function caseRunInternalFetch(fetchImpl: typeof fetch, internalApiUrl: st }; } -function resolveCwd(env: Record) { - return path.resolve(String(env.HWLAB_CASERUN_REPO_ROOT ?? process.cwd())); -} - -function resolveStateRoot(cwd: string, env: Record) { - return path.resolve(cwd, String(env.HWLAB_CASERUN_STATE_DIR ?? path.join(".state", "hwlab-cloud-api", "caserun"))); -} - -function resolveCaseRepo(cwd: string, env: Record) { - return path.resolve(cwd, String(env.HWLAB_CASERUN_CASE_REPO ?? env.HWLAB_CASE_REPO ?? ".")); -} - -function describeCaseAuthority(caseRepo: string, cwd: string) { - const rel = relativeOrAbsolute(cwd, caseRepo); - return { - kind: rel === "." ? "builtin-hwlab-v03" : "configured-case-repo", - repoPath: rel, - casesPath: path.join(rel, "cases") - }; -} - -async function listCases(caseRepo: string) { - const casesRoot = path.join(caseRepo, "cases"); - const entries = await readdir(casesRoot, { withFileTypes: true }).catch(() => []); - const cases = [] as any[]; - for (const entry of entries) { - if (!entry.isDirectory()) continue; - const loaded = await loadCase(caseRepo, entry.name).catch(() => null); - if (loaded) cases.push(loaded); - } - cases.sort((left, right) => String(left.caseId).localeCompare(String(right.caseId))); - return cases; -} - -async function loadCase(caseRepo: string, caseId: string) { - if (!safeOpaqueId(caseId)) throw new Error("invalid case id"); - const caseDir = path.join(caseRepo, "cases", caseId); - const caseFile = path.join(caseDir, "case.json"); - const definition = JSON.parse(await readFile(caseFile, "utf8")); - const mode = String(definition.mode ?? "compile-only").trim() || "compile-only"; - const hardwareRequired = mode !== SOFTWARE_SMOKE_MODE; - const hwpodSpec = hardwareRequired ? String(definition.hwpodSpec ?? definition.specPath ?? "hwpod-spec.yaml").trim() || "hwpod-spec.yaml" : null; - const specPath = hwpodSpec ? path.resolve(caseDir, hwpodSpec) : null; - const specInfo = specPath ? await stat(specPath).catch(() => null) : null; - return { - caseId: String(definition.caseId ?? caseId).trim() || caseId, - title: String(definition.title ?? caseId).trim() || caseId, - mode, - hwpodSpec, - hardwareRequired, - available: hardwareRequired ? Boolean(specInfo?.isFile()) : true, - subject: definition.subject ?? null, - expected: definition.expected ?? null, - runtime: definition.runtime ?? null - }; -} - -async function findRecord(stateRoot: string, runId: string) { - const memory = memoryRuns.get(runId); - if (memory) return memory; - const text = await readFile(recordPath(stateRoot, runId), "utf8").catch(() => ""); - if (!text) return null; - const record = JSON.parse(text); - remember(record); - return record; -} - -async function saveRecord(stateRoot: string, record: any) { - await mkdir(path.dirname(recordPath(stateRoot, record.runId)), { recursive: true }); - await mkdir(record.runDir, { recursive: true }); - await writeFile(recordPath(stateRoot, record.runId), `${JSON.stringify(record, null, 2)}\n`, "utf8"); -} - -function recordPath(stateRoot: string, runId: string) { - return path.join(stateRoot, "records", `${runId}.json`); -} - -function remember(record: any) { - memoryRuns.set(record.runId, record); -} - -function appendEvent(record: any, status: string, stage: string, payload: Record = {}) { - const event = { at: new Date().toISOString(), status, stage, payload }; - record.events = Array.isArray(record.events) ? [...record.events, event] : [event]; - record.updatedAt = event.at; - return event; -} - -async function aggregateForRecord(record: any) { - const evidence = record.result?.evidence ?? await readJson(path.join(record.runDir, "evidence.json")).catch(() => null); - const run = record.result?.run ?? await readJson(path.join(record.runDir, "run.json")).catch(() => null); - const summary = record.result?.summary ?? summarizeEvidence(evidence); - const source = { runId: record.runId, caseId: record.caseId, status: record.status, summary, evidence, run }; - return { - ok: true, - contractVersion: AGGREGATE_CONTRACT_VERSION, - runId: record.runId, - caseId: record.caseId, - status: record.status, - stage: record.stage, - terminal: terminalStatus(record.status), - sourceAuthority: sourceAuthorityForRecord(record), - sha256: sha256Json(source), - summary, - evidence, - run - }; -} - -function compactResult(result: any) { - return { - summary: result?.summary ?? null, - evidence: result?.evidence ?? null, - run: result?.run ?? result?.collected?.run ?? result?.build?.run ?? result?.prepared?.run ?? null, - collected: result?.collected ? { status: result.collected.status, summary: result.collected.summary ?? null } : null - }; -} - -function summarizeEvidence(evidence: any) { - if (!evidence || typeof evidence !== "object") return null; - return { - status: evidence.status ?? null, - jobId: evidence.keilJob?.jobId ?? null, - keilStatus: evidence.keilJob?.status ?? null, - hwpodExitCode: evidence.hwpod?.exitCode ?? null, - artifactCount: Array.isArray(evidence.artifacts) ? evidence.artifacts.length : 0, - artifacts: evidence.artifacts ?? [] - }; -} - -async function projectRunReadModel(record: any) { - const aggregate = await aggregateForRecord(record); - const summary = aggregate.summary && typeof aggregate.summary === "object" ? aggregate.summary : {}; - const evidence = aggregate.evidence && typeof aggregate.evidence === "object" ? aggregate.evidence : {}; - const archivedRun = aggregate.run && typeof aggregate.run === "object" ? aggregate.run : {}; - const blocker = typedBlocker(record.blocker ?? archivedRun.blocker ?? evidence.validation?.blocker ?? evidence.hwpod?.buildWaitBlocker ?? evidence.hwpod?.operation?.blocker ?? record.error?.details?.blocker ?? null); - const manifestPath = firstText(summary.artifactManifestPath, evidence.artifactManifest?.path, archivedRun.artifactManifestPath); - const manifestSha256 = firstSha256(summary.artifactManifestSha256, evidence.artifactManifest?.sha256, archivedRun.artifactManifestSha256); - const replay = evidence.replay && typeof evidence.replay === "object" ? evidence.replay : null; - const replayPath = firstText(evidence.validation?.archivePath, archivedRun.replayPath); - const trace = firstObject(summary.trace, summary.traceLookup, evidence.agentTrace, evidence.traceLookup, archivedRun.agentTrace); - const hwpod = firstObject(evidence.hwpod?.operation?.rawArtifactRef, evidence.hwpod?.operation, evidence.keilJob, archivedRun.hwpod); - return { - ok: true, - contractVersion: CONTRACT_VERSION, - runId: record.runId, - caseId: record.caseId, - status: record.status, - stage: record.stage, - terminal: terminalStatus(record.status), - blocker, - createdAt: record.createdAt, - updatedAt: record.updatedAt, - facts: { - jobId: firstText(summary.jobId, evidence.keilJob?.jobId, evidence.hwpod?.operation?.jobId) || null, - artifactCount: Number(summary.artifactCount ?? (Array.isArray(evidence.artifacts) ? evidence.artifacts.length : 0)), - evidenceStatus: firstText(evidence.status, summary.status) || null - }, - references: { - status: { href: `/v1/caserun/runs/${encodeURIComponent(record.runId)}` }, - events: { href: `/v1/caserun/runs/${encodeURIComponent(record.runId)}/events`, count: Array.isArray(record.events) ? record.events.length : 0 }, - aggregate: { href: `/v1/caserun/runs/${encodeURIComponent(record.runId)}/aggregate`, sha256: aggregate.sha256 }, - manifest: manifestPath ? { ref: manifestPath, sha256: manifestSha256 } : null, - replay: replay || replayPath ? { ref: replayPath || null, relationship: replay } : null, - trace, - hwpod - }, - sourceAuthority: sourceAuthorityForRecord(record), - error: record.error ? { code: record.error.code ?? "caserun_failed", message: record.error.message ?? "CaseRun failed." } : null - }; -} - -function acceptedRecord(record: any) { - return { - ok: true, - contractVersion: CONTRACT_VERSION, - runId: record.runId, - caseId: record.caseId, - status: record.status, - statusUrl: `/v1/caserun/runs/${encodeURIComponent(record.runId)}` - }; -} - -function sourceAuthorityForRecord(record: any) { - return { - projector: "hwlab-cloud-api", - authority: "caserun-archived-facts", - runRecord: "cloud-api-caserun-record", - archive: "caserun-run-directory", - execution: "hwpod-agentrun-caserun", - case: record.caseAuthority ?? null - }; -} - -function terminalStatus(status: unknown) { - return ["completed", "failed", "cancelled", "canceled", "timed_out", "timeout"].includes(String(status ?? "").toLowerCase()); -} - -function typedBlocker(value: any) { - if (!value || typeof value !== "object" || Array.isArray(value)) return null; - const code = firstText(value.code); - if (!code) return null; - return { - code, - summary: firstText(value.summary, value.message) || code, - layer: firstText(value.layer) || null, - details: value.details && typeof value.details === "object" ? value.details : null - }; -} - -function firstObject(...values: any[]) { - return values.find((value) => value && typeof value === "object" && !Array.isArray(value)) ?? null; -} - -function firstText(...values: unknown[]) { - return values.map((value) => String(value ?? "").trim()).find(Boolean) ?? ""; -} - -function firstSha256(...values: unknown[]) { - return values.map((value) => String(value ?? "").trim().toLowerCase()).find((value) => /^[a-f0-9]{64}$/u.test(value)) ?? null; -} - -function runtimeApiUrlFromRequest(request: IncomingMessage, env: Record) { - const explicit = String(env.HWLAB_CASERUN_RUNTIME_API_URL ?? "").trim(); - if (explicit) return explicit.replace(/\/+$/u, ""); - const host = String(request.headers["x-forwarded-host"] ?? request.headers.host ?? "127.0.0.1:8080").split(",")[0]?.trim() || "127.0.0.1:8080"; - const proto = String(request.headers["x-forwarded-proto"] ?? "http").split(",")[0]?.trim() || "http"; - return `${proto}://${host}`.replace(/\/+$/u, ""); -} - -function internalRuntimeApiUrlForCaseRun(record: any, env: Record) { - const explicit = String(env.HWLAB_CASERUN_INTERNAL_API_URL ?? env.HWLAB_RUNTIME_INTERNAL_API_URL ?? "").trim(); - if (explicit) return explicit.replace(/\/+$/u, ""); - const namespace = String(env.HWLAB_NAMESPACE ?? env.POD_NAMESPACE ?? env.KUBERNETES_NAMESPACE ?? namespaceFromEnvironment(env.HWLAB_ENVIRONMENT) ?? "").trim(); - const port = String(env.HWLAB_CLOUD_API_PORT ?? env.HWLAB_PORT ?? env.PORT ?? "6667").trim() || "6667"; - if (namespace) return `http://hwlab-cloud-api.${namespace}.svc.cluster.local:${port}`; - return String(record.runtimeApiUrl ?? "").replace(/\/+$/u, ""); -} - -function namespaceFromEnvironment(value: string | undefined) { - const text = String(value ?? "").trim(); - if (!text) return ""; - if (/^hwlab-[a-z0-9-]+$/u.test(text)) return text; - if (/^[a-z0-9-]+$/u.test(text)) return `hwlab-${text}`; - return ""; -} - -async function readJsonBody(request: IncomingMessage, limitBytes: number) { +async function readBody(request: IncomingMessage, limitBytes: number) { let size = 0; - const chunks = [] as Buffer[]; + const chunks: Buffer[] = []; for await (const chunk of request) { const buffer = Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk); size += buffer.length; - if (size > limitBytes) return { ok: false, error: { ok: false, error: { code: "body_too_large", message: `body exceeds ${limitBytes} bytes` } } }; + if (size > limitBytes) throw Object.assign(new Error(`body exceeds ${limitBytes} bytes`), { code: "body_too_large" }); chunks.push(buffer); } - const body = Buffer.concat(chunks).toString("utf8").trim(); - if (!body) return { ok: true, body: {} }; - try { - return { ok: true, body: JSON.parse(body) }; - } catch (error) { - return { ok: false, error: { ok: false, error: { code: "invalid_json", message: error instanceof Error ? error.message : String(error) } } }; - } + return Buffer.concat(chunks); } -async function readJson(file: string) { - return JSON.parse(await readFile(file, "utf8")); -} - -function sendError(response: ServerResponse, statusCode: number, code: string, message: string, details: Record = {}) { - return sendJson(response, statusCode, { ok: false, error: { code, message, details } }); -} - -function sendJson(response: ServerResponse, statusCode: number, body: unknown) { - if (response.headersSent || response.writableEnded) return false; - response.writeHead(statusCode, { "content-type": "application/json; charset=utf-8", "cache-control": "no-store" }); +function sendJson(response: ServerResponse, status: number, body: unknown) { + response.statusCode = status; + response.setHeader("content-type", "application/json; charset=utf-8"); + response.setHeader("cache-control", "no-store"); response.end(`${JSON.stringify(body)}\n`); - return true; -} - -function safeOpaqueId(value: unknown) { - const text = String(value ?? "").trim(); - return RUN_ID_PATTERN.test(text) ? text : ""; -} - -function nextRunId(caseId: string) { - const stamp = new Date().toISOString().replace(/[^0-9]/gu, "").slice(0, 14); - return `web-${slug(caseId)}-${stamp}-${randomUUID().slice(0, 8)}`; -} - -function slug(value: string) { - return String(value).trim().toLowerCase().replace(/[^a-z0-9_.:-]+/gu, "-").replace(/^-+|-+$/gu, "") || "caserun"; -} - -function sha256Json(value: unknown) { - return createHash("sha256").update(JSON.stringify(value)).digest("hex"); -} - -function relativeOrAbsolute(cwd: string, target: string) { - const relative = path.relative(cwd, target) || "."; - return relative.startsWith("..") ? target : relative; } diff --git a/internal/harnessrl/activities.ts b/internal/harnessrl/activities.ts new file mode 100644 index 00000000..f60f50bf --- /dev/null +++ b/internal/harnessrl/activities.ts @@ -0,0 +1,112 @@ +import { heartbeat } from "@temporalio/activity"; +import { createHash } from "node:crypto"; +import { mkdir, readFile, writeFile } from "node:fs/promises"; +import path from "node:path"; + +import { buildCaseRun, collectCaseRun, prepareCaseRun } from "../../tools/src/hwlab-caserun-lib.ts"; +import type { HarnessRLRegistry } from "./contracts.ts"; +import { caseRunInternalFetch } from "../cloud/server-caserun-http.ts"; + +export function createHarnessRLActivities(options: { registry: HarnessRLRegistry; env?: Record }) { + const env = options.env ?? process.env; + const registry = options.registry; + return { + async markRunning({ runId }: { runId: string }) { + await transition(registry, runId, "running", "prepare", { worker: "hwlab-harnessrl-worker" }); + }, + async prepare(input: { runId: string; identity: string }) { + return idempotent(registry, input, async () => { + const record = required(await registry.getRun(input.runId)); + const definition = await loadDefinition(record.caseRepo, record.caseId); + activityHeartbeat({ runId: input.runId, stage: "prepare" }); + if (definition.mode === "software-smoke") { + await mkdir(record.runDir, { recursive: true }); + const output = { mode: "software-smoke", run: { runId: record.runId, caseId: record.caseId, mode: "software-smoke", hardwareRequired: false } }; + await transition(registry, input.runId, "running", "prepared", { mode: "software-smoke" }); + return output; + } + const context = contextForRecord(record, env); + const prepared = await prepareCaseRun(context, "harnessrl-worker"); + const output = { mode: definition.mode, prepared } as Record; + await transition(registry, input.runId, "running", "prepared", { runDir: prepared.runDir, specPath: prepared.specPath }); + return output; + }); + }, + async build(input: { runId: string; identity: string }) { + return idempotent(registry, input, async () => { + const record = required(await registry.getRun(input.runId)); + const prepared = required(await registry.getActivityResult(input.runId, "prepare", `${input.runId}:prepare:v1`)).output as any; + activityHeartbeat({ runId: input.runId, stage: "build" }); + if (prepared.mode === "software-smoke") { + const artifact = { + contractVersion: "hwlab-caserun-software-smoke-v1", caseId: record.caseId, runId: record.runId, + mode: "software-smoke", status: "completed", hardwareRequired: false, + checks: { caseDefinitionLoaded: true, runDirectoryWritable: true, temporalWorkerRunning: true } + }; + const text = `${JSON.stringify(artifact, null, 2)}\n`; + await writeFile(path.join(record.runDir, "software-smoke.json"), text, "utf8"); + const artifactRef = { path: "software-smoke.json", bytes: Buffer.byteLength(text), sha256: createHash("sha256").update(text).digest("hex") }; + const output = { mode: "software-smoke", run: prepared.run, summary: { status: "completed", mode: "software-smoke", artifactCount: 1, artifacts: [artifactRef] }, evidence: { status: "recorded", mode: "software-smoke", artifacts: [artifactRef] } }; + await transition(registry, input.runId, "running", "build-completed", { artifactSha256: artifactRef.sha256 }); + return output; + } + const context = contextForRecord(record, env); + const build = await buildCaseRun(context, prepared.prepared.run); + await transition(registry, input.runId, "running", "build-completed", { jobId: build.summary?.jobId ?? build.evidence?.keilJob?.jobId ?? null }); + return { mode: prepared.mode, run: build.run ?? prepared.prepared.run, build }; + }); + }, + async collect(input: { runId: string; identity: string }) { + return idempotent(registry, input, async () => { + const record = required(await registry.getRun(input.runId)); + const built = required(await registry.getActivityResult(input.runId, "build", `${input.runId}:build:v1`)).output as any; + activityHeartbeat({ runId: input.runId, stage: "collect" }); + let result: Record; + if (built.mode === "software-smoke") result = { summary: built.summary, evidence: built.evidence, run: built.run }; + else { + const context = contextForRecord(record, env); + const collected = await collectCaseRun(context, built.run, built.build.evidence); + result = { summary: built.build.summary, evidence: built.build.evidence, run: collected.run ?? built.run, collected: { status: collected.status, summary: collected.summary ?? null } }; + } + await registry.updateRun(input.runId, { result }); + await transition(registry, input.runId, "running", "collect-completed", {}); + return result; + }); + }, + 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 function idempotent(registry: HarnessRLRegistry, input: { runId: string; identity: string }, execute: () => Promise>) { + const activityName = input.identity.split(":").at(-2) ?? "activity"; + const existing = await registry.getActivityResult(input.runId, activityName, input.identity); + if (existing) return existing.output; + const output = await execute(); + return (await registry.saveActivityResult({ runId: input.runId, activityName, identity: input.identity, output })).output; +} + +async function transition(registry: HarnessRLRegistry, runId: string, status: any, stage: string, payload: Record) { + await registry.updateRun(runId, { status, stage }); + await registry.appendEvent(runId, status, stage, payload); +} + +function contextForRecord(record: any, env: Record) { + const apiUrl = String(env.HWLAB_CASERUN_INTERNAL_API_URL ?? env.HWLAB_RUNTIME_INTERNAL_API_URL ?? record.runtimeApiUrl).replace(/\/+$/u, ""); + return { + parsed: { _: [], caseId: record.caseId, runId: record.runId, runDir: record.runDir, caseRepo: record.caseRepo, apiUrl, noCaseRepoRecord: true, 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" }, + fetchImpl: caseRunInternalFetch(globalThis.fetch, apiUrl, env), cwd: record.caseRepo, + now: () => new Date().toISOString(), sleep: (ms: number) => new Promise((resolve) => setTimeout(resolve, ms)), argv: [], rest: ["worker", record.caseId] + }; +} + +async function loadDefinition(caseRepo: string, caseId: string) { + return JSON.parse(await readFile(path.join(caseRepo, "cases", caseId, "case.json"), "utf8")); +} +function required(value: T | null): T { if (value === null) throw Object.assign(new Error("required registry record was not found"), { code: "registry_record_not_found" }); return value; } +function activityHeartbeat(details: Record) { try { heartbeat(details); } catch {} } diff --git a/internal/harnessrl/contracts.ts b/internal/harnessrl/contracts.ts new file mode 100644 index 00000000..608e2669 --- /dev/null +++ b/internal/harnessrl/contracts.ts @@ -0,0 +1,67 @@ +export type CaseRunStatus = "queued" | "running" | "completed" | "failed" | "blocked" | "cancel_requested" | "canceled"; + +export type CaseRunEvent = { + id: number; + runId: string; + status: CaseRunStatus; + stage: string; + payload: Record; + createdAt: string; +}; + +export type CaseRunRecord = { + runId: string; + caseId: string; + workflowId: string; + workflowRunId: string | null; + status: CaseRunStatus; + stage: string; + terminal: boolean; + caseRepo: string; + runDir: string; + runtimeApiUrl: string; + result: Record | null; + error: { code: string; message: string; details?: unknown } | null; + blocker: Record | null; + createdAt: string; + updatedAt: string; +}; + +export type StartCaseRunInput = { + runId: string; + caseId: string; + workflowId: string; + caseRepo: string; + runDir: string; + runtimeApiUrl: string; +}; + +export type ActivityResult = { + runId: string; + activityName: string; + identity: string; + output: Record; +}; + +export interface HarnessRLRegistry { + ensureSchema(): Promise; + createRun(input: StartCaseRunInput): Promise; + getRun(runId: string): Promise; + updateRun(runId: string, patch: Partial>): Promise; + appendEvent(runId: string, status: CaseRunStatus, stage: string, payload?: Record): Promise; + listEvents(runId: string): Promise; + getActivityResult(runId: string, activityName: string, identity: string): Promise; + saveActivityResult(result: ActivityResult): Promise; + health(): Promise<{ ok: true }>; + close(): Promise; +} + +export interface HarnessRLTemporalGateway { + start(input: { workflowId: string; runId: string }): Promise<{ workflowRunId: string }>; + requestCancel(workflowId: string): Promise; + close(): Promise; +} + +export function terminalStatus(status: CaseRunStatus) { + return status === "completed" || status === "failed" || status === "blocked" || status === "canceled"; +} diff --git a/internal/harnessrl/harnessrl.test.ts b/internal/harnessrl/harnessrl.test.ts new file mode 100644 index 00000000..fad960ba --- /dev/null +++ b/internal/harnessrl/harnessrl.test.ts @@ -0,0 +1,129 @@ +import assert from "node:assert/strict"; +import { mkdtemp, mkdir, readFile, writeFile } from "node:fs/promises"; +import os from "node:os"; +import path from "node:path"; +import { test } from "node:test"; + +import { createHarnessRLActivities } from "./activities.ts"; +import type { ActivityResult, CaseRunEvent, CaseRunRecord, CaseRunStatus, HarnessRLRegistry, HarnessRLTemporalGateway, StartCaseRunInput } from "./contracts.ts"; +import { terminalStatus } from "./contracts.ts"; +import { createHarnessRLHttpApp } from "./http.ts"; +import { createHarnessRLService } from "./service.ts"; + +test("API restart preserves PostgreSQL-shaped registry read model", async () => { + const fixture = await softwareSmokeFixture(); + const registry = new MemoryRegistry(); + const temporal = new FakeTemporal(); + const first = createHarnessRLService({ registry, temporal, cwd: fixture.root, caseRepo: fixture.root, stateRoot: fixture.stateRoot }); + const app = createHarnessRLHttpApp({ service: first }); + const response = await app.fetch(new Request("http://harnessrl.test/v1/caserun/runs", { + method: "POST", headers: { "content-type": "application/json" }, body: JSON.stringify({ caseId: fixture.caseId, runId: "restart-run" }) + })); + assert.equal(response.status, 202); + assert.equal(temporal.starts, 1); + + const restarted = createHarnessRLService({ registry, temporal: new FakeTemporal(), cwd: fixture.root, caseRepo: fixture.root, stateRoot: fixture.stateRoot }); + const record = await restarted.getRun("restart-run"); + assert.equal(record.status, "queued"); + assert.equal(record.workflowRunId, "temporal-run-1"); +}); + +test("worker restart reuses idempotent prepare/build/collect activity results", async () => { + const fixture = await softwareSmokeFixture(); + const registry = new MemoryRegistry(); + await registry.createRun({ runId: "worker-restart", caseId: fixture.caseId, workflowId: "workflow-worker-restart", caseRepo: fixture.root, runDir: path.join(fixture.stateRoot, "runs", "worker-restart"), runtimeApiUrl: "http://127.0.0.1:6667" }); + const firstWorker = createHarnessRLActivities({ registry, env: {} }); + await firstWorker.markRunning({ runId: "worker-restart" }); + const prepared = await firstWorker.prepare({ runId: "worker-restart", identity: "worker-restart:prepare:v1" }); + const built = await firstWorker.build({ runId: "worker-restart", identity: "worker-restart:build:v1" }); + + const restartedWorker = createHarnessRLActivities({ registry, env: {} }); + assert.deepEqual(await restartedWorker.prepare({ runId: "worker-restart", identity: "worker-restart:prepare:v1" }), prepared); + assert.deepEqual(await restartedWorker.build({ runId: "worker-restart", identity: "worker-restart:build:v1" }), built); + await restartedWorker.collect({ runId: "worker-restart", identity: "worker-restart:collect:v1" }); + await restartedWorker.markCompleted({ runId: "worker-restart" }); + + const record = await registry.getRun("worker-restart"); + assert.equal(record?.status, "completed"); + assert.equal((record?.result as any)?.summary?.mode, "software-smoke"); + const artifact = JSON.parse(await readFile(path.join(record!.runDir, "software-smoke.json"), "utf8")); + assert.equal(artifact.checks.temporalWorkerRunning, true); + assert.equal(registry.activityResults.size, 3); +}); + +test("cancel converges to durable canceled through workflow signal", async () => { + const fixture = await softwareSmokeFixture(); + const registry = new MemoryRegistry(); + const temporal = new FakeTemporal(async (workflowId) => { + const record = [...registry.runs.values()].find((item) => item.workflowId === workflowId)!; + await registry.updateRun(record.runId, { status: "canceled", stage: "canceled" }); + await registry.appendEvent(record.runId, "canceled", "canceled", { source: "temporal-signal" }); + }); + const service = createHarnessRLService({ registry, temporal, cwd: fixture.root, caseRepo: fixture.root, stateRoot: fixture.stateRoot }); + await service.submit({ caseId: fixture.caseId, runId: "cancel-run", runtimeApiUrl: "http://harnessrl.test" }); + const canceled = await service.cancel("cancel-run"); + assert.equal(canceled?.status, "canceled"); + assert.equal(canceled?.terminal, true); + assert.deepEqual((await registry.listEvents("cancel-run")).map((event) => event.status), ["queued", "cancel_requested", "canceled"]); +}); + +test("read model keeps workflow identity separate from result authority", async () => { + const fixture = await softwareSmokeFixture(); + const registry = new MemoryRegistry(); + const service = createHarnessRLService({ registry, temporal: new FakeTemporal(), cwd: fixture.root, caseRepo: fixture.root, stateRoot: fixture.stateRoot }); + await service.submit({ caseId: fixture.caseId, runId: "read-model", runtimeApiUrl: "http://harnessrl.test" }); + await registry.updateRun("read-model", { status: "completed", stage: "completed", result: { artifactManifestSha256: "a".repeat(64), sourceAuthority: "hwpod" } }); + const record = await service.getRun("read-model"); + assert.equal(record.workflowId, "harnessrl-caserun-read-model"); + assert.equal(record.status, "completed"); + assert.equal(record.result?.sourceAuthority, "hwpod"); +}); + +async function softwareSmokeFixture() { + const root = await mkdtemp(path.join(os.tmpdir(), "harnessrl-native-")); + const caseId = "software-smoke"; + await mkdir(path.join(root, "cases", caseId), { recursive: true }); + await writeFile(path.join(root, "cases", caseId, "case.json"), `${JSON.stringify({ caseId, title: "Native software smoke", mode: "software-smoke" }, null, 2)}\n`); + return { root, caseId, stateRoot: path.join(root, ".state", "harnessrl") }; +} + +class FakeTemporal implements HarnessRLTemporalGateway { + starts = 0; + constructor(private readonly onCancel?: (workflowId: string) => Promise) {} + async start() { this.starts += 1; return { workflowRunId: `temporal-run-${this.starts}` }; } + async requestCancel(workflowId: string) { await this.onCancel?.(workflowId); } + async close() {} +} + +class MemoryRegistry implements HarnessRLRegistry { + runs = new Map(); + events = new Map(); + activityResults = new Map(); + async ensureSchema() {} + async createRun(input: StartCaseRunInput) { + const existing = this.runs.get(input.runId); + if (existing) return structuredClone(existing); + const now = new Date().toISOString(); + const record: CaseRunRecord = { ...input, workflowRunId: null, status: "queued", stage: "queued", terminal: false, result: null, error: null, blocker: null, createdAt: now, updatedAt: now }; + this.runs.set(input.runId, record); + return structuredClone(record); + } + async getRun(runId: string) { const value = this.runs.get(runId); return value ? structuredClone(value) : null; } + async updateRun(runId: string, patch: any) { + const current = this.runs.get(runId); + if (!current) throw new Error("missing run"); + const next = { ...current, ...patch, terminal: terminalStatus((patch.status ?? current.status) as CaseRunStatus), updatedAt: new Date().toISOString() }; + this.runs.set(runId, next); + return structuredClone(next); + } + async appendEvent(runId: string, status: CaseRunStatus, stage: string, payload: Record = {}) { + const events = this.events.get(runId) ?? []; + const event = { id: events.length + 1, runId, status, stage, payload, createdAt: new Date().toISOString() }; + events.push(event); this.events.set(runId, events); return structuredClone(event); + } + async listEvents(runId: string) { return structuredClone(this.events.get(runId) ?? []); } + async getActivityResult(runId: string, activityName: string, identity: string) { return structuredClone(this.activityResults.get(`${runId}:${activityName}:${identity}`) ?? null); } + async saveActivityResult(result: ActivityResult) { const key = `${result.runId}:${result.activityName}:${result.identity}`; if (!this.activityResults.has(key)) this.activityResults.set(key, structuredClone(result)); return structuredClone(this.activityResults.get(key)!); } + async health() { return { ok: true as const }; } + async close() {} +} diff --git a/internal/harnessrl/http.ts b/internal/harnessrl/http.ts new file mode 100644 index 00000000..13aef424 --- /dev/null +++ b/internal/harnessrl/http.ts @@ -0,0 +1,47 @@ +export function createHarnessRLHttpApp(options: { service: ReturnType }) { + return { + async fetch(request: Request) { + const url = new URL(request.url); + try { + if (url.pathname === "/healthz" || url.pathname === "/health/live") return json(200, { ok: true, serviceId: "hwlab-harnessrl-api" }); + if (url.pathname === "/readyz" || url.pathname === "/health/ready") return json(200, await options.service.health()); + if (url.pathname === "/v1/caserun" && request.method === "GET") return json(200, { ok: true, contractVersion: "hwlab-caserun-web-v1", casesRoute: "/v1/caserun/cases", runsRoute: "/v1/caserun/runs", executionAuthority: "temporal", registryAuthority: "postgresql" }); + if (url.pathname === "/v1/caserun/cases" && request.method === "GET") { + const cases = await options.service.listCases(); + return json(200, { ok: true, contractVersion: "hwlab-caserun-cases-v1", count: cases.length, cases }); + } + if (url.pathname === "/v1/caserun/runs" && request.method === "POST") { + const body = await bodyObject(request); + const record = await options.service.submit({ caseId: String(body.caseId ?? ""), runId: body.runId ? String(body.runId) : undefined, runtimeApiUrl: runtimeApiUrl(request) }); + return json(202, accepted(record)); + } + const match = /^\/v1\/caserun\/runs\/([^/]+)(?:\/(events|aggregate|cancel))?$/u.exec(url.pathname); + if (match) { + const runId = decodeURIComponent(match[1]); + const suffix = match[2] ?? ""; + if (!suffix && request.method === "GET") return json(200, await options.service.getRun(runId)); + if (suffix === "events" && request.method === "GET") return json(200, await options.service.events(runId)); + if (suffix === "aggregate" && request.method === "GET") return json(200, await options.service.aggregate(runId)); + if (suffix === "cancel" && request.method === "POST") return json(202, await options.service.cancel(runId)); + return json(405, errorBody("method_not_allowed", suffix === "cancel" ? "POST required" : "GET required")); + } + return json(404, errorBody("caserun_route_not_found", "CaseRun REST route is not implemented")); + } catch (error: any) { + const code = error?.code ?? "harnessrl_error"; + const status = code.includes("not_found") ? 404 : code.startsWith("invalid_") ? 400 : 500; + return json(status, errorBody(code, error?.message ?? String(error), error?.details)); + } + }, + close: () => options.service.close() + }; +} + +function accepted(record: any) { + return { ok: true, contractVersion: "hwlab-caserun-web-v1", runId: record.runId, caseId: record.caseId, status: record.status, stage: record.stage, + statusUrl: `/v1/caserun/runs/${encodeURIComponent(record.runId)}`, eventsUrl: `/v1/caserun/runs/${encodeURIComponent(record.runId)}/events`, + aggregateUrl: `/v1/caserun/runs/${encodeURIComponent(record.runId)}/aggregate`, cancelUrl: `/v1/caserun/runs/${encodeURIComponent(record.runId)}/cancel` }; +} +async function bodyObject(request: Request) { const body = await request.json().catch(() => null); if (!body || typeof body !== "object" || Array.isArray(body)) throw Object.assign(new Error("JSON object body is required"), { code: "invalid_json" }); return body as Record; } +function runtimeApiUrl(request: Request) { const url = new URL(request.url); return `${url.protocol}//${url.host}`; } +function errorBody(code: string, message: string, details?: unknown) { return { ok: false, error: { code, message, details } }; } +function json(status: number, body: unknown) { return Response.json(body, { status, headers: { "cache-control": "no-store" } }); } diff --git a/internal/harnessrl/migrations/0001_harnessrl_registry.sql b/internal/harnessrl/migrations/0001_harnessrl_registry.sql new file mode 100644 index 00000000..b68dca0f --- /dev/null +++ b/internal/harnessrl/migrations/0001_harnessrl_registry.sql @@ -0,0 +1,45 @@ +CREATE TABLE IF NOT EXISTS harnessrl_schema_migrations ( + migration_id TEXT PRIMARY KEY, + applied_at TIMESTAMPTZ NOT NULL DEFAULT now() +); + +CREATE TABLE IF NOT EXISTS harnessrl_runs ( + run_id TEXT PRIMARY KEY, + case_id TEXT NOT NULL, + workflow_id TEXT NOT NULL UNIQUE, + workflow_run_id TEXT, + status TEXT NOT NULL, + stage TEXT NOT NULL, + case_repo TEXT NOT NULL, + run_dir TEXT NOT NULL, + runtime_api_url TEXT NOT NULL, + result JSONB, + error JSONB, + blocker JSONB, + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + updated_at TIMESTAMPTZ NOT NULL DEFAULT now() +); + +CREATE TABLE IF NOT EXISTS harnessrl_run_events ( + id BIGSERIAL PRIMARY KEY, + run_id TEXT NOT NULL REFERENCES harnessrl_runs(run_id) ON DELETE CASCADE, + status TEXT NOT NULL, + stage TEXT NOT NULL, + payload JSONB NOT NULL DEFAULT '{}'::jsonb, + created_at TIMESTAMPTZ NOT NULL DEFAULT now() +); + +CREATE TABLE IF NOT EXISTS harnessrl_activity_results ( + run_id TEXT NOT NULL REFERENCES harnessrl_runs(run_id) ON DELETE CASCADE, + activity_name TEXT NOT NULL, + identity TEXT NOT NULL, + output JSONB NOT NULL, + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + PRIMARY KEY (run_id, activity_name, identity) +); + +CREATE INDEX IF NOT EXISTS idx_harnessrl_events_run_id ON harnessrl_run_events(run_id, id); + +INSERT INTO harnessrl_schema_migrations (migration_id) +VALUES ('0001_harnessrl_registry') +ON CONFLICT DO NOTHING; diff --git a/internal/harnessrl/registry.ts b/internal/harnessrl/registry.ts new file mode 100644 index 00000000..7550bb58 --- /dev/null +++ b/internal/harnessrl/registry.ts @@ -0,0 +1,134 @@ +import pg from "pg"; +import { readFile } from "node:fs/promises"; + +import type { ActivityResult, CaseRunEvent, CaseRunRecord, CaseRunStatus, HarnessRLRegistry, StartCaseRunInput } from "./contracts.ts"; +import { terminalStatus } from "./contracts.ts"; + +const { Pool } = pg; + +const migrationUrl = new URL("./migrations/0001_harnessrl_registry.sql", import.meta.url); + +export class PostgresHarnessRLRegistry implements HarnessRLRegistry { + private readonly pool: pg.Pool; + private ready?: Promise; + + constructor(databaseUrl: string) { + if (!databaseUrl) throw codedError("harnessrl_database_url_required", "HARNESSRL_DATABASE_URL is required"); + this.pool = new Pool({ connectionString: databaseUrl, max: 8 }); + } + + ensureSchema() { + this.ready ??= this.migrate(); + return this.ready; + } + + private async migrate() { + const client = await this.pool.connect(); + try { + await client.query("BEGIN"); + await client.query(await readFile(migrationUrl, "utf8")); + await client.query("COMMIT"); + } catch (error) { + await client.query("ROLLBACK"); + throw error; + } finally { + client.release(); + } + } + + async createRun(input: StartCaseRunInput) { + await this.ensureSchema(); + const result = await this.pool.query( + `INSERT INTO harnessrl_runs (run_id,case_id,workflow_id,status,stage,case_repo,run_dir,runtime_api_url) + VALUES ($1,$2,$3,'queued','queued',$4,$5,$6) + ON CONFLICT (run_id) DO UPDATE SET run_id=EXCLUDED.run_id + RETURNING *`, + [input.runId, input.caseId, input.workflowId, input.caseRepo, input.runDir, input.runtimeApiUrl] + ); + return runRow(result.rows[0]); + } + + async getRun(runId: string) { + await this.ensureSchema(); + const result = await this.pool.query("SELECT * FROM harnessrl_runs WHERE run_id=$1", [runId]); + return result.rows[0] ? runRow(result.rows[0]) : null; + } + + async updateRun(runId: string, patch: Partial>) { + await this.ensureSchema(); + const current = await this.getRun(runId); + if (!current) throw codedError("caserun_run_not_found", "CaseRun run was not found"); + const result = await this.pool.query( + `UPDATE harnessrl_runs SET workflow_run_id=$2,status=$3,stage=$4,result=$5,error=$6,blocker=$7,updated_at=now() + WHERE run_id=$1 RETURNING *`, + [runId, patch.workflowRunId === undefined ? current.workflowRunId : patch.workflowRunId, + patch.status ?? current.status, patch.stage ?? current.stage, + patch.result === undefined ? current.result : patch.result, + patch.error === undefined ? current.error : patch.error, + patch.blocker === undefined ? current.blocker : patch.blocker] + ); + return runRow(result.rows[0]); + } + + async appendEvent(runId: string, status: CaseRunStatus, stage: string, payload: Record = {}) { + await this.ensureSchema(); + const result = await this.pool.query( + "INSERT INTO harnessrl_run_events (run_id,status,stage,payload) VALUES ($1,$2,$3,$4) RETURNING *", + [runId, status, stage, payload] + ); + return eventRow(result.rows[0]); + } + + async listEvents(runId: string) { + await this.ensureSchema(); + const result = await this.pool.query("SELECT * FROM harnessrl_run_events WHERE run_id=$1 ORDER BY id", [runId]); + return result.rows.map(eventRow); + } + + async getActivityResult(runId: string, activityName: string, identity: string) { + await this.ensureSchema(); + const result = await this.pool.query( + "SELECT * FROM harnessrl_activity_results WHERE run_id=$1 AND activity_name=$2 AND identity=$3", + [runId, activityName, identity] + ); + return result.rows[0] ? activityRow(result.rows[0]) : null; + } + + async saveActivityResult(result: ActivityResult) { + await this.ensureSchema(); + const saved = await this.pool.query( + `INSERT INTO harnessrl_activity_results (run_id,activity_name,identity,output) VALUES ($1,$2,$3,$4) + ON CONFLICT (run_id,activity_name,identity) DO UPDATE SET output=harnessrl_activity_results.output RETURNING *`, + [result.runId, result.activityName, result.identity, result.output] + ); + return activityRow(saved.rows[0]); + } + + async health() { + await this.ensureSchema(); + await this.pool.query("SELECT 1"); + return { ok: true as const }; + } + + async close() { await this.pool.end(); } +} + +function runRow(row: any): CaseRunRecord { + const status = row.status as CaseRunStatus; + return { + runId: row.run_id, caseId: row.case_id, workflowId: row.workflow_id, workflowRunId: row.workflow_run_id, + status, stage: row.stage, terminal: terminalStatus(status), caseRepo: row.case_repo, runDir: row.run_dir, + runtimeApiUrl: row.runtime_api_url, result: row.result, error: row.error, blocker: row.blocker, + createdAt: new Date(row.created_at).toISOString(), updatedAt: new Date(row.updated_at).toISOString() + }; +} + +function eventRow(row: any): CaseRunEvent { + return { id: Number(row.id), runId: row.run_id, status: row.status, stage: row.stage, payload: row.payload ?? {}, createdAt: new Date(row.created_at).toISOString() }; +} + +function activityRow(row: any): ActivityResult { + return { runId: row.run_id, activityName: row.activity_name, identity: row.identity, output: row.output ?? {} }; +} + +function codedError(code: string, message: string) { return Object.assign(new Error(message), { code }); } diff --git a/internal/harnessrl/runtime.ts b/internal/harnessrl/runtime.ts new file mode 100644 index 00000000..e06c7b3c --- /dev/null +++ b/internal/harnessrl/runtime.ts @@ -0,0 +1,20 @@ +import path from "node:path"; + +import { PostgresHarnessRLRegistry } from "./registry.ts"; +import { createHarnessRLService } from "./service.ts"; +import { createHarnessRLTemporalGateway } from "./temporal.ts"; + +export function harnessRLRuntime(env: Record = process.env) { + const cwd = path.resolve(String(env.HWLAB_CASERUN_REPO_ROOT ?? process.cwd())); + const caseRepo = path.resolve(cwd, String(env.HWLAB_CASERUN_CASE_REPO ?? env.HWLAB_CASE_REPO ?? ".")); + const stateRoot = path.resolve(cwd, String(env.HARNESSRL_STATE_DIR ?? ".state/harnessrl")); + const registry = new PostgresHarnessRLRegistry(String(env.HARNESSRL_DATABASE_URL ?? "")); + const temporalAddress = String(env.HARNESSRL_TEMPORAL_ADDRESS ?? env.TEMPORAL_ADDRESS ?? ""); + const temporalNamespace = String(env.HARNESSRL_TEMPORAL_NAMESPACE ?? "unidesk"); + const taskQueue = String(env.HARNESSRL_TEMPORAL_TASK_QUEUE ?? "hwlab-v03-harnessrl"); + const temporal = createHarnessRLTemporalGateway({ address: temporalAddress, namespace: temporalNamespace, taskQueue }); + return { + cwd, caseRepo, stateRoot, registry, temporalAddress, temporalNamespace, taskQueue, + service: createHarnessRLService({ registry, temporal, cwd, caseRepo, stateRoot }) + }; +} diff --git a/internal/harnessrl/service.ts b/internal/harnessrl/service.ts new file mode 100644 index 00000000..1cea755e --- /dev/null +++ b/internal/harnessrl/service.ts @@ -0,0 +1,82 @@ +import { randomUUID } from "node:crypto"; +import { mkdir, readFile, readdir, stat } from "node:fs/promises"; +import path from "node:path"; + +import type { HarnessRLRegistry, HarnessRLTemporalGateway } from "./contracts.ts"; + +const RUN_ID_PATTERN = /^[A-Za-z0-9_.:-]{3,180}$/u; + +export function createHarnessRLService(options: { registry: HarnessRLRegistry; temporal: HarnessRLTemporalGateway; cwd: string; caseRepo: string; stateRoot: string }) { + const { registry, temporal } = options; + return { + async health() { return { ok: true, registry: await registry.health() }; }, + async listCases() { + const root = path.join(options.caseRepo, "cases"); + const entries = await readdir(root, { withFileTypes: true }).catch(() => []); + const cases = []; + for (const entry of entries) if (entry.isDirectory()) { + const item = await loadCase(options.caseRepo, entry.name).catch(() => null); + if (item) cases.push(item); + } + return cases.sort((left, right) => left.caseId.localeCompare(right.caseId)); + }, + async submit(input: { caseId: string; runId?: string; runtimeApiUrl: string }) { + const caseId = opaqueId(input.caseId, "invalid_case_id"); + const definition = await loadCase(options.caseRepo, caseId).catch(() => null); + if (!definition) throw codedError("case_not_found", "CaseRun case was not found", { caseId }); + const runId = input.runId ? opaqueId(input.runId, "invalid_run_id") : `${caseId}-${Date.now()}-${randomUUID().slice(0, 8)}`; + const workflowId = `harnessrl-caserun-${runId}`; + const runDir = path.join(options.stateRoot, "runs", runId); + await mkdir(runDir, { recursive: true }); + const record = await registry.createRun({ runId, caseId, workflowId, caseRepo: options.caseRepo, runDir, runtimeApiUrl: input.runtimeApiUrl }); + if (!record.workflowRunId) { + await registry.appendEvent(runId, "queued", "queued", { caseId }); + const started = await temporal.start({ workflowId, runId }); + await registry.updateRun(runId, { workflowRunId: started.workflowRunId }); + } + return await registry.getRun(runId); + }, + async getRun(runId: string) { + const record = await registry.getRun(opaqueId(runId, "invalid_run_id")); + if (!record) throw codedError("caserun_run_not_found", "CaseRun run was not found", { runId }); + return record; + }, + async events(runId: string) { + const record = await this.getRun(runId); + return { ...record, events: await registry.listEvents(record.runId) }; + }, + async aggregate(runId: string) { + const record = await this.getRun(runId); + return { ok: true, contractVersion: "hwlab-caserun-aggregate-v1", runId: record.runId, caseId: record.caseId, + status: record.status, stage: record.stage, terminal: record.terminal, result: record.result, + sha256: record.result?.artifactManifestSha256 ?? null }; + }, + async cancel(runId: string) { + const record = await this.getRun(runId); + if (record.terminal) return record; + await registry.updateRun(runId, { status: "cancel_requested", stage: "cancel_requested" }); + await registry.appendEvent(runId, "cancel_requested", "cancel_requested", {}); + await temporal.requestCancel(record.workflowId); + return await registry.getRun(runId); + }, + async close() { await temporal.close(); await registry.close(); } + }; +} + +async function loadCase(caseRepo: string, caseId: string) { + const caseDir = path.join(caseRepo, "cases", caseId); + const definition = JSON.parse(await readFile(path.join(caseDir, "case.json"), "utf8")); + const mode = String(definition.mode ?? "compile-only").trim() || "compile-only"; + const hardwareRequired = mode !== "software-smoke"; + const hwpodSpec = hardwareRequired ? String(definition.hwpodSpec ?? definition.specPath ?? "hwpod-spec.yaml").trim() || "hwpod-spec.yaml" : null; + const available = hwpodSpec ? Boolean((await stat(path.resolve(caseDir, hwpodSpec)).catch(() => null))?.isFile()) : true; + return { caseId: String(definition.caseId ?? caseId), title: String(definition.title ?? caseId), mode, hwpodSpec, hardwareRequired, available, + subject: definition.subject ?? null, expected: definition.expected ?? null, runtime: definition.runtime ?? null }; +} + +function opaqueId(value: unknown, code: string) { + const text = String(value ?? "").trim(); + if (!RUN_ID_PATTERN.test(text)) throw codedError(code, `${code === "invalid_case_id" ? "caseId" : "runId"} is invalid`); + return text; +} +function codedError(code: string, message: string, details?: unknown) { return Object.assign(new Error(message), { code, details }); } diff --git a/internal/harnessrl/temporal.ts b/internal/harnessrl/temporal.ts new file mode 100644 index 00000000..3e4b9bae --- /dev/null +++ b/internal/harnessrl/temporal.ts @@ -0,0 +1,34 @@ +import { Client, Connection } from "@temporalio/client"; + +import type { HarnessRLTemporalGateway } from "./contracts.ts"; + +export function createHarnessRLTemporalGateway(options: { address: string; namespace: string; taskQueue: string }): HarnessRLTemporalGateway { + if (!options.address) throw codedError("harnessrl_temporal_address_required", "HARNESSRL_TEMPORAL_ADDRESS is required"); + let connection: Connection | undefined; + let client: Client | undefined; + const getClient = async () => { + connection ??= await Connection.connect({ address: options.address }); + client ??= new Client({ connection, namespace: options.namespace }); + return client; + }; + return { + async start(input) { + const handle = await (await getClient()).workflow.start("caseRunWorkflow", { + taskQueue: options.taskQueue, + workflowId: input.workflowId, + args: [{ runId: input.runId }] + }); + return { workflowRunId: handle.firstExecutionRunId }; + }, + async requestCancel(workflowId) { + await (await getClient()).workflow.getHandle(workflowId).signal("cancelCaseRun"); + }, + async close() { + await connection?.close(); + connection = undefined; + client = undefined; + } + }; +} + +function codedError(code: string, message: string) { return Object.assign(new Error(message), { code }); } diff --git a/internal/harnessrl/workflows.ts b/internal/harnessrl/workflows.ts new file mode 100644 index 00000000..3995be71 --- /dev/null +++ b/internal/harnessrl/workflows.ts @@ -0,0 +1,44 @@ +import { defineSignal, proxyActivities, setHandler } from "@temporalio/workflow"; + +type Activities = { + markRunning(input: { runId: string }): Promise; + prepare(input: { runId: string; identity: string }): Promise>; + build(input: { runId: string; identity: string }): Promise>; + collect(input: { runId: string; identity: string }): Promise>; + markCompleted(input: { runId: string }): Promise; + markCanceled(input: { runId: string }): Promise; + markFailed(input: { runId: string; code: string; message: string }): Promise; +}; + +const activities = proxyActivities({ + startToCloseTimeout: "30 minutes", + heartbeatTimeout: "30 seconds", + retry: { initialInterval: "1 second", maximumInterval: "30 seconds", maximumAttempts: 5 } +}); + +export const cancelCaseRun = defineSignal("cancelCaseRun"); + +export async function caseRunWorkflow(input: { runId: string }) { + let cancelRequested = false; + setHandler(cancelCaseRun, () => { cancelRequested = true; }); + const canceled = async () => { + if (!cancelRequested) return false; + await activities.markCanceled({ runId: input.runId }); + return true; + }; + try { + await activities.markRunning(input); + if (await canceled()) return { status: "canceled", runId: input.runId }; + await activities.prepare({ runId: input.runId, identity: `${input.runId}:prepare:v1` }); + if (await canceled()) return { status: "canceled", runId: input.runId }; + await activities.build({ runId: input.runId, identity: `${input.runId}:build:v1` }); + if (await canceled()) return { status: "canceled", runId: input.runId }; + await activities.collect({ runId: input.runId, identity: `${input.runId}:collect:v1` }); + if (await canceled()) return { status: "canceled", runId: input.runId }; + 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) }); + throw error; + } +} diff --git a/package.json b/package.json index cfa99dea..9edb7f0b 100644 --- a/package.json +++ b/package.json @@ -12,6 +12,9 @@ "check:plan": "node scripts/check-runner.mjs --profile check --list", "check:cloud-api": "node scripts/check-runner.mjs --profile check --group cloud-api", "check:cloud-web": "node scripts/check-runner.mjs --profile check --group cloud-web", + "harnessrl:api:dev": "bun --watch cmd/hwlab-harnessrl-api/main.ts", + "harnessrl:worker:dev": "bun --watch cmd/hwlab-harnessrl-worker/main.ts", + "harnessrl:native:smoke": "bun test internal/harnessrl/harnessrl.test.ts internal/cloud/server-caserun-http.test.ts", "workbench:p1:inventory": "node scripts/workbench-long-reliability-p1-inventory.mjs", "dev-runtime-base:build": "node scripts/dev-runtime-base-image.mjs", "cloud-api:smoke": "node scripts/run-bun.mjs scripts/cloud-api-runtime-smoke.mjs",