Merge pull request #551 from pikasTech/fix/v02-device-pod-executor-jobs-20260529
feat: route device pod jobs through executor
This commit is contained in:
@@ -45,10 +45,37 @@ test("device pod executor exposes cloud-api authority boundary and method guards
|
||||
const internalJob = 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({ intent: "workspace.ls", traceId: "trc_device_pod_test" })
|
||||
body: JSON.stringify({ jobId: "job_devicepod_test", intent: "workspace.ls", traceId: "trc_device_pod_test", operationId: "op_device_pod_test" })
|
||||
});
|
||||
assert.equal(internalJob.status, 409);
|
||||
assert.equal((await internalJob.json()).blocker.code, "gateway_dispatch_unavailable");
|
||||
const internalJobPayload = await internalJob.json();
|
||||
assert.equal(internalJobPayload.blocker.code, "gateway_dispatch_unavailable");
|
||||
assert.equal(internalJobPayload.job.id, "job_devicepod_test");
|
||||
assert.equal(internalJobPayload.outputUrl, "/v1/device-pods/device-pod-test/jobs/job_devicepod_test/output");
|
||||
|
||||
const storedJob = await fetchJson(`http://127.0.0.1:${service.port}/v1/device-pods/device-pod-test/jobs/job_devicepod_test`, {
|
||||
headers: { "x-hwlab-internal-service": "hwlab-cloud-api" }
|
||||
});
|
||||
assert.equal(storedJob.job.id, "job_devicepod_test");
|
||||
assert.equal(storedJob.status, "blocked");
|
||||
|
||||
const storedOutput = await fetchJson(`http://127.0.0.1:${service.port}/v1/device-pods/device-pod-test/jobs/job_devicepod_test/output`, {
|
||||
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.truncation.truncated, false);
|
||||
|
||||
const externalJobRead = await fetch(`http://127.0.0.1:${service.port}/v1/device-pods/device-pod-test/jobs/job_devicepod_test`);
|
||||
assert.equal(externalJobRead.status, 403);
|
||||
assert.equal((await externalJobRead.json()).error.code, "cloud_api_authority_required");
|
||||
|
||||
const cancelStored = await fetch(`http://127.0.0.1:${service.port}/v1/device-pods/device-pod-test/jobs/job_devicepod_test/cancel`, {
|
||||
method: "POST",
|
||||
headers: { "x-hwlab-internal-service": "hwlab-cloud-api" }
|
||||
});
|
||||
assert.equal(cancelStored.status, 200);
|
||||
assert.equal((await cancelStored.json()).job.id, "job_devicepod_test");
|
||||
|
||||
const missing = await fetch(`http://127.0.0.1:${service.port}/missing`);
|
||||
assert.equal(missing.status, 404);
|
||||
@@ -81,8 +108,8 @@ async function startDevicePod(devicePodId) {
|
||||
};
|
||||
}
|
||||
|
||||
async function fetchJson(url) {
|
||||
const response = await fetch(url);
|
||||
async function fetchJson(url, options = {}) {
|
||||
const response = await fetch(url, options);
|
||||
assert.equal(response.ok, true, `${url} returned ${response.status}`);
|
||||
return response.json();
|
||||
}
|
||||
|
||||
@@ -10,6 +10,7 @@ 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 jobs = new Map();
|
||||
|
||||
listen(createServer(async (request, response) => {
|
||||
try {
|
||||
@@ -43,6 +44,14 @@ listen(createServer(async (request, response) => {
|
||||
await handleInternalJob(request, response, route.devicePodId);
|
||||
return;
|
||||
}
|
||||
if (request.method === "GET" && route.subpath.startsWith("jobs/")) {
|
||||
handleGetInternalJob(request, response, route.devicePodId, route.subpath);
|
||||
return;
|
||||
}
|
||||
if (request.method === "POST" && route.subpath.startsWith("jobs/") && route.subpath.endsWith("/cancel")) {
|
||||
handleCancelInternalJob(request, response, route.devicePodId, route.subpath);
|
||||
return;
|
||||
}
|
||||
jsonResponse(response, 405, { error: "method_not_allowed", method: request.method ?? "UNKNOWN", path: url.pathname });
|
||||
} catch (error) {
|
||||
jsonResponse(response, 500, { error: "internal_error", message: error?.message ?? String(error), valuesRedacted: true });
|
||||
@@ -141,14 +150,85 @@ async function handleInternalJob(request, response, id) {
|
||||
}
|
||||
const body = await readJson(request);
|
||||
const traceId = typeof body.traceId === "string" && body.traceId.startsWith("trc_") ? body.traceId : `trc_devicepod_${randomUUID()}`;
|
||||
jsonResponse(response, 409, {
|
||||
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()}`,
|
||||
devicePodId: id,
|
||||
status: "blocked",
|
||||
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"),
|
||||
source: sourcePayload()
|
||||
};
|
||||
jobs.set(jobKey(id, job.id), job);
|
||||
jsonResponse(response, 409, jobPayload(job, { accepted: false }));
|
||||
}
|
||||
|
||||
function handleGetInternalJob(request, response, id, subpath) {
|
||||
if (!isInternalCaller(request)) return rejectExternalJobRoute(response);
|
||||
const parts = subpath.split("/");
|
||||
const job = jobs.get(jobKey(id, decodeURIComponent(parts[1] ?? "")));
|
||||
if (!job) {
|
||||
jsonResponse(response, 404, { error: { code: "device_job_not_found", message: "Device Pod executor job was not found" }, source: sourcePayload() });
|
||||
return;
|
||||
}
|
||||
if (parts[2] === "output") {
|
||||
jsonResponse(response, 200, jobOutputPayload(job));
|
||||
return;
|
||||
}
|
||||
jsonResponse(response, 200, jobPayload(job));
|
||||
}
|
||||
|
||||
function handleCancelInternalJob(request, response, id, subpath) {
|
||||
if (!isInternalCaller(request)) return rejectExternalJobRoute(response);
|
||||
const jobId = decodeURIComponent(subpath.split("/")[1] ?? "");
|
||||
const job = jobs.get(jobKey(id, jobId));
|
||||
if (!job) {
|
||||
jsonResponse(response, 404, { error: { code: "device_job_not_found", message: "Device Pod executor job was not found" }, source: sourcePayload() });
|
||||
return;
|
||||
}
|
||||
if (terminalJobStatus(job.status)) {
|
||||
jsonResponse(response, 200, jobPayload(job));
|
||||
return;
|
||||
}
|
||||
const now = new Date().toISOString();
|
||||
const canceled = { ...job, status: "canceled", updatedAt: now, completedAt: now, blocker: { code: "device_job_canceled", layer: "device-pod", retryable: false, summary: "Device Pod executor job was canceled by cloud-api" } };
|
||||
jobs.set(jobKey(id, jobId), canceled);
|
||||
jsonResponse(response, 200, jobPayload(canceled));
|
||||
}
|
||||
|
||||
function jobPayload(job, { accepted = !["blocked", "failed"].includes(job.status) } = {}) {
|
||||
return {
|
||||
accepted,
|
||||
status: job.status,
|
||||
contractVersion: CONTRACT_VERSION,
|
||||
devicePodId: job.devicePodId,
|
||||
traceId: job.traceId,
|
||||
operationId: job.operationId,
|
||||
job: publicJob(job),
|
||||
blocker: job.blocker,
|
||||
freshness: freshness(job.updatedAt, job.blocker),
|
||||
outputUrl: `/v1/device-pods/${encodeURIComponent(job.devicePodId)}/jobs/${encodeURIComponent(job.id)}/output`,
|
||||
cancelUrl: `/v1/device-pods/${encodeURIComponent(job.devicePodId)}/jobs/${encodeURIComponent(job.id)}/cancel`,
|
||||
source: job.source ?? sourcePayload()
|
||||
};
|
||||
}
|
||||
|
||||
function rejectExternalJobRoute(response) {
|
||||
jsonResponse(response, 403, {
|
||||
accepted: false,
|
||||
status: "blocked",
|
||||
contractVersion: CONTRACT_VERSION,
|
||||
devicePodId: id,
|
||||
traceId,
|
||||
operationId: typeof body.operationId === "string" ? body.operationId : `op_devicepod_${randomUUID()}`,
|
||||
blocker: gatewayAdapterBlocker("gateway/device-host-cli adapter is not connected to this standalone executor service"),
|
||||
error: {
|
||||
code: "cloud_api_authority_required",
|
||||
message: "Device Pod jobs must be authorized by hwlab-cloud-api before reaching the internal executor."
|
||||
},
|
||||
source: sourcePayload()
|
||||
});
|
||||
}
|
||||
@@ -172,3 +252,35 @@ function sourcePayload() {
|
||||
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 jobOutputPayload(job) {
|
||||
return {
|
||||
...jobPayload(job),
|
||||
output: job.output,
|
||||
text: job.output.text,
|
||||
bytes: job.output.bytes,
|
||||
truncation: job.output.truncation
|
||||
};
|
||||
}
|
||||
|
||||
function publicJob(job) {
|
||||
return { id: job.id, devicePodId: job.devicePodId, status: job.status, intent: job.intent, reason: job.reason, traceId: job.traceId, operationId: job.operationId, createdAt: job.createdAt, updatedAt: job.updatedAt, completedAt: job.completedAt };
|
||||
}
|
||||
|
||||
function boundedOutput(output, maxBytes = 12000) {
|
||||
const text = String(output?.text ?? "");
|
||||
const bytes = Buffer.byteLength(text, "utf8");
|
||||
return { ...output, text: bytes > maxBytes ? text.slice(0, maxBytes) : text, bytes: Math.min(bytes, maxBytes), truncation: { maxBytes, truncated: bytes > maxBytes } };
|
||||
}
|
||||
|
||||
function freshness(observedAt, blocker) {
|
||||
return { observedAt, ageMs: 0, stale: Boolean(blocker), source: blocker ? "blocked" : "device-pod-executor" };
|
||||
}
|
||||
|
||||
function terminalJobStatus(status) {
|
||||
return ["completed", "failed", "blocked", "canceled"].includes(status);
|
||||
}
|
||||
|
||||
function jobKey(id, jobId) {
|
||||
return `${id}\u0000${jobId}`;
|
||||
}
|
||||
|
||||
@@ -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;无在线 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 时返回 blocker。 |
|
||||
| 禁止 fake 作为 DEV-LIVE | 已实现/持续约束 | 规格和服务 payload 要求显式标记 fake/source。 |
|
||||
|
||||
|
||||
@@ -23,7 +23,8 @@
|
||||
| `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` 内部调用;当前无 gateway/device-host-cli 连接时返回 `gateway_dispatch_unavailable`。 |
|
||||
| `POST /v1/device-pods/{devicePodId}/jobs` | 只接受 `hwlab-cloud-api` 内部调用;创建内部 executor job 并返回 job、freshness、output/cancel URL;当前无 gateway/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) 为目标。
|
||||
|
||||
@@ -44,6 +45,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 创建和持久化;executor 侧当前只保留内部 job 边界。 |
|
||||
| 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。 |
|
||||
|
||||
|
||||
@@ -1,4 +1,5 @@
|
||||
import assert from "node:assert/strict";
|
||||
import { createServer } from "node:http";
|
||||
import { test } from "bun:test";
|
||||
|
||||
import { createCloudApiServer } from "./server.ts";
|
||||
@@ -204,6 +205,106 @@ test("cloud api protects device-pod routes when access control is required", asy
|
||||
}
|
||||
});
|
||||
|
||||
test("cloud api dispatches authorized device jobs to the internal device-pod executor", async () => {
|
||||
const executorRequests = [];
|
||||
const executor = createServer(async (request, response) => {
|
||||
const body = await requestJson(request);
|
||||
executorRequests.push({ method: request.method, url: request.url, internalService: request.headers["x-hwlab-internal-service"], body });
|
||||
if (request.method === "POST" && body.args?.path === "src") {
|
||||
response.writeHead(202, { "content-type": "application/json; charset=utf-8" });
|
||||
response.end(JSON.stringify({
|
||||
accepted: true,
|
||||
status: "running",
|
||||
contractVersion: "device-pod-executor-v1",
|
||||
devicePodId: "device-pod-71-freq",
|
||||
traceId: body.traceId,
|
||||
operationId: body.operationId,
|
||||
job: { id: body.jobId, devicePodId: body.devicePodId, status: "running", intent: body.intent, reason: body.reason, traceId: body.traceId, operationId: body.operationId },
|
||||
blocker: null
|
||||
}));
|
||||
return;
|
||||
}
|
||||
if (request.method === "GET" && request.url.endsWith("/output")) {
|
||||
response.writeHead(200, { "content-type": "application/json; charset=utf-8" });
|
||||
response.end(JSON.stringify({
|
||||
accepted: true,
|
||||
status: "completed",
|
||||
contractVersion: "device-pod-executor-v1",
|
||||
devicePodId: "device-pod-71-freq",
|
||||
traceId: "trc_executor_refresh",
|
||||
operationId: "op_executor_refresh",
|
||||
job: { id: request.url.split("/").at(-2), devicePodId: "device-pod-71-freq", status: "completed", intent: "workspace.ls" },
|
||||
output: { text: "executor output", bytes: 15, truncation: { maxBytes: 12000, truncated: false } },
|
||||
text: "executor output"
|
||||
}));
|
||||
return;
|
||||
}
|
||||
response.writeHead(409, { "content-type": "application/json; charset=utf-8" });
|
||||
response.end(JSON.stringify({
|
||||
accepted: false,
|
||||
status: "blocked",
|
||||
contractVersion: "device-pod-executor-v1",
|
||||
devicePodId: "device-pod-71-freq",
|
||||
traceId: body.traceId,
|
||||
operationId: body.operationId,
|
||||
job: { id: body.jobId, devicePodId: body.devicePodId, status: "blocked", intent: body.intent, reason: body.reason, traceId: body.traceId, operationId: body.operationId },
|
||||
blocker: { code: "gateway_dispatch_unavailable", layer: "device-pod", retryable: true, summary: "executor test blocker" },
|
||||
output: { text: "executor output", bytes: 15, truncation: { maxBytes: 12000, truncated: false } }
|
||||
}));
|
||||
});
|
||||
await new Promise((resolve) => executor.listen(0, "127.0.0.1", resolve));
|
||||
const executorPort = executor.address().port;
|
||||
const server = createCloudApiServer({
|
||||
env: {
|
||||
HWLAB_ACCESS_CONTROL_REQUIRED: "1",
|
||||
HWLAB_BOOTSTRAP_ADMIN_USERNAME: "admin",
|
||||
HWLAB_BOOTSTRAP_ADMIN_PASSWORD: "admin-pass",
|
||||
HWLAB_DEVICE_POD_URL: `http://127.0.0.1:${executorPort}`
|
||||
},
|
||||
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 adminLogin = await postJson(port, "/auth/login", { username: "admin", password: "admin-pass" });
|
||||
const userCreate = await postJson(port, "/v1/admin/users", { username: "alice", password: "alice-pass" }, adminLogin.cookie);
|
||||
await postJson(port, "/v1/admin/device-pods", {
|
||||
devicePodId: "device-pod-71-freq",
|
||||
profile: { schemaVersion: 1, target: { id: "target-71-freq" }, route: { gatewaySessionId: "gws_unused" } }
|
||||
}, adminLogin.cookie);
|
||||
await postJson(port, "/v1/admin/device-pod-grants", { devicePodId: "device-pod-71-freq", userId: userCreate.body.user.id }, adminLogin.cookie);
|
||||
const aliceLogin = await postJson(port, "/auth/login", { username: "alice", password: "alice-pass" });
|
||||
|
||||
const job = await postJson(port, "/v1/device-pods/device-pod-71-freq/jobs", { intent: "workspace.ls", args: { path: "." } }, aliceLogin.cookie);
|
||||
assert.equal(job.status, 409);
|
||||
assert.equal(job.body.blocker.code, "gateway_dispatch_unavailable");
|
||||
assert.equal(job.body.job.status, "blocked");
|
||||
assert.equal(executorRequests.length, 1);
|
||||
assert.equal(executorRequests[0].internalService, "hwlab-cloud-api");
|
||||
assert.equal(executorRequests[0].body.profileHash, job.body.profileHash);
|
||||
assert.equal(executorRequests[0].body.ownerUserId, userCreate.body.user.id);
|
||||
assert.equal(executorRequests[0].body.intent, "workspace.ls");
|
||||
|
||||
const stored = await getJson(port, `/v1/device-pods/device-pod-71-freq/jobs/${job.body.job.id}`, aliceLogin.cookie);
|
||||
assert.equal(stored.status, 200);
|
||||
assert.equal(stored.body.job.id, job.body.job.id);
|
||||
assert.equal(stored.body.blocker.code, "gateway_dispatch_unavailable");
|
||||
|
||||
const runningJob = await postJson(port, "/v1/device-pods/device-pod-71-freq/jobs", { intent: "workspace.ls", args: { path: "src" } }, aliceLogin.cookie);
|
||||
assert.equal(runningJob.status, 202);
|
||||
assert.equal(runningJob.body.status, "running");
|
||||
const output = await getJson(port, `/v1/device-pods/device-pod-71-freq/jobs/${runningJob.body.job.id}/output`, aliceLogin.cookie);
|
||||
assert.equal(output.status, 200);
|
||||
assert.equal(output.body.status, "completed");
|
||||
assert.equal(output.body.output.text, "executor output");
|
||||
assert.ok(executorRequests.some((item) => item.method === "GET" && item.url.endsWith(`/jobs/${runningJob.body.job.id}/output`)));
|
||||
} finally {
|
||||
await new Promise((resolve, reject) => server.close((error) => (error ? reject(error) : resolve())));
|
||||
await new Promise((resolve, reject) => executor.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" },
|
||||
@@ -267,3 +368,10 @@ async function getJson(port, path, cookie = null) {
|
||||
body: await response.json()
|
||||
};
|
||||
}
|
||||
|
||||
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) : {};
|
||||
}
|
||||
|
||||
@@ -131,10 +131,13 @@ function accessStoreForRuntime(runtimeStore, options = {}) {
|
||||
}
|
||||
|
||||
class AccessController {
|
||||
constructor({ store, env = process.env, gatewayRegistry = null, now = () => new Date().toISOString(), required = truthyFlag(env.HWLAB_ACCESS_CONTROL_REQUIRED) } = {}) {
|
||||
constructor({ store, env = process.env, gatewayRegistry = null, fetchImpl = fetch, devicePodExecutorUrl = env.HWLAB_DEVICE_POD_URL, devicePodExecutorTimeoutMs = 1200, now = () => new Date().toISOString(), required = truthyFlag(env.HWLAB_ACCESS_CONTROL_REQUIRED) } = {}) {
|
||||
this.store = store;
|
||||
this.env = env;
|
||||
this.gatewayRegistry = gatewayRegistry;
|
||||
this.fetchImpl = fetchImpl;
|
||||
this.devicePodExecutorUrl = normalizeBaseUrl(devicePodExecutorUrl);
|
||||
this.devicePodExecutorTimeoutMs = Number.parseInt(String(devicePodExecutorTimeoutMs), 10) || 1200;
|
||||
this.now = now;
|
||||
this.required = required;
|
||||
this.bootstrapAttempted = false;
|
||||
@@ -631,6 +634,26 @@ class AccessController {
|
||||
const traceId = `trc_devicepod_${randomUUID()}`;
|
||||
const operationId = `op_devicepod_${randomUUID()}`;
|
||||
const now = this.now();
|
||||
const args = normalizeObject(body.args);
|
||||
if (this.devicePodExecutorUrl) {
|
||||
const job = await this.store.createDevicePodJob({
|
||||
id: `job_devicepod_${randomUUID()}`,
|
||||
devicePodId: pod.id,
|
||||
ownerUserId: actor.id,
|
||||
status: "running",
|
||||
intent,
|
||||
args,
|
||||
reason,
|
||||
traceId,
|
||||
operationId,
|
||||
output: {},
|
||||
blocker: null,
|
||||
now,
|
||||
completedAt: null
|
||||
});
|
||||
const dispatched = await this.dispatchDevicePodExecutorJob({ job, pod, actor });
|
||||
return sendJson(response, dispatched.httpStatus, this.jobPayload(dispatched.job, pod, { lease }));
|
||||
}
|
||||
const route = pod.profile.route ?? {};
|
||||
const gatewaySessionId = textOr(route.gatewaySessionId, "");
|
||||
const gatewayOnline = gatewaySessionId && this.gatewayRegistry?.isOnline?.(gatewaySessionId) === true;
|
||||
@@ -641,7 +664,7 @@ class AccessController {
|
||||
ownerUserId: actor.id,
|
||||
status: blocker ? "blocked" : "running",
|
||||
intent,
|
||||
args: normalizeObject(body.args),
|
||||
args,
|
||||
reason,
|
||||
traceId,
|
||||
operationId,
|
||||
@@ -655,12 +678,55 @@ class AccessController {
|
||||
sendJson(response, blocker ? 409 : 202, this.jobPayload(job, pod, { lease }));
|
||||
}
|
||||
|
||||
async dispatchDevicePodExecutorJob({ job, pod, actor }) {
|
||||
const target = `${this.devicePodExecutorUrl}/v1/device-pods/${encodeURIComponent(pod.id)}/jobs`;
|
||||
try {
|
||||
const response = await fetchJsonWithTimeout(this.fetchImpl, target, {
|
||||
method: "POST",
|
||||
headers: {
|
||||
accept: "application/json",
|
||||
"content-type": "application/json",
|
||||
"x-hwlab-internal-service": CLOUD_API_SERVICE_ID
|
||||
},
|
||||
body: stableJson({
|
||||
jobId: job.id,
|
||||
devicePodId: pod.id,
|
||||
targetId: targetIdFromProfile(pod.profile),
|
||||
profileHash: pod.profileHash,
|
||||
profile: pod.profile,
|
||||
ownerUserId: actor.id,
|
||||
intent: job.intent,
|
||||
args: job.args,
|
||||
reason: job.reason,
|
||||
traceId: job.traceId,
|
||||
operationId: job.operationId
|
||||
})
|
||||
}, this.devicePodExecutorTimeoutMs);
|
||||
const updated = await this.updateDevicePodJobFromExecutorResponse({ job, pod, response });
|
||||
const status = updated.status;
|
||||
return { job: updated ?? job, httpStatus: response.status >= 400 ? response.status : status === "running" ? 202 : 200 };
|
||||
} catch (error) {
|
||||
const blocker = devicePodExecutorBlocker(error?.message ?? "device-pod executor request failed");
|
||||
const updated = await this.store.updateDevicePodJob(pod.id, job.id, {
|
||||
status: "blocked",
|
||||
output: { error: blocker.summary },
|
||||
blocker,
|
||||
completedAt: this.now(),
|
||||
updatedAt: this.now()
|
||||
});
|
||||
return { job: updated ?? job, httpStatus: 409 };
|
||||
}
|
||||
}
|
||||
|
||||
async getDevicePodJob(response, pod, route, actor) {
|
||||
const parts = route.split("/");
|
||||
const jobId = decodeURIComponent(parts[1] ?? "");
|
||||
const job = await this.store.getDevicePodJob(pod.id, jobId);
|
||||
let job = await this.store.getDevicePodJob(pod.id, jobId);
|
||||
if (!job) return sendJson(response, 404, errorPayload("device_job_not_found", `Device job ${jobId} was not found`, 404));
|
||||
if (job.ownerUserId !== actor.id && actor.role !== "admin") return sendJson(response, 403, errorPayload("device_job_owner_required", "Only the owner or admin can inspect the job", 403));
|
||||
if (this.devicePodExecutorUrl && !terminalJobStatus(job.status)) {
|
||||
job = await this.refreshDevicePodExecutorJob({ job, pod, output: parts[2] === "output" });
|
||||
}
|
||||
if (parts[2] === "output") return sendJson(response, 200, this.jobOutputPayload(job, pod));
|
||||
return sendJson(response, 200, this.jobPayload(job, pod));
|
||||
}
|
||||
@@ -670,6 +736,10 @@ class AccessController {
|
||||
const job = await this.store.getDevicePodJob(pod.id, jobId);
|
||||
if (!job) return sendJson(response, 404, errorPayload("device_job_not_found", `Device job ${jobId} was not found`, 404));
|
||||
if (job.ownerUserId !== actor.id && actor.role !== "admin") return sendJson(response, 403, errorPayload("device_job_owner_required", "Only the owner or admin can cancel the job", 403));
|
||||
if (this.devicePodExecutorUrl && !terminalJobStatus(job.status)) {
|
||||
const canceledByExecutor = await this.cancelDevicePodExecutorJob({ job, pod });
|
||||
return sendJson(response, 200, this.jobPayload(canceledByExecutor, pod));
|
||||
}
|
||||
const canceled = await this.store.updateDevicePodJob(pod.id, jobId, {
|
||||
status: terminalJobStatus(job.status) ? job.status : "canceled",
|
||||
blocker: terminalJobStatus(job.status) ? job.blocker : { code: "device_job_canceled", summary: "Device job was canceled by user" },
|
||||
@@ -679,6 +749,58 @@ class AccessController {
|
||||
return sendJson(response, 200, this.jobPayload(canceled, pod));
|
||||
}
|
||||
|
||||
async refreshDevicePodExecutorJob({ job, pod, output = false }) {
|
||||
const suffix = output ? "/output" : "";
|
||||
const target = `${this.devicePodExecutorUrl}/v1/device-pods/${encodeURIComponent(pod.id)}/jobs/${encodeURIComponent(job.id)}${suffix}`;
|
||||
try {
|
||||
const response = await fetchJsonWithTimeout(this.fetchImpl, target, {
|
||||
method: "GET",
|
||||
headers: { accept: "application/json", "x-hwlab-internal-service": CLOUD_API_SERVICE_ID }
|
||||
}, this.devicePodExecutorTimeoutMs);
|
||||
return await this.updateDevicePodJobFromExecutorResponse({ job, pod, response });
|
||||
} catch (error) {
|
||||
return await this.blockDevicePodJobOnExecutorError({ job, pod, error });
|
||||
}
|
||||
}
|
||||
|
||||
async cancelDevicePodExecutorJob({ job, pod }) {
|
||||
const target = `${this.devicePodExecutorUrl}/v1/device-pods/${encodeURIComponent(pod.id)}/jobs/${encodeURIComponent(job.id)}/cancel`;
|
||||
try {
|
||||
const response = await fetchJsonWithTimeout(this.fetchImpl, target, {
|
||||
method: "POST",
|
||||
headers: { accept: "application/json", "x-hwlab-internal-service": CLOUD_API_SERVICE_ID }
|
||||
}, this.devicePodExecutorTimeoutMs);
|
||||
return await this.updateDevicePodJobFromExecutorResponse({ job, pod, response });
|
||||
} catch (error) {
|
||||
return await this.blockDevicePodJobOnExecutorError({ job, pod, error });
|
||||
}
|
||||
}
|
||||
|
||||
async updateDevicePodJobFromExecutorResponse({ job, pod, response }) {
|
||||
const status = normalizeDeviceJobStatus(response.body?.status ?? response.body?.job?.status, response.status);
|
||||
const blocker = response.body?.blocker ?? (status === "completed" || status === "running" || status === "queued" ? null : devicePodExecutorBlocker(`device-pod executor returned HTTP ${response.status}`));
|
||||
const updated = await this.store.updateDevicePodJob(pod.id, job.id, {
|
||||
status,
|
||||
output: executorOutputPayload(response.body, response.status),
|
||||
blocker,
|
||||
completedAt: terminalJobStatus(status) ? job.completedAt ?? this.now() : null,
|
||||
updatedAt: this.now()
|
||||
});
|
||||
return updated ?? job;
|
||||
}
|
||||
|
||||
async blockDevicePodJobOnExecutorError({ job, pod, error }) {
|
||||
const blocker = devicePodExecutorBlocker(error?.message ?? "device-pod executor request failed");
|
||||
const updated = await this.store.updateDevicePodJob(pod.id, job.id, {
|
||||
status: "blocked",
|
||||
output: { error: blocker.summary },
|
||||
blocker,
|
||||
completedAt: this.now(),
|
||||
updatedAt: this.now()
|
||||
});
|
||||
return updated ?? job;
|
||||
}
|
||||
|
||||
dispatchDevicePodJob({ job, pod, actor }) {
|
||||
const gatewaySessionId = pod.profile.route?.gatewaySessionId;
|
||||
const request = createGatewayShellRequest({
|
||||
@@ -922,6 +1044,7 @@ async function jsonBody(request) {
|
||||
function requiredText(value, field) { const text = textOr(value, ""); if (!text) throw Object.assign(new Error(`${field} is required`), { statusCode: 400, code: "invalid_params" }); return text; }
|
||||
function textOr(value, fallback) { const text = String(value ?? "").trim(); return text || fallback; }
|
||||
function normalizeObject(value) { return value && typeof value === "object" && !Array.isArray(value) ? { ...value } : {}; }
|
||||
function normalizeBaseUrl(value) { const text = textOr(value, "").replace(/\/+$/u, ""); return /^https?:\/\//u.test(text) ? text : ""; }
|
||||
function boundedLeaseTtlSeconds(value) { const parsed = Number.parseInt(String(value ?? ""), 10); return Math.min(Math.max(Number.isInteger(parsed) && parsed > 0 ? parsed : DEFAULT_DEVICE_LEASE_TTL_SECONDS, 1), MAX_DEVICE_LEASE_TTL_SECONDS); }
|
||||
function safeAgentSessionId(value) { const text = textOr(value, ""); return /^ses_[A-Za-z0-9_.:-]+$/u.test(text) ? text : ""; }
|
||||
function leaseTokenFromRequest(request, body) { return textOr(body?.leaseToken, "") || textOr(getHeader(request, "x-hwlab-device-lease-token"), ""); }
|
||||
@@ -934,7 +1057,10 @@ function profileHash(profile) { return `sha256:${sha256(stableJson(profile))}`;
|
||||
function grantKey(devicePodId, userId) { return `${devicePodId}\u0000${userId}`; }
|
||||
function authoritySource() { return { kind: "CLOUD_API_PROFILE_AUTHORITY", serviceId: CLOUD_API_SERVICE_ID, fake: false, devLiveEvidence: false, note: "Profile, grant, lease, and job authority are owned by hwlab-cloud-api; this is not fake data." }; }
|
||||
function accessRoutes() { return { session: "/auth/session", login: "/auth/login", logout: "/auth/logout", v1Session: "/v1/auth/session", currentUser: "/v1/users/me", accessStatus: "/v1/access/status", setupStatus: "/v1/setup/status", devicePods: "/v1/device-pods" }; }
|
||||
function devicePodExecutorBlocker(summary) { return { code: "device_pod_executor_unavailable", layer: "device-pod", retryable: true, summary, userMessage: "Device Pod 已授权,但内部执行服务当前不可用。" }; }
|
||||
function gatewayDispatchBlocker(summary) { return { code: "gateway_dispatch_unavailable", layer: "device-pod", retryable: true, summary, userMessage: "Device Pod 已授权,但当前没有可用 gateway/device-host-cli 执行通道。" }; }
|
||||
function normalizeDeviceJobStatus(status, httpStatus) { const text = textOr(status, ""); return ["queued", "running", "completed", "failed", "blocked", "canceled"].includes(text) ? text : httpStatus >= 400 ? "blocked" : "running"; }
|
||||
function executorOutputPayload(body, httpStatus) { return { executor: body ?? {}, output: normalizeObject(body?.output), text: typeof body?.text === "string" ? body.text : typeof body?.output?.text === "string" ? body.output.text : "", httpStatus }; }
|
||||
function freshness(observedAt, blocker) { return { observedAt, ageMs: 0, stale: Boolean(blocker), source: blocker ? "blocked" : "cloud-api" }; }
|
||||
function publicActor(user) { return user ? { id: user.id, username: user.username, displayName: user.displayName, role: user.role, status: user.status } : null; }
|
||||
function redactedUser(user) { return publicActor(user); }
|
||||
@@ -965,3 +1091,15 @@ function pgJob(row) { return row ? { id: row.id, devicePodId: row.device_pod_id,
|
||||
function pgAgentSession(row) { return row ? { id: row.id, projectId: row.project_id, agentId: row.agent_id, status: row.status, startedAt: row.started_at, endedAt: row.ended_at, ownerUserId: row.owner_user_id, conversationId: row.conversation_id, threadId: row.thread_id, lastTraceId: row.last_trace_id, session: parseJson(row.session_json, {}), updatedAt: row.updated_at } : null; }
|
||||
function jobParams(job) { return [job.id, job.devicePodId, job.ownerUserId, job.status, job.intent, stableJson(job.args), job.reason, job.traceId, job.operationId, stableJson(job.output), stableJson(job.blocker ?? {}), job.createdAt, job.updatedAt, job.completedAt]; }
|
||||
function parseJson(value, fallback) { if (!value) return fallback; if (typeof value === "object") return value; try { return JSON.parse(String(value)); } catch { return fallback; } }
|
||||
|
||||
async function fetchJsonWithTimeout(fetchImpl, url, options, timeoutMs) {
|
||||
const controller = new AbortController();
|
||||
const timeout = setTimeout(() => controller.abort(), timeoutMs);
|
||||
try {
|
||||
const response = await fetchImpl(url, { ...options, signal: controller.signal });
|
||||
const text = await response.text();
|
||||
return { status: response.status, body: parseJson(text, {}) };
|
||||
} finally {
|
||||
clearTimeout(timeout);
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user