diff --git a/cmd/hwlab-device-pod/main.test.ts b/cmd/hwlab-device-pod/main.test.ts index 346a61b8..62301705 100644 --- a/cmd/hwlab-device-pod/main.test.ts +++ b/cmd/hwlab-device-pod/main.test.ts @@ -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)); diff --git a/cmd/hwlab-device-pod/main.ts b/cmd/hwlab-device-pod/main.ts index 615e55c4..7910740e 100644 --- a/cmd/hwlab-device-pod/main.ts +++ b/cmd/hwlab-device-pod/main.ts @@ -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), diff --git a/cmd/hwlab-edge-proxy/main.test.ts b/cmd/hwlab-edge-proxy/main.test.ts index 1e1a95f8..945d7fa9 100644 --- a/cmd/hwlab-edge-proxy/main.test.ts +++ b/cmd/hwlab-edge-proxy/main.test.ts @@ -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(); diff --git a/cmd/hwlab-edge-proxy/main.ts b/cmd/hwlab-edge-proxy/main.ts index dc82757b..4afba871 100644 --- a/cmd/hwlab-edge-proxy/main.ts +++ b/cmd/hwlab-edge-proxy/main.ts @@ -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 }); }); diff --git a/deploy/deploy.json b/deploy/deploy.json index 2112718e..03f67e46 100644 --- a/deploy/deploy.json +++ b/deploy/deploy.json @@ -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" } diff --git a/docs/reference/spec-device-pod.md b/docs/reference/spec-device-pod.md index cc57aaac..4dbac0ef 100644 --- a/docs/reference/spec-device-pod.md +++ b/docs/reference/spec-device-pod.md @@ -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` executor;executor 已实现内部 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` executor;executor 已实现内部 job create/get/output/cancel lifecycle 和 gateway/device-host-cli dispatch adapter,无在线 gateway/device-host-cli 时返回 blocker。 | | 禁止 fake 作为 DEV-LIVE | 已实现/持续约束 | 规格和服务 payload 要求显式标记 fake/source。 | diff --git a/docs/reference/spec-v02-hwlab-cloud-api.md b/docs/reference/spec-v02-hwlab-cloud-api.md index cea9377d..748e23cf 100644 --- a/docs/reference/spec-v02-hwlab-cloud-api.md +++ b/docs/reference/spec-v02-hwlab-cloud-api.md @@ -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。 | diff --git a/docs/reference/spec-v02-hwlab-device-pod-service.md b/docs/reference/spec-v02-hwlab-device-pod-service.md index 30c20d86..54440cda 100644 --- a/docs/reference/spec-v02-hwlab-device-pod-service.md +++ b/docs/reference/spec-v02-hwlab-device-pod-service.md @@ -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。 | diff --git a/docs/reference/spec-v02-hwlab-edge-proxy.md b/docs/reference/spec-v02-hwlab-edge-proxy.md index ef0b85c3..584e36bf 100644 --- a/docs/reference/spec-v02-hwlab-edge-proxy.md +++ b/docs/reference/spec-v02-hwlab-edge-proxy.md @@ -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 承担。 | diff --git a/internal/cloud/access-control.test.ts b/internal/cloud/access-control.test.ts index b576bf77..efcf83e8 100644 --- a/internal/cloud/access-control.test.ts +++ b/internal/cloud/access-control.test.ts @@ -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) }); diff --git a/internal/cloud/server.ts b/internal/cloud/server.ts index b15f3203..15f1b4a3 100644 --- a/internal/cloud/server.ts +++ b/internal/cloud/server.ts @@ -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 = {}; diff --git a/scripts/g14-gitops-render.test.ts b/scripts/g14-gitops-render.test.ts index ddf4c47a..c48b8cc6 100644 --- a/scripts/g14-gitops-render.test.ts +++ b/scripts/g14-gitops-render.test.ts @@ -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) {