Merge pull request #2784 from pikasTech/feat/embdagentbench-native-caserun
feat: CaseRun 原生 API/worker/Web 聚合生命周期
This commit is contained in:
Lyon
@@ -15,11 +15,11 @@ if (argv[0] === "tasktree") {
|
||||
process.exitCode = 1;
|
||||
}
|
||||
} else if (argv[0] === "caserun") {
|
||||
const { caserunNativeServiceCommand } = await import("../../src/caserun-native-service.ts");
|
||||
const { caserunNativeServiceCommand, caserunNativeServicesCommand } = await import("../../src/caserun-native-service.ts");
|
||||
try {
|
||||
const service = argv[1] === "service" ? argv[2] : "";
|
||||
const action = argv[1] === "service" ? argv[3] ?? "status" : "status";
|
||||
const result = await caserunNativeServiceCommand({ service, action, cwd: process.cwd(), env: process.env });
|
||||
const result = argv[1] === "service" && ["start", "stop", "restart", "status", "logs"].includes(argv[2] ?? "") && argv[3] === undefined
|
||||
? await caserunNativeServicesCommand({ action: argv[2], cwd: process.cwd(), env: process.env })
|
||||
: await caserunNativeServiceCommand({ service: argv[1] === "service" ? argv[2] ?? "" : "", action: argv[1] === "service" ? argv[3] ?? "status" : "status", cwd: process.cwd(), env: process.env });
|
||||
console.log(JSON.stringify(result, null, 2));
|
||||
process.exitCode = result?.ok === false ? 1 : 0;
|
||||
} catch (error: any) {
|
||||
|
||||
@@ -0,0 +1,49 @@
|
||||
import { mkdtemp, rm } from "node:fs/promises";
|
||||
import { tmpdir } from "node:os";
|
||||
import path from "node:path";
|
||||
|
||||
import { expect, test } from "bun:test";
|
||||
|
||||
import { caserunNativeServiceOrder, caserunNativeServicesCommand } from "./caserun-native-service.ts";
|
||||
|
||||
test("CaseRun 聚合状态同时披露 API、worker 和 Web", async () => {
|
||||
const cwd = await mkdtemp(path.join(tmpdir(), "caserun-native-status-"));
|
||||
try {
|
||||
const result = await caserunNativeServicesCommand({ action: "status", cwd, env: { CASERUN_NATIVE_SERVICE_STATE_DIR: ".state/services" } });
|
||||
expect(result).toMatchObject({
|
||||
ok: false,
|
||||
operation: "caserun.service.status",
|
||||
status: "stopped",
|
||||
summary: { total: 3, running: 0, unavailable: 3 },
|
||||
services: {
|
||||
api: { service: "api", status: "stopped" },
|
||||
worker: { service: "worker", status: "stopped" },
|
||||
web: { service: "web", status: "stopped" },
|
||||
},
|
||||
valuesPrinted: false,
|
||||
});
|
||||
} finally {
|
||||
await rm(cwd, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
|
||||
test("CaseRun 聚合生命周期保持依赖顺序并幂等停止", async () => {
|
||||
const cwd = await mkdtemp(path.join(tmpdir(), "caserun-native-stop-"));
|
||||
try {
|
||||
expect(caserunNativeServiceOrder("start")).toEqual(["worker", "api", "web"]);
|
||||
expect(caserunNativeServiceOrder("restart")).toEqual(["worker", "api", "web"]);
|
||||
expect(caserunNativeServiceOrder("stop")).toEqual(["web", "api", "worker"]);
|
||||
|
||||
const result = await caserunNativeServicesCommand({ action: "stop", cwd, env: { CASERUN_NATIVE_SERVICE_STATE_DIR: ".state/services" } });
|
||||
expect(result).toMatchObject({
|
||||
ok: true,
|
||||
operation: "caserun.service.stop",
|
||||
status: "stopped",
|
||||
summary: { total: 3, succeeded: 3, failed: 0 },
|
||||
});
|
||||
expect(Object.keys(result.services)).toEqual(["web", "api", "worker"]);
|
||||
expect(Object.values(result.services).every((service: any) => service.alreadyStopped === true)).toBe(true);
|
||||
} finally {
|
||||
await rm(cwd, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
@@ -4,7 +4,66 @@ import { createServer } from "node:net";
|
||||
import { mkdir, readFile, readdir, rm, writeFile } from "node:fs/promises";
|
||||
import path from "node:path";
|
||||
|
||||
type ServiceName = "api" | "web";
|
||||
export type ServiceName = "api" | "worker" | "web";
|
||||
type ServiceAction = "start" | "stop" | "restart" | "status" | "logs";
|
||||
const SERVICE_NAMES: ServiceName[] = ["api", "worker", "web"];
|
||||
const SERVICE_START_ORDER: ServiceName[] = ["worker", "api", "web"];
|
||||
const SERVICE_STOP_ORDER: ServiceName[] = ["web", "api", "worker"];
|
||||
|
||||
export async function caserunNativeServicesCommand(input: { action: string; cwd: string; env: Record<string, string | undefined> }) {
|
||||
const action = validServiceAction(input.action);
|
||||
const order = action === "stop" ? SERVICE_STOP_ORDER : action === "start" || action === "restart" ? SERVICE_START_ORDER : SERVICE_NAMES;
|
||||
const rawResults = action === "status" || action === "logs"
|
||||
? await Promise.all(order.map((service) => safeServiceCommand(service, action, input)))
|
||||
: await runServiceCommandsInOrder(order, action, input);
|
||||
const results = action === "logs" ? rawResults.map(boundedAggregateLogs) : rawResults;
|
||||
const succeeded = results.filter((result) => result.ok).length;
|
||||
const running = results.filter((result) => result.status === "running").length;
|
||||
return {
|
||||
ok: action === "start" || action === "restart" || action === "stop" ? succeeded === SERVICE_NAMES.length : running === SERVICE_NAMES.length,
|
||||
operation: `caserun.service.${action}`,
|
||||
action,
|
||||
status: action === "stop" && succeeded === SERVICE_NAMES.length
|
||||
? "stopped"
|
||||
: running === SERVICE_NAMES.length
|
||||
? "running"
|
||||
: running === 0 && results.every((result) => result.status === "stopped")
|
||||
? "stopped"
|
||||
: action === "start" || action === "restart"
|
||||
? "starting"
|
||||
: "degraded",
|
||||
summary: { total: SERVICE_NAMES.length, succeeded, failed: SERVICE_NAMES.length - succeeded, running, unavailable: SERVICE_NAMES.length - running },
|
||||
services: Object.fromEntries(results.map((result) => [result.service, result])),
|
||||
nextCommand: action === "start" || action === "restart" ? "hwlab-cli caserun service status" : undefined,
|
||||
valuesPrinted: false,
|
||||
};
|
||||
}
|
||||
|
||||
export function caserunNativeServiceOrder(action: string): readonly ServiceName[] {
|
||||
const parsed = validServiceAction(action);
|
||||
return parsed === "stop" ? SERVICE_STOP_ORDER : parsed === "start" || parsed === "restart" ? SERVICE_START_ORDER : SERVICE_NAMES;
|
||||
}
|
||||
|
||||
async function runServiceCommandsInOrder(order: readonly ServiceName[], action: ServiceAction, input: { cwd: string; env: Record<string, string | undefined> }) {
|
||||
const results = [];
|
||||
for (const service of order) results.push(await safeServiceCommand(service, action, input));
|
||||
return results;
|
||||
}
|
||||
|
||||
async function safeServiceCommand(service: ServiceName, action: ServiceAction, input: { cwd: string; env: Record<string, string | undefined> }) {
|
||||
try {
|
||||
return await caserunNativeServiceCommand({ service, action, cwd: input.cwd, env: input.env });
|
||||
} catch (error: any) {
|
||||
return { ok: false, operation: `caserun.service.${service}.${action}`, service, status: "failed", error: { code: String(error?.code ?? "service_action_failed"), message: error instanceof Error ? error.message : String(error) } };
|
||||
}
|
||||
}
|
||||
|
||||
function boundedAggregateLogs(result: any) {
|
||||
if (!Array.isArray(result.lines)) return result;
|
||||
const selected = result.lines.slice(-8);
|
||||
const lines = selected.map((line: unknown) => String(line).slice(0, 300));
|
||||
return { ...result, lines, truncated: result.truncated === true || result.lines.length > lines.length || selected.some((line: unknown) => String(line).length > 300), aggregateTail: { maxLines: 8, maxLineBytes: 300 } };
|
||||
}
|
||||
|
||||
export async function caserunNativeServiceCommand(input: { service: string; action: string; cwd: string; env: Record<string, string | undefined> }) {
|
||||
const service = validService(input.service);
|
||||
@@ -30,23 +89,24 @@ async function start(service: ServiceName, stateDir: string, stateFile: string,
|
||||
await mkdir(stateDir, { recursive: true });
|
||||
const endpoints = serviceEndpoints(service, env);
|
||||
await releaseStaleServicePort(service, env);
|
||||
const command = serviceCommand(service);
|
||||
const command = serviceCommand(service, env);
|
||||
const logFd = openSync(logFile, "w");
|
||||
const serviceCwd = path.join(cwd, "web/hwlab-cloud-web");
|
||||
const serviceCwd = service === "web" ? path.join(cwd, "web/hwlab-cloud-web") : cwd;
|
||||
const child = spawn(command[0], command.slice(1), { cwd: serviceCwd, env: serviceEnv(service, env), detached: true, stdio: ["ignore", logFd, logFd] });
|
||||
closeSync(logFd);
|
||||
child.unref();
|
||||
const state = { service, pid: child.pid, status: "starting", command, cwd, logFile, endpoints, startedAt: new Date().toISOString() };
|
||||
const state = { service, pid: child.pid, status: "starting", command, cwd: serviceCwd, logFile, endpoints, healthUrl: endpoints.probe.url, startedAt: new Date().toISOString() };
|
||||
await writeFile(stateFile, `${JSON.stringify(state, null, 2)}\n`, "utf8");
|
||||
return { ok: true, operation: `caserun.service.${service}.start`, ...state, nextCommand: `hwlab-cli caserun service ${service} status` };
|
||||
}
|
||||
|
||||
async function releaseStaleServicePort(service: ServiceName, env: Record<string, string | undefined>) {
|
||||
const bindHost = requiredEnv(env, service === "api" ? "CASERUN_API_BIND_HOST" : "CASERUN_WEB_BIND_HOST");
|
||||
const port = Number(requiredEnv(env, service === "api" ? "CASERUN_API_PORT" : "CASERUN_WEB_PORT"));
|
||||
const prefix = service === "api" ? "CASERUN_API" : service === "worker" ? "CASERUN_WORKER" : "CASERUN_WEB";
|
||||
const bindHost = requiredEnv(env, `${prefix}_BIND_HOST`);
|
||||
const port = Number(requiredEnv(env, service === "worker" ? "CASERUN_WORKER_HEALTH_PORT" : `${prefix}_PORT`));
|
||||
if (await portAvailable(bindHost, port)) return;
|
||||
const expectedPortKey = service === "api" ? "HWLAB_CASERUN_NATIVE_PORT" : "WORKBENCH_WEB_PORT";
|
||||
const commandMarker = service === "api" ? "caserun-native-supervisor.ts" : "dev:native-caserun";
|
||||
const expectedPortKey = service === "api" ? "HARNESSRL_API_PORT" : service === "worker" ? "HARNESSRL_WORKER_HEALTH_PORT" : "WORKBENCH_WEB_PORT";
|
||||
const commandMarker = service === "api" ? "hwlab-harnessrl-api/main.ts" : service === "worker" ? "hwlab-harnessrl-worker/main.ts" : "dev:native-caserun";
|
||||
const candidates = await matchingServiceGroups(commandMarker, expectedPortKey, String(port));
|
||||
if (candidates.length === 0) {
|
||||
throw codedError("native_port_owned_by_other_service", `CaseRun ${service} port ${port} is occupied by another service`);
|
||||
@@ -116,8 +176,10 @@ async function stop(service: ServiceName, stateFile: string, logFile: string) {
|
||||
|
||||
async function status(service: ServiceName, stateFile: string, logFile: string) {
|
||||
const state = await readState(stateFile);
|
||||
const running = Boolean(state?.pid && alive(state.pid));
|
||||
return { ok: running, operation: `caserun.service.${service}.status`, service, status: running ? "running" : "stopped", pid: running ? state.pid : null, endpoints: state?.endpoints ?? null, stateFile, logFile, nextCommand: `hwlab-cli caserun service ${service} logs` };
|
||||
const processRunning = Boolean(state?.pid && alive(state.pid));
|
||||
const health = processRunning && state?.healthUrl ? await probeHealth(state.healthUrl) : { ok: false, status: null, error: processRunning ? "health-url-not-declared" : "process-not-running" };
|
||||
const running = processRunning && health.ok;
|
||||
return { ok: running, operation: `caserun.service.${service}.status`, service, status: running ? "running" : processRunning ? "degraded" : "stopped", pid: processRunning ? state.pid : null, endpoints: state?.endpoints ?? null, health, stateFile, logFile, nextCommand: `hwlab-cli caserun service ${service} logs` };
|
||||
}
|
||||
|
||||
async function logs(service: ServiceName, stateFile: string, logFile: string) {
|
||||
@@ -126,28 +188,37 @@ async function logs(service: ServiceName, stateFile: string, logFile: string) {
|
||||
return { ...state, operation: `caserun.service.${service}.logs`, lines, truncated: lines.length === 80 };
|
||||
}
|
||||
|
||||
function serviceCommand(service: ServiceName) {
|
||||
return service === "api"
|
||||
? ["bun", "scripts/caserun-native-supervisor.ts"]
|
||||
: ["bun", "run", "dev:native-caserun"];
|
||||
function serviceCommand(service: ServiceName, env: Record<string, string | undefined>) {
|
||||
const watch = env.CASERUN_BACKEND_WATCH === "1" ? ["--watch"] : [];
|
||||
if (service === "api") return ["bun", ...watch, "cmd/hwlab-harnessrl-api/main.ts"];
|
||||
if (service === "worker") return ["bun", ...watch, "cmd/hwlab-harnessrl-worker/main.ts"];
|
||||
return ["bun", "run", "dev:native-caserun"];
|
||||
}
|
||||
function serviceEnv(service: ServiceName, env: Record<string, string | undefined>) {
|
||||
return { ...process.env, ...env, HWLAB_CASERUN_NATIVE_TEST: "1", ...(service === "web"
|
||||
? {
|
||||
WORKBENCH_WEB_HOST: requiredEnv(env, "CASERUN_WEB_BIND_HOST"),
|
||||
WORKBENCH_WEB_PORT: requiredEnv(env, "CASERUN_WEB_PORT"),
|
||||
HWLAB_CASERUN_NATIVE_URL: `http://${requiredEnv(env, "CASERUN_PROBE_HOST")}:${requiredEnv(env, "CASERUN_API_PORT")}`
|
||||
}
|
||||
: { HWLAB_CASERUN_NATIVE_HOST: requiredEnv(env, "CASERUN_API_BIND_HOST"), HWLAB_CASERUN_NATIVE_PORT: requiredEnv(env, "CASERUN_API_PORT") }) };
|
||||
if (service === "api") return { ...process.env, ...env, HWLAB_CASERUN_NATIVE_TEST: "1", HARNESSRL_API_HOST: requiredEnv(env, "CASERUN_API_BIND_HOST"), HARNESSRL_API_PORT: requiredEnv(env, "CASERUN_API_PORT") };
|
||||
if (service === "worker") return { ...process.env, ...env, HWLAB_CASERUN_NATIVE_TEST: "1", HARNESSRL_WORKER_HEALTH_HOST: requiredEnv(env, "CASERUN_WORKER_BIND_HOST"), HARNESSRL_WORKER_HEALTH_PORT: requiredEnv(env, "CASERUN_WORKER_HEALTH_PORT") };
|
||||
return {
|
||||
...process.env,
|
||||
...env,
|
||||
HWLAB_CASERUN_NATIVE_TEST: "1",
|
||||
WORKBENCH_WEB_HOST: requiredEnv(env, "CASERUN_WEB_BIND_HOST"),
|
||||
WORKBENCH_WEB_PORT: requiredEnv(env, "CASERUN_WEB_PORT"),
|
||||
HWLAB_CASERUN_NATIVE_URL: `http://${requiredEnv(env, "CASERUN_PROBE_HOST")}:${requiredEnv(env, "CASERUN_API_PORT")}`,
|
||||
};
|
||||
}
|
||||
function serviceEndpoints(service: ServiceName, env: Record<string, string | undefined>) {
|
||||
const port = service === "api" ? requiredEnv(env, "CASERUN_API_PORT") : requiredEnv(env, "CASERUN_WEB_PORT");
|
||||
const prefix = service === "api" ? "CASERUN_API" : service === "worker" ? "CASERUN_WORKER" : "CASERUN_WEB";
|
||||
const bindHost = requiredEnv(env, `${prefix}_BIND_HOST`);
|
||||
const probeHost = service === "api" ? requiredEnv(env, "CASERUN_PROBE_HOST") : requiredEnv(env, `${prefix}_PROBE_HOST`);
|
||||
const port = requiredEnv(env, service === "worker" ? "CASERUN_WORKER_HEALTH_PORT" : `${prefix}_PORT`);
|
||||
const publicBaseUrl = fixedHttpsOrigin(requiredEnv(env, "CASERUN_PUBLIC_BASE_URL"), "CASERUN_PUBLIC_BASE_URL");
|
||||
validPort(port);
|
||||
return { url: service === "api" ? publicBaseUrl : `${publicBaseUrl}/caserun` };
|
||||
const pathname = service === "web" ? "/caserun" : "/health/ready";
|
||||
return { bind: { host: bindHost, port: validPort(port) }, probe: { host: probeHost, port: validPort(port), url: `http://${probeHost}:${port}${pathname}` }, public: service === "worker" ? null : { url: service === "api" ? publicBaseUrl : `${publicBaseUrl}/caserun` } };
|
||||
}
|
||||
function validPort(value: string) { const port = Number(value); if (!Number.isInteger(port) || port < 1 || port > 65535) throw codedError("invalid_service_port", `invalid CaseRun service port: ${value}`); return port; }
|
||||
function validService(value: string): ServiceName { if (value === "api" || value === "web") return value; throw codedError("invalid_service", "CaseRun service must be api or web"); }
|
||||
function validService(value: string): ServiceName { if (value === "api" || value === "worker" || value === "web") return value; throw codedError("invalid_service", "CaseRun service must be api, worker, or web"); }
|
||||
function validServiceAction(value: string): ServiceAction { if (value === "start" || value === "stop" || value === "restart" || value === "status" || value === "logs") return value; throw codedError("unsupported_service_action", `unsupported CaseRun service action: ${value}`); }
|
||||
async function readState(file: string) { return JSON.parse(await readFile(file, "utf8").catch(() => "null")); }
|
||||
function alive(pid: number) { try { process.kill(pid, 0); return true; } catch { return false; } }
|
||||
async function waitForExit(pid: number, timeoutMs: number) {
|
||||
@@ -165,4 +236,8 @@ function fixedHttpsOrigin(value: string, key: string) {
|
||||
}
|
||||
throw codedError("native_exposure_config_invalid", `${key} must be a pathless HTTPS origin`);
|
||||
}
|
||||
async function probeHealth(url: string) {
|
||||
try { const response = await fetch(url, { signal: AbortSignal.timeout(1_000) }); return { ok: response.ok, status: response.status, url }; }
|
||||
catch (error) { return { ok: false, status: null, url, error: error instanceof Error ? error.message : String(error) }; }
|
||||
}
|
||||
function codedError(code: string, message: string) { return Object.assign(new Error(message), { code }); }
|
||||
|
||||
Reference in New Issue
Block a user