feat: wire device pod gateway dispatch

This commit is contained in:
Codex
2026-05-29 10:44:59 +08:00
parent dfc1b42fe3
commit 4cdc80e672
12 changed files with 543 additions and 18 deletions
+108 -3
View File
@@ -63,7 +63,7 @@ test("device pod executor exposes cloud-api authority boundary and method guards
headers: { "x-hwlab-internal-service": "hwlab-cloud-api" }
});
assert.equal(storedOutput.job.id, "job_devicepod_test");
assert.equal(storedOutput.output.summary, "gateway/device-host-cli dispatch is not connected");
assert.equal(storedOutput.output.summary, "HWLAB_CLOUD_API_INTERNAL_URL is not configured for gateway dispatch");
assert.equal(storedOutput.truncation.truncated, false);
const externalJobRead = await fetch(`http://127.0.0.1:${service.port}/v1/device-pods/device-pod-test/jobs/job_devicepod_test`);
@@ -89,7 +89,91 @@ test("device pod executor exposes cloud-api authority boundary and method guards
}
});
async function startDevicePod(devicePodId) {
test("device pod executor dispatches internal jobs through cloud-api gateway adapter", async () => {
const dispatches = [];
const cloudApi = createServer(async (request, response) => {
const body = await requestJson(request);
dispatches.push({ method: request.method, url: request.url, internalService: request.headers["x-hwlab-internal-service"], body });
response.writeHead(200, { "content-type": "application/json; charset=utf-8" });
response.end(JSON.stringify({
jsonrpc: "2.0",
id: body.id,
result: {
accepted: true,
status: "completed",
operationId: body.params.operationId,
traceId: body.params.traceId,
gatewaySessionId: body.params.gatewaySessionId,
resourceId: body.params.resourceId,
capabilityId: body.params.capabilityId,
dispatch: {
shellExecuted: true,
dispatchStatus: "succeeded",
stdout: "device-host-cli ok",
exitCode: 0
},
auditId: "aud_devicepod_test",
evidenceId: "evd_devicepod_test",
gateway: { gatewayId: "gtw_devicepod_test" }
}
}));
});
await new Promise((resolve) => cloudApi.listen(0, "127.0.0.1", resolve));
const cloudPort = cloudApi.address().port;
const service = await startDevicePod("device-pod-test", {
HWLAB_CLOUD_API_INTERNAL_URL: `http://127.0.0.1:${cloudPort}`
});
try {
const jobResponse = await fetch(`http://127.0.0.1:${service.port}/v1/device-pods/device-pod-test/jobs`, {
method: "POST",
headers: { "content-type": "application/json", "x-hwlab-internal-service": "hwlab-cloud-api" },
body: JSON.stringify({
jobId: "job_devicepod_gateway_test",
intent: "workspace.ls",
args: { path: "src" },
traceId: "trc_device_pod_gateway_test",
operationId: "op_device_pod_gateway_test",
targetId: "target-device-pod-test",
profileHash: "sha256:test",
ownerUserId: "usr_device_pod_test",
profile: {
schemaVersion: 1,
devicePodId: "device-pod-test",
target: { id: "target-device-pod-test" },
projectWorkspace: { workspaceRoot: "F:\\Work\\Project" },
route: {
gatewaySessionId: "gws_devicepod_test",
resourceId: "res_devicepod_test",
capabilityId: "cap_device_host_cli",
hostWorkspaceRoot: "F:\\Work\\Project",
hostCli: "node tools\\device-host-cli.mjs"
}
}
})
});
assert.equal(jobResponse.status, 202);
const job = await jobResponse.json();
assert.equal(job.status, "running");
assert.equal(job.targetId, "target-device-pod-test");
assert.equal(job.profileHash, "sha256:test");
const output = await waitForJobStatus(service.port, "device-pod-test", "job_devicepod_gateway_test", "completed");
assert.equal(output.status, "completed");
assert.equal(output.output.text, "device-host-cli ok");
assert.equal(output.output.auditId, "aud_devicepod_test");
assert.equal(dispatches.length, 1);
assert.equal(dispatches[0].internalService, "hwlab-device-pod");
assert.equal(dispatches[0].body.params.gatewaySessionId, "gws_devicepod_test");
assert.equal(dispatches[0].body.params.input.cwd, "F:\\Work\\Project");
assert.match(dispatches[0].body.params.input.command, /^node -e /u);
} finally {
await service.stop();
await new Promise((resolve, reject) => cloudApi.close((error) => (error ? reject(error) : resolve())));
}
});
async function startDevicePod(devicePodId, extraEnv = {}) {
const port = await freePort();
const child = spawn(bunCommand, ["run", "cmd/hwlab-device-pod/main.ts"], {
cwd: process.cwd(),
@@ -97,7 +181,8 @@ async function startDevicePod(devicePodId) {
...process.env,
HWLAB_DEVICE_POD_PORT: String(port),
HWLAB_DEVICE_POD_ID: devicePodId,
HWLAB_ENVIRONMENT: "v02"
HWLAB_ENVIRONMENT: "v02",
...extraEnv
},
stdio: ["ignore", "pipe", "pipe"]
});
@@ -114,6 +199,26 @@ async function fetchJson(url, options = {}) {
return response.json();
}
async function waitForJobStatus(port, devicePodId, jobId, status) {
let last;
for (let attempt = 0; attempt < 30; attempt += 1) {
last = await fetchJson(`http://127.0.0.1:${port}/v1/device-pods/${devicePodId}/jobs/${jobId}/output`, {
headers: { "x-hwlab-internal-service": "hwlab-cloud-api" }
});
if (last.status === status) return last;
await new Promise((resolve) => setTimeout(resolve, 50));
}
assert.equal(last?.status, status);
return last;
}
async function requestJson(request) {
const chunks = [];
for await (const chunk of request) chunks.push(chunk);
const text = Buffer.concat(chunks).toString("utf8").trim();
return text ? JSON.parse(text) : {};
}
async function freePort() {
const server = createServer();
await new Promise((resolve) => server.listen(0, "127.0.0.1", resolve));
+195 -7
View File
@@ -10,6 +10,8 @@ const DEFAULT_DEVICE_POD_ID = "device-pod-71-freq";
const port = parsePort(process.env.HWLAB_DEVICE_POD_PORT, parsePort(process.env.PORT, 7601));
const devicePodId = process.env.HWLAB_DEVICE_POD_ID || DEFAULT_DEVICE_POD_ID;
const environment = process.env.HWLAB_ENVIRONMENT || process.env.HWLAB_GITOPS_PROFILE || "dev";
const cloudApiInternalUrl = normalizeBaseUrl(process.env.HWLAB_CLOUD_API_INTERNAL_URL || process.env.HWLAB_CLOUD_API_URL);
const dispatchTimeoutMs = numberOr(process.env.HWLAB_DEVICE_POD_GATEWAY_DISPATCH_TIMEOUT_MS, 120000);
const jobs = new Map();
listen(createServer(async (request, response) => {
@@ -71,6 +73,7 @@ function healthPayload() {
authority: "hwlab-cloud-api",
acceptsUserAuthority: false,
fake: false,
gatewayAdapter: gatewayAdapterState(),
source: sourcePayload(),
note: "Profile, grant, lease, and user-facing job authority live in hwlab-cloud-api; this service only exposes the internal executor boundary."
};
@@ -88,6 +91,7 @@ function listPayload() {
status: "executor-ready",
authority: "hwlab-cloud-api",
fake: false,
gatewayAdapter: gatewayAdapterState(),
routes: {
cloudAuthority: "/v1/device-pods",
internalJob: `/v1/device-pods/${encodeURIComponent(devicePodId)}/jobs`
@@ -152,23 +156,38 @@ async function handleInternalJob(request, response, id) {
const traceId = typeof body.traceId === "string" && body.traceId.startsWith("trc_") ? body.traceId : `trc_devicepod_${randomUUID()}`;
const operationId = typeof body.operationId === "string" ? body.operationId : `op_devicepod_${randomUUID()}`;
const now = new Date().toISOString();
const job = {
id: typeof body.jobId === "string" && body.jobId.startsWith("job_") ? body.jobId : `job_devicepod_${randomUUID()}`,
const profile = normalizeObject(body.profile);
const route = normalizeObject(profile.route);
const jobId = typeof body.jobId === "string" && body.jobId.startsWith("job_") ? body.jobId : `job_devicepod_${randomUUID()}`;
const command = deviceHostCommand(profile, {
id: jobId,
devicePodId: id,
status: "blocked",
intent: typeof body.intent === "string" ? body.intent : "unknown",
args: normalizeObject(body.args)
});
const dispatchBlocker = gatewayDispatchPreflightBlocker(route, command);
const job = {
id: jobId,
devicePodId: id,
status: dispatchBlocker ? "blocked" : "running",
intent: typeof body.intent === "string" ? body.intent : "unknown",
reason: typeof body.reason === "string" ? body.reason : "",
traceId,
operationId,
createdAt: now,
updatedAt: now,
completedAt: now,
output: boundedOutput({ text: "", summary: "gateway/device-host-cli dispatch is not connected" }),
blocker: gatewayAdapterBlocker("gateway/device-host-cli adapter is not connected to this standalone executor service"),
completedAt: dispatchBlocker ? now : null,
targetId: typeof body.targetId === "string" ? body.targetId : targetIdFromProfile(profile),
profileHash: typeof body.profileHash === "string" ? body.profileHash : "",
ownerUserId: typeof body.ownerUserId === "string" ? body.ownerUserId : "",
command,
output: dispatchBlocker ? boundedOutput({ text: "", summary: dispatchBlocker.summary }) : boundedOutput({ text: "", summary: "gateway/device-host-cli dispatch queued" }),
blocker: dispatchBlocker,
source: sourcePayload()
};
jobs.set(jobKey(id, job.id), job);
jsonResponse(response, 409, jobPayload(job, { accepted: false }));
if (!dispatchBlocker) dispatchGatewayJob({ job, route });
jsonResponse(response, dispatchBlocker ? 409 : 202, jobPayload(job, { accepted: !dispatchBlocker }));
}
function handleGetInternalJob(request, response, id, subpath) {
@@ -204,12 +223,69 @@ function handleCancelInternalJob(request, response, id, subpath) {
jsonResponse(response, 200, jobPayload(canceled));
}
async function dispatchGatewayJob({ job, route }) {
const target = `${cloudApiInternalUrl}/v1/internal/device-pod/gateway-dispatch`;
const controller = new AbortController();
const timer = setTimeout(() => controller.abort(), dispatchTimeoutMs + 10000);
try {
const response = await fetch(target, {
method: "POST",
signal: controller.signal,
headers: {
accept: "application/json",
"content-type": "application/json",
"x-hwlab-internal-service": SERVICE_ID,
"x-trace-id": job.traceId,
"x-request-id": `req_${job.id}`
},
body: JSON.stringify({
id: `req_${job.id}`,
actorId: job.ownerUserId || "svc_hwlab-device-pod",
params: {
projectId: "prj_v02_device_pod",
gatewaySessionId: route.gatewaySessionId,
resourceId: route.resourceId ?? "res_device_pod",
capabilityId: route.capabilityId ?? "cap_device_host_cli",
operationId: job.operationId,
traceId: job.traceId,
input: {
command: job.command,
cwd: route.hostWorkspaceRoot,
timeoutMs: dispatchTimeoutMs
}
}
})
});
const payload = await response.json().catch(() => ({}));
const current = jobs.get(jobKey(job.devicePodId, job.id));
if (current && terminalJobStatus(current.status)) return;
const next = jobFromGatewayDispatch(job, payload, response.status);
jobs.set(jobKey(job.devicePodId, job.id), next);
} catch (error) {
const current = jobs.get(jobKey(job.devicePodId, job.id));
if (current && terminalJobStatus(current.status)) return;
const now = new Date().toISOString();
jobs.set(jobKey(job.devicePodId, job.id), {
...job,
status: "failed",
updatedAt: now,
completedAt: now,
output: boundedOutput({ text: "", error: error?.message ?? "gateway dispatch failed" }),
blocker: gatewayDispatchFailedBlocker(error?.message ?? "gateway dispatch failed")
});
} finally {
clearTimeout(timer);
}
}
function jobPayload(job, { accepted = !["blocked", "failed"].includes(job.status) } = {}) {
return {
accepted,
status: job.status,
contractVersion: CONTRACT_VERSION,
devicePodId: job.devicePodId,
targetId: job.targetId ?? null,
profileHash: job.profileHash ?? "",
traceId: job.traceId,
operationId: job.operationId,
job: publicJob(job),
@@ -253,6 +329,118 @@ function gatewayAdapterBlocker(summary) {
return { code: "gateway_dispatch_unavailable", layer: "device-pod", retryable: true, summary, userMessage: "Device Pod executor boundary is present, but gateway/device-host-cli dispatch is not connected here." };
}
function gatewayDispatchFailedBlocker(summary) {
return { code: "gateway_dispatch_failed", layer: "device-pod", retryable: true, summary, userMessage: "Device Pod gateway/device-host-cli dispatch failed." };
}
function gatewayDispatchPreflightBlocker(route, command) {
if (!cloudApiInternalUrl) return gatewayAdapterBlocker("HWLAB_CLOUD_API_INTERNAL_URL is not configured for gateway dispatch");
if (!route.gatewaySessionId) return gatewayAdapterBlocker("profile route.gatewaySessionId is missing");
if (!command) return gatewayAdapterBlocker("device-host-cli command mapping is not available for this intent");
return null;
}
function gatewayAdapterState() {
return { mode: cloudApiInternalUrl ? "cloud-api-internal-dispatch" : "blocked", cloudApiInternalUrlConfigured: Boolean(cloudApiInternalUrl) };
}
function jobFromGatewayDispatch(job, payload, httpStatus) {
const result = normalizeObject(payload.result);
const dispatch = normalizeObject(result.dispatch);
const blocker = payload.blocker ?? result.blocker ?? (payload.error ? gatewayDispatchFailedBlocker(payload.error.message ?? "gateway dispatch failed") : null);
const completed = !blocker && httpStatus < 400 && (result.status === "completed" || result.status === "succeeded" || dispatch.dispatchStatus === "succeeded" || dispatch.dispatchStatus === "completed");
const status = completed ? "completed" : blocker?.code === "gateway_dispatch_unavailable" ? "blocked" : "failed";
const now = new Date().toISOString();
return {
...job,
status,
updatedAt: now,
completedAt: now,
output: boundedOutput({
text: gatewayDispatchText(result, dispatch),
httpStatus,
dispatch,
gateway: result.gateway ?? null,
auditId: result.auditId ?? null,
evidenceId: result.evidenceId ?? null,
summary: completed ? "gateway/device-host-cli dispatch completed" : blocker?.summary ?? "gateway/device-host-cli dispatch failed"
}),
blocker: completed ? null : blocker ?? gatewayDispatchFailedBlocker(dispatch.message ?? dispatch.stderr ?? result.status ?? "gateway dispatch failed")
};
}
function gatewayDispatchText(result, dispatch) {
if (typeof dispatch.stdout === "string" && dispatch.stdout) return dispatch.stdout;
if (typeof dispatch.stderr === "string" && dispatch.stderr) return dispatch.stderr;
if (typeof result.text === "string") return result.text;
return JSON.stringify(result);
}
function deviceHostCommand(profile, job) {
const args = deviceHostArgs(job.intent, job.args);
if (!args) return null;
const payload = {
devicePodId: profile.devicePodId || job.devicePodId,
profile,
profilePath: `.device-pod/.runtime/${job.id}.json`,
hostCli: profile.route?.hostCli || "node tools/device-host-cli.mjs",
args
};
const runner = "const cp=require('node:child_process'),fs=require('node:fs'),path=require('node:path');const p=JSON.parse(Buffer.from(process.argv[1],'base64').toString('utf8'));fs.mkdirSync(path.dirname(p.profilePath),{recursive:true});fs.writeFileSync(p.profilePath,JSON.stringify(p.profile,null,2)+'\\n','utf8');function split(s){const out=[];let q=null,b='';for(const ch of String(s)){if(q){if(ch===q)q=null;else b+=ch}else if(ch.charCodeAt(0)===39||ch.charCodeAt(0)===34)q=ch;else if(/\\s/u.test(ch)){if(b){out.push(b);b=''}}else b+=ch}if(b)out.push(b);return out}const parts=split(p.hostCli||'node tools/device-host-cli.mjs');const child=cp.spawn(parts[0],parts.slice(1).concat(['--profile',p.profilePath,'--pod-id',p.devicePodId],p.args),{stdio:'inherit',shell:false});child.on('exit',(c)=>process.exit(c??0));child.on('error',(e)=>{console.error(e.message);process.exit(127)});";
const encoded = Buffer.from(JSON.stringify(payload), "utf8").toString("base64");
return `node -e "${runner}" "${encoded}"`;
}
function deviceHostArgs(intent, args = {}) {
if (intent === "workspace.ls") return ["workspace", "ls", textOr(args.path, ".")];
if (intent === "workspace.cat") {
const path = textOr(args.path, "");
return path ? ["workspace", "cat", path, "--limit", String(numberOr(args.limit ?? args.maxBytes, 40000))] : null;
}
if (intent === "workspace.rg") {
const pattern = textOr(args.pattern ?? args.query, "");
return pattern ? ["workspace", "rg", pattern, textOr(args.path, ".")] : null;
}
if (intent === "workspace.apply-patch") return ["workspace", "apply-patch", textOr(args.base, "."), "--patch-b64", patchBase64(args)];
if (intent === "workspace.build") return ["workspace", "build", textOr(args.action, "start")];
if (intent === "debug.status") return ["debug-probe", "status"];
if (intent === "debug.chip-id") return ["debug-probe", "chip-id"];
if (intent === "debug.download") return ["debug-probe", "download", textOr(args.action, "start")];
if (intent === "debug.reset") return ["debug-probe", "reset"];
if (intent === "io.ports") return ["io-probe", textOr(args.uartId, "uart/1"), "ports"];
if (intent === "io.uart.read") return ["io-probe", textOr(args.uartId, "uart/1"), "read", "--duration-ms", String(numberOr(args.durationMs ?? args["duration-ms"], 1000))];
if (intent === "io.uart.write") return ["io-probe", textOr(args.uartId, "uart/1"), "write", ...(args.hex ? ["--hex"] : []), textOr(args.message ?? args.text ?? args.data, "")];
return null;
}
function patchBase64(args = {}) {
if (typeof args.patchB64 === "string") return args.patchB64;
if (typeof args["patch-b64"] === "string") return args["patch-b64"];
return Buffer.from(String(args.patch ?? ""), "utf8").toString("base64");
}
function targetIdFromProfile(profile = {}) {
return profile.target?.id ?? profile.targetId ?? null;
}
function normalizeObject(value) {
return value && typeof value === "object" && !Array.isArray(value) ? value : {};
}
function textOr(value, fallback) {
return typeof value === "string" && value.trim() ? value : fallback;
}
function numberOr(value, fallback) {
const parsed = Number.parseInt(String(value ?? ""), 10);
return Number.isInteger(parsed) && parsed > 0 ? parsed : fallback;
}
function normalizeBaseUrl(value) {
const text = String(value ?? "").trim();
return text ? text.replace(/\/+$/u, "") : "";
}
function jobOutputPayload(job) {
return {
...jobPayload(job),
+8
View File
@@ -29,6 +29,14 @@ test("edge proxy reports local health and proxies live health to upstream", asyn
assert.equal(live.environment, "v02");
assert.equal(upstream.captured.method, "GET");
assert.equal(upstream.captured.url, "/health/live");
const internal = await fetch(`http://127.0.0.1:${proxy.port}/v1/internal/device-pod/gateway-dispatch`, {
method: "POST",
headers: { "x-hwlab-internal-service": "hwlab-device-pod" }
});
assert.equal(internal.status, 404);
assert.equal((await internal.json()).error.code, "not_found");
assert.notEqual(upstream.captured.url, "/v1/internal/device-pod/gateway-dispatch");
} finally {
await proxy.stop();
await upstream.stop();
+10
View File
@@ -77,6 +77,16 @@ const server = createServer((request, response) => {
return;
}
if (url.pathname === "/v1/internal" || url.pathname.startsWith("/v1/internal/")) {
sendJson(response, 404, {
error: {
code: "not_found",
message: "Internal HWLAB routes are not exposed through the public edge proxy."
}
});
return;
}
proxyHttpRequest({ request, response, upstream, timeoutMs: proxyTimeoutMs });
});
+1
View File
@@ -303,6 +303,7 @@
"replicas": 1,
"env": {
"HWLAB_ENVIRONMENT": "dev",
"HWLAB_CLOUD_API_INTERNAL_URL": "http://hwlab-cloud-api.hwlab-dev.svc.cluster.local:6667",
"HWLAB_DEVICE_POD_ID": "device-pod-71-freq",
"HWLAB_DEVICE_POD_PORT": "7601"
}
+1 -1
View File
@@ -269,6 +269,6 @@ manages: many devicePodId
| 逻辑 device-pod 模型 | 已实现为规格 | 四要素、profile shape 和 Kubernetes 口径已定义。 |
| profile server authority | 部分实现 | cloud-api 保存正式 DB profile 并向用户返回脱敏摘要;device-pod executor 不接受用户上传 profile。 |
| 用户 grant + lease | 部分实现 | cloud-api 已实现 admin grant、可见性过滤、lease acquire/current/release、强副作用 job lease 校验和撤销授权释放 lease。 |
| REST/job API | 部分实现 | cloud-api 已实现 list/status/events/probe/job/output/cancel 和 lease API,并可把已授权 job 转发给内部 `hwlab-device-pod` executorexecutor 已实现内部 job create/get/output/cancel lifecycle,无在线 gateway/device-host-cli 时返回 blocker。 |
| REST/job API | 部分实现 | cloud-api 已实现 list/status/events/probe/job/output/cancel 和 lease API,并可把已授权 job 转发给内部 `hwlab-device-pod` executorexecutor 已实现内部 job create/get/output/cancel lifecycle 和 gateway/device-host-cli dispatch adapter,无在线 gateway/device-host-cli 时返回 blocker。 |
| 禁止 fake 作为 DEV-LIVE | 已实现/持续约束 | 规格和服务 payload 要求显式标记 fake/source。 |
+2 -1
View File
@@ -34,6 +34,7 @@
| `GET /v1/m3/status``POST /v1/m3/io` | M3 只读/受控 IO 入口;写操作必须有明确 approval。 |
| `GET /v1/diagnostics/gate``GET /v1/live-builds` | 诊断和 live build inventory。 |
| `GET /v1/gateway/sessions``POST /v1/gateway/poll``POST /v1/gateway/result` | gateway 主动出站注册、取任务和回传结果。 |
| `POST /v1/internal/device-pod/gateway-dispatch` | 仅接受 `hwlab-device-pod` 内部服务凭据,用于把 executor job dispatch 到 gateway poll/result;普通用户和 Code Agent 不可调用。 |
| `POST /v1/agent/chat``GET /v1/agent/chat/result/{traceId}``GET /v1/agent/chat/trace/{traceId}``POST /v1/agent/chat/cancel` | Code Agent 短连接提交、轮询、trace 和取消。 |
用户、权限、device-pod 管理 API 的最终规格见 [spec-user-access.md](spec-user-access.md) 和 [spec-device-pod.md](spec-device-pod.md)。
@@ -60,6 +61,6 @@
| Code Agent 短连接 submit/result/trace/cancel | 已实现 | repo-owned Codex stdio 和 provider profile 已接入。 |
| Postgres durable runtime | 已实现 | 通过 v02 独立 DB SecretRef 和 migration ledger 判定。 |
| gateway outbound poll/result | 已实现 | 支持 gateway 主动轮询和 `hardware.invoke.shell` 分发。 |
| device-pod 正式权限/profile/job | 部分实现 | profile/grant/list/status/job 持久化在 cloud-api;无在线 gateway/device-host-cli 时返回 blocker。 |
| device-pod 正式权限/profile/job | 部分实现 | profile/grant/list/status/job 持久化在 cloud-api已提供内部 gateway dispatch route 供 `hwlab-device-pod` executor 下发到 device-host-cli无在线 gateway/device-host-cli 时返回 blocker。 |
| v0.2 admin/user 权限模型 | 部分实现 | `/auth/*`、admin user/device-pod/grant API 和 Code Agent owner binding 已接入;生产 bootstrap 依赖 SecretRef。 |
@@ -14,6 +14,7 @@
- `cmd/hwlab-device-pod/main.ts` 返回 `contractVersion=device-pod-executor-v1`,并声明 `authority=hwlab-cloud-api``fake=false`
- `internal/device-pod/fake-data.mjs` 只保留为 legacy 前端/smoke fixture,不是 `hwlab-device-pod` 服务运行契约。
- `HWLAB_DEVICE_POD_ID` 可指定默认 executor devicePodId;正式 profile 仍由 `hwlab-cloud-api``device_pods.profile_json` 管理。
- `HWLAB_CLOUD_API_INTERNAL_URL` 配置后,executor 可用内部服务凭据回调 `hwlab-cloud-api` 的 gateway dispatch route,把已授权 job 下发到 gateway/device-host-cli;未配置、profile route 缺失或 gateway 不在线时返回 blocker。
## API 接口说明
@@ -23,7 +24,7 @@
| `GET /v1/device-pods` | 返回 executor boundary 摘要,不表达用户可见 device pod authority。 |
| `GET /v1/device-pods/{devicePodId}/status` | 返回 blocked executor status;真实 profile/status 以 cloud-api `/v1/device-pods/{devicePodId}/status` 为准。 |
| `GET /v1/device-pods/{devicePodId}/events` | 返回 bounded executor boundary event,不伪造硬件事件。 |
| `POST /v1/device-pods/{devicePodId}/jobs` | 只接受 `hwlab-cloud-api` 内部调用;创建内部 executor job 并返回 job、freshness、output/cancel URL当前无 gateway/device-host-cli 连接时返回 `gateway_dispatch_unavailable`。 |
| `POST /v1/device-pods/{devicePodId}/jobs` | 只接受 `hwlab-cloud-api` 内部调用;创建内部 executor job 并返回 job、freshness、output/cancel URL有可用 profile route 和 `HWLAB_CLOUD_API_INTERNAL_URL` 时通过 cloud-api gateway dispatch 下发到 device-host-cli,否则返回 `gateway_dispatch_unavailable`。 |
| `GET /v1/device-pods/{devicePodId}/jobs/{jobId}``GET /output``POST /cancel` | 只接受 `hwlab-cloud-api` 内部调用,用于查询内部 job、bounded output 和取消非终态 job;普通用户仍必须走 cloud-api 用户态 API。 |
用户态 `POST /jobs`、job output/cancel 和 admin profile/grant API 由 `hwlab-cloud-api` 实现,应以 [spec-device-pod.md](spec-device-pod.md) 为目标。
@@ -45,6 +46,6 @@
| health 和 executor boundary REST | 已实现 | 服务声明内部 executor、cloud-api authority 和非 fake 来源。 |
| bounded events/status | 已实现 | 只返回 executor boundary/blocker,不伪造硬件事件。 |
| 正式 profile authority | 已在 cloud-api 实现 | `hwlab-device-pod` 不读取或覆盖 `device_pods.profile_json`。 |
| job lifecycle | 部分实现 | 用户态 job 由 cloud-api 鉴权、授权、lease 校验和持久化;executor 侧已提供内部 job create/get/output/cancel lifecycle当前执行结果仍受 gateway/device-host-cli adapter 可用性限制。 |
| gateway/device-host-cli adapter | 部分实现 | cloud-api 已能向 gateway registry 分发;无在线 gateway 时返回 blocker。 |
| job lifecycle | 部分实现 | 用户态 job 由 cloud-api 鉴权、授权、lease 校验和持久化;executor 侧已提供内部 job create/get/output/cancel lifecycle并能把 job dispatch 结果写回 bounded output。 |
| gateway/device-host-cli adapter | 部分实现 | executor 已通过 cloud-api internal dispatch 接入 gateway poll/result 和 device-host-cli 命令映射;真实执行仍依赖 profile route、在线 gateway 和 host CLI。 |
+4 -1
View File
@@ -6,6 +6,7 @@
- 接收 FRP 转入的 API/health 请求,并代理到 `hwlab-cloud-api:6667`
- `/health/live` 必须透传 cloud-api live health,因此公网 API readiness 以 cloud-api 为准。
- `/v1/internal/*` 是集群内部服务路由,必须在 public edge 上 fail-closed,不透传到 cloud-api。
- 不实现业务 API,不读取 Secret,不替代 cloud-api 权限和 runtime readiness。
## 内部架构
@@ -21,6 +22,7 @@
| `GET /health/live` | 代理到 upstream `/health/live`,代表公网 API live readiness。 |
| `GET /health``GET /edge/health` | 返回 edge-proxy 自身 health 和 upstream 配置。 |
| `GET /status``GET /routes` | 返回代理路由表。 |
| `/v1/internal/*` | 公网 edge 返回 `404`,不代理内部服务路由。 |
| 其他路径 | 透明代理到 `hwlab-cloud-api`。 |
## 测试规格
@@ -37,8 +39,9 @@
| 规格项 | 状态 | 说明 |
| --- | --- | --- |
| API edge proxy | 已实现 | 其他路径透明代理 cloud-api。 |
| API edge proxy | 已实现 | 用户态路径透明代理 cloud-api。 |
| live health 透传 | 已实现 | `/health/live` 代理 upstream。 |
| internal route 阻断 | 已实现 | `/v1/internal/*` 不经 public edge 暴露。 |
| 长请求 timeout | 已实现 | 默认支持 Code Agent 长窗口。 |
| 业务权限/鉴权 | 不在本服务 | 由 cloud-api 承担。 |
+98 -2
View File
@@ -305,6 +305,101 @@ test("cloud api dispatches authorized device jobs to the internal device-pod exe
}
});
test("cloud api internal device-pod gateway dispatch is service-only and fail-closed without online gateway", async () => {
const server = createCloudApiServer({
env: { HWLAB_ACCESS_CONTROL_REQUIRED: "1" },
now: () => "2026-05-28T00:00:00.000Z"
});
await new Promise((resolve) => server.listen(0, "127.0.0.1", resolve));
try {
const { port } = server.address();
const external = await postJson(port, "/v1/internal/device-pod/gateway-dispatch", { params: {} });
assert.equal(external.status, 403);
assert.equal(external.body.error.code, "device_pod_internal_authority_required");
const blocked = await postJson(port, "/v1/internal/device-pod/gateway-dispatch", {
id: "req_devicepod_dispatch_test",
params: {
gatewaySessionId: "gws_missing",
resourceId: "res_devicepod_test",
capabilityId: "cap_device_host_cli",
operationId: "op_devicepod_dispatch_test",
traceId: "trc_devicepod_dispatch_test",
input: { command: "node tools/device-host-cli.mjs health" }
}
}, null, { "x-hwlab-internal-service": "hwlab-device-pod" });
assert.equal(blocked.status, 409);
assert.equal(blocked.body.blocker.code, "gateway_dispatch_unavailable");
} finally {
await new Promise((resolve, reject) => server.close((error) => (error ? reject(error) : resolve())));
}
});
test("cloud api internal device-pod gateway dispatch uses gateway poll result", async () => {
const server = createCloudApiServer({
env: { HWLAB_ACCESS_CONTROL_REQUIRED: "1" },
now: () => "2026-05-28T00:00:00.000Z"
});
await new Promise((resolve) => server.listen(0, "127.0.0.1", resolve));
try {
const { port } = server.address();
const poll = await postJson(port, "/v1/gateway/poll", {
serviceId: "hwlab-gateway",
gatewayId: "gtw_devicepod_test",
gatewaySessionId: "gws_devicepod_test",
resourceId: "res_devicepod_test",
capabilities: [{ capabilityId: "cap_device_host_cli", resourceId: "res_devicepod_test" }]
});
assert.equal(poll.status, 200);
const dispatchPromise = postJson(port, "/v1/internal/device-pod/gateway-dispatch", {
id: "req_devicepod_dispatch_test",
params: {
gatewaySessionId: "gws_devicepod_test",
resourceId: "res_devicepod_test",
capabilityId: "cap_device_host_cli",
operationId: "op_devicepod_dispatch_test",
traceId: "trc_devicepod_dispatch_test",
input: { command: "node tools/device-host-cli.mjs health", timeoutMs: 1000 }
}
}, null, { "x-hwlab-internal-service": "hwlab-device-pod" });
const queued = await postJson(port, "/v1/gateway/poll", { gatewayId: "gtw_devicepod_test", gatewaySessionId: "gws_devicepod_test" });
assert.equal(queued.status, 200);
assert.equal(queued.body.type, "request");
assert.equal(queued.body.request.method, "hardware.invoke.shell");
assert.equal(queued.body.request.params.input.command, "node tools/device-host-cli.mjs health");
const result = await postJson(port, "/v1/gateway/result", {
gatewayId: "gtw_devicepod_test",
gatewaySessionId: "gws_devicepod_test",
response: {
jsonrpc: "2.0",
id: queued.body.request.id,
result: {
accepted: true,
status: "succeeded",
shellExecuted: true,
dispatchStatus: "succeeded",
stdout: "host cli ok",
exitCode: 0,
gatewaySessionId: "gws_devicepod_test"
}
}
});
assert.equal(result.status, 200);
const dispatch = await dispatchPromise;
assert.equal(dispatch.status, 200);
assert.equal(dispatch.body.result.status, "completed");
assert.equal(dispatch.body.result.dispatch.stdout, "host cli ok");
} finally {
await new Promise((resolve, reject) => server.close((error) => (error ? reject(error) : resolve())));
}
});
test("cloud api exposes v1 access status routes and returns structured REST errors", async () => {
const server = createCloudApiServer({
env: { HWLAB_ACCESS_CONTROL_REQUIRED: "1" },
@@ -343,12 +438,13 @@ test("cloud api exposes v1 access status routes and returns structured REST erro
}
});
async function postJson(port, path, body, cookie = null) {
async function postJson(port, path, body, cookie = null, extraHeaders = {}) {
const response = await fetch(`http://127.0.0.1:${port}${path}`, {
method: "POST",
headers: {
"content-type": "application/json",
...(cookie ? { cookie } : {})
...(cookie ? { cookie } : {}),
...extraHeaders
},
body: JSON.stringify(body)
});
+108
View File
@@ -16,6 +16,7 @@ import {
import {
SUPPORTED_RPC_METHODS,
createErrorEnvelope,
createResultEnvelope,
handleJsonRpcRequest
} from "./json-rpc.ts";
import {
@@ -341,6 +342,11 @@ async function handleRestAdapter(request, response, url, options) {
return;
}
if (url.pathname === "/v1/internal/device-pod/gateway-dispatch") {
await handleInternalDevicePodGatewayDispatch(request, response, options);
return;
}
if (request.method === "POST" && url.pathname === "/v1/gateway/poll") {
await handleGatewayPollHttp(request, response, options);
return;
@@ -459,6 +465,91 @@ async function handleRestAdapter(request, response, url, options) {
sendJson(response, statusForRpcResponse(rpcResponse), rpcResponse);
}
async function handleInternalDevicePodGatewayDispatch(request, response, options) {
if (request.method !== "POST") {
sendJson(response, 405, { error: { code: "method_not_allowed", message: "Internal device-pod gateway dispatch only supports POST" } });
return;
}
if (internalServiceHeader(request) !== "hwlab-device-pod") {
sendJson(response, 403, { error: { code: "device_pod_internal_authority_required", message: "Only hwlab-device-pod may use the internal gateway dispatch route" } });
return;
}
const body = await readJsonObject(request, options.bodyLimitBytes);
if (!body.ok) {
sendJson(response, 400, body.error);
return;
}
const params = body.value?.params && typeof body.value.params === "object" && !Array.isArray(body.value.params) ? body.value.params : {};
const gatewaySessionId = typeof params.gatewaySessionId === "string" ? params.gatewaySessionId.trim() : "";
if (!gatewaySessionId || options.gatewayRegistry?.isOnline?.(gatewaySessionId) !== true) {
sendJson(response, 409, {
accepted: false,
status: "blocked",
blocker: {
code: "gateway_dispatch_unavailable",
layer: "device-pod",
retryable: true,
summary: gatewaySessionId ? "gateway session is not connected" : "profile route.gatewaySessionId is missing"
}
});
return;
}
const operationId = typeof params.operationId === "string" && params.operationId ? params.operationId : `op_devicepod_gateway_${randomUUID()}`;
const traceId = typeof params.traceId === "string" && params.traceId ? params.traceId : `trc_devicepod_gateway_${randomUUID()}`;
const rpcResponse = await handleJsonRpcRequest({
jsonrpc: "2.0",
id: typeof body.value.id === "string" && body.value.id ? body.value.id : getHeader(request, "x-request-id") || `req_devicepod_gateway_${randomUUID()}`,
method: "hardware.invoke.shell",
params: { ...params, gatewaySessionId, operationId, traceId },
meta: {
traceId,
actorId: typeof body.value.actorId === "string" ? body.value.actorId : "svc_hwlab-device-pod",
serviceId: "hwlab-device-pod",
environment: runtimeEnvironment(options.env ?? process.env)
}
}, {
adapter: "device-pod-internal",
runtimeStore: options.runtimeStore,
env: options.env,
dbProbe: options.dbProbe,
m3IoRequestJson: options.m3IoRequestJson,
gatewayRegistry: options.gatewayRegistry
});
if (rpcResponse.error) {
sendJson(response, statusForRpcResponse(rpcResponse), rpcResponse);
return;
}
const result = rpcResponse.result ?? {};
const dispatch = result.dispatch ?? {};
const completed = result.status === "succeeded" || result.status === "completed" || dispatch.dispatchStatus === "succeeded" || dispatch.dispatchStatus === "completed";
sendJson(response, completed ? 200 : 409, createResultEnvelope({
id: rpcResponse.id,
result: {
accepted: completed,
status: completed ? "completed" : "failed",
operationId: result.operationId ?? operationId,
traceId,
gatewaySessionId,
resourceId: result.resourceId ?? params.resourceId ?? null,
capabilityId: result.capabilityId ?? params.capabilityId ?? null,
dispatch,
auditId: result.auditId ?? null,
evidenceId: result.evidenceId ?? null,
gateway: result.gateway ?? null,
blocker: completed ? null : {
code: "gateway_dispatch_failed",
layer: "device-pod",
retryable: true,
summary: dispatch.message ?? dispatch.stderr ?? dispatch.status ?? result.status ?? "gateway dispatch failed"
}
},
requestMeta: { traceId, serviceId: "hwlab-device-pod" }
}));
}
async function codeAgentOptions(request, response, options) {
const auth = await options.accessController.authenticate(request, { required: options.accessController.required });
if (!auth.ok) {
@@ -468,6 +559,23 @@ async function codeAgentOptions(request, response, options) {
return { ...options, actor: auth.actor ?? null, authSession: auth.session ?? null };
}
async function readJsonObject(request, limitBytes) {
const body = await readBody(request, limitBytes);
try {
const value = body ? JSON.parse(body) : {};
if (!value || typeof value !== "object" || Array.isArray(value)) {
return { ok: false, error: { accepted: false, error: { code: "invalid_params", message: "body must be a JSON object" } } };
}
return { ok: true, value };
} catch (error) {
return { ok: false, error: { accepted: false, error: { code: "parse_error", message: "Invalid JSON body", reason: error.message } } };
}
}
function internalServiceHeader(request) {
return String(getHeader(request, "x-hwlab-internal-service") ?? "").trim();
}
async function handleGatewayPollHttp(request, response, options) {
const body = await readBody(request, options.bodyLimitBytes);
let params = {};
+4
View File
@@ -82,6 +82,10 @@ test("v02 render follows TypeScript runtime checks and does not self-patch boots
assert.match(workloads, /"name": "HWLAB_ACCESS_CONTROL_REQUIRED"[\s\S]{0,80}"value": "1"/u);
assert.match(workloads, /"name": "HWLAB_BOOTSTRAP_ADMIN_PASSWORD_HASH"/u);
assert.match(workloads, /"name": "hwlab-v02-bootstrap-admin"/u);
const devicePod = workloadsJson.items.find((item) => item.metadata?.name === "hwlab-device-pod");
assert.ok(devicePod, "expected v02 hwlab-device-pod workload");
const devicePodEnv = devicePod.spec.template.spec.containers[0].env;
assert.ok(devicePodEnv.some((entry) => entry.name === "HWLAB_CLOUD_API_INTERNAL_URL" && entry.value === "http://hwlab-cloud-api.hwlab-v02.svc.cluster.local:6667"));
const workloadTemplates = workloadsJson.items.filter((item) => item.spec?.template?.metadata?.labels?.["hwlab.pikastech.local/source-commit"]);
assert.ok(workloadTemplates.length > 0, "expected rendered pod-template source labels");
for (const item of workloadTemplates) {