Merge pull request #556 from pikasTech/fix/v02-device-pod-bounded-output
fix: bound device pod job output
This commit is contained in:
@@ -65,6 +65,7 @@ test("device pod executor exposes cloud-api authority boundary and method guards
|
|||||||
assert.equal(storedOutput.job.id, "job_devicepod_test");
|
assert.equal(storedOutput.job.id, "job_devicepod_test");
|
||||||
assert.equal(storedOutput.output.summary, "HWLAB_CLOUD_API_INTERNAL_URL is not configured for gateway dispatch");
|
assert.equal(storedOutput.output.summary, "HWLAB_CLOUD_API_INTERNAL_URL is not configured for gateway dispatch");
|
||||||
assert.equal(storedOutput.truncation.truncated, false);
|
assert.equal(storedOutput.truncation.truncated, false);
|
||||||
|
assert.equal(storedOutput.truncation.originalBytes, 0);
|
||||||
|
|
||||||
const externalJobRead = await fetch(`http://127.0.0.1:${service.port}/v1/device-pods/device-pod-test/jobs/job_devicepod_test`);
|
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(externalJobRead.status, 403);
|
||||||
@@ -89,6 +90,61 @@ test("device pod executor exposes cloud-api authority boundary and method guards
|
|||||||
}
|
}
|
||||||
});
|
});
|
||||||
|
|
||||||
|
test("device-pod executor bounds gateway output text", async () => {
|
||||||
|
const longText = "z".repeat(13000);
|
||||||
|
const cloudApi = createServer(async (request, response) => {
|
||||||
|
const body = await requestJson(request);
|
||||||
|
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",
|
||||||
|
dispatch: { shellExecuted: true, dispatchStatus: "succeeded", stdout: longText, exitCode: 0 }
|
||||||
|
}
|
||||||
|
}));
|
||||||
|
});
|
||||||
|
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_bounded_output_test",
|
||||||
|
intent: "workspace.ls",
|
||||||
|
args: { path: "src" },
|
||||||
|
traceId: "trc_bounded_output_test",
|
||||||
|
operationId: "op_bounded_output_test",
|
||||||
|
targetId: "target-device-pod-test",
|
||||||
|
profileHash: "sha256:test",
|
||||||
|
profile: {
|
||||||
|
schemaVersion: 1,
|
||||||
|
devicePodId: "device-pod-test",
|
||||||
|
target: { id: "target-device-pod-test" },
|
||||||
|
route: { gatewaySessionId: "gws_devicepod_test", resourceId: "res_devicepod_test", capabilityId: "cap_device_host_cli" }
|
||||||
|
}
|
||||||
|
})
|
||||||
|
});
|
||||||
|
assert.equal(jobResponse.status, 202);
|
||||||
|
const output = await waitForJobStatus(service.port, "device-pod-test", "job_bounded_output_test", "completed");
|
||||||
|
assert.equal(output.output.text.length, 12000);
|
||||||
|
assert.equal(output.text.length, 12000);
|
||||||
|
assert.equal(output.bytes, 12000);
|
||||||
|
assert.equal(output.truncation.truncated, true);
|
||||||
|
assert.equal(output.truncation.originalBytes, 13000);
|
||||||
|
assert.equal(JSON.stringify(output).includes(longText), false);
|
||||||
|
} finally {
|
||||||
|
await service.stop();
|
||||||
|
await new Promise((resolve, reject) => cloudApi.close((error) => (error ? reject(error) : resolve())));
|
||||||
|
}
|
||||||
|
});
|
||||||
|
|
||||||
test("device pod executor dispatches internal jobs through cloud-api gateway adapter", async () => {
|
test("device pod executor dispatches internal jobs through cloud-api gateway adapter", async () => {
|
||||||
const dispatches = [];
|
const dispatches = [];
|
||||||
const cloudApi = createServer(async (request, response) => {
|
const cloudApi = createServer(async (request, response) => {
|
||||||
|
|||||||
@@ -12,6 +12,7 @@ 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 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 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 dispatchTimeoutMs = numberOr(process.env.HWLAB_DEVICE_POD_GATEWAY_DISPATCH_TIMEOUT_MS, 120000);
|
||||||
|
const DEVICE_JOB_OUTPUT_MAX_BYTES = 12000;
|
||||||
const jobs = new Map();
|
const jobs = new Map();
|
||||||
|
|
||||||
listen(createServer(async (request, response) => {
|
listen(createServer(async (request, response) => {
|
||||||
@@ -473,10 +474,17 @@ 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 };
|
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) {
|
function boundedOutput(output, maxBytes = DEVICE_JOB_OUTPUT_MAX_BYTES) {
|
||||||
const text = String(output?.text ?? "");
|
const text = String(output?.text ?? "");
|
||||||
const bytes = Buffer.byteLength(text, "utf8");
|
const buffer = Buffer.from(text, "utf8");
|
||||||
return { ...output, text: bytes > maxBytes ? text.slice(0, maxBytes) : text, bytes: Math.min(bytes, maxBytes), truncation: { maxBytes, truncated: bytes > maxBytes } };
|
const clipped = buffer.length > maxBytes;
|
||||||
|
const boundedText = clipped ? buffer.subarray(0, maxBytes).toString("utf8") : text;
|
||||||
|
const bounded = { ...output, text: boundedText };
|
||||||
|
if (clipped) {
|
||||||
|
delete bounded.dispatch;
|
||||||
|
bounded.omitted = { reason: "device_job_output_truncated", originalBytes: buffer.length };
|
||||||
|
}
|
||||||
|
return { ...bounded, bytes: Math.min(buffer.length, maxBytes), truncation: { maxBytes, truncated: clipped, originalBytes: buffer.length } };
|
||||||
}
|
}
|
||||||
|
|
||||||
function freshness(observedAt, blocker) {
|
function freshness(observedAt, blocker) {
|
||||||
|
|||||||
@@ -203,7 +203,7 @@ DELETE /v1/admin/device-pod-grants/{devicePodId}/{userId}
|
|||||||
- `io.uart.read-after-launch-flash`
|
- `io.uart.read-after-launch-flash`
|
||||||
- `io.uart.write`
|
- `io.uart.write`
|
||||||
|
|
||||||
所有响应必须包含 `devicePodId`、`targetId`、`profileHash`、`traceId`、`operationId`、`status`、`freshness`、`blocker` 和 bounded output metadata。真实硬件响应不得把 fake、dry-run、SOURCE、LOCAL 或过期缓存标为 `DEV-LIVE`。
|
所有 job/status/output 响应必须包含 `devicePodId`、`targetId`、`profileHash`、`traceId`、`operationId`、`status`、`freshness`、`blocker` 和 bounded output metadata。job output 文本默认最大 12000 bytes;超出时必须设置 `truncation.truncated=true`、`truncation.originalBytes`,并避免把完整 executor/gateway 原始输出嵌回 JSON。真实硬件响应不得把 fake、dry-run、SOURCE、LOCAL 或过期缓存标为 `DEV-LIVE`。
|
||||||
|
|
||||||
## 微服务职责
|
## 微服务职责
|
||||||
|
|
||||||
|
|||||||
@@ -25,7 +25,7 @@
|
|||||||
| `GET /v1/device-pods/{devicePodId}/status` | 返回 blocked executor status;真实 profile/status 以 cloud-api `/v1/device-pods/{devicePodId}/status` 为准。 |
|
| `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,不伪造硬件事件。 |
|
| `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;有可用 profile route 和 `HWLAB_CLOUD_API_INTERNAL_URL` 时通过 cloud-api gateway dispatch 下发到 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。 |
|
| `GET /v1/device-pods/{devicePodId}/jobs/{jobId}`、`GET /output`、`POST /cancel` | 只接受 `hwlab-cloud-api` 内部调用,用于查询内部 job、最大 12000 bytes 的 bounded output 和取消非终态 job;普通用户仍必须走 cloud-api 用户态 API。 |
|
||||||
|
|
||||||
用户态 `POST /jobs`、job output/cancel、admin profile/grant API 和正式 `device-pod-cli` REST 调用由 `hwlab-cloud-api` 实现,应以 [spec-device-pod.md](spec-device-pod.md) 为目标。`hwlab-device-pod` 不接受 CLI、浏览器或 Code Agent 直接上传 profile snapshot。
|
用户态 `POST /jobs`、job output/cancel、admin profile/grant API 和正式 `device-pod-cli` REST 调用由 `hwlab-cloud-api` 实现,应以 [spec-device-pod.md](spec-device-pod.md) 为目标。`hwlab-device-pod` 不接受 CLI、浏览器或 Code Agent 直接上传 profile snapshot。
|
||||||
|
|
||||||
|
|||||||
@@ -298,6 +298,7 @@ test("cloud api dispatches authorized device jobs to the internal device-pod exe
|
|||||||
assert.equal(output.status, 200);
|
assert.equal(output.status, 200);
|
||||||
assert.equal(output.body.status, "completed");
|
assert.equal(output.body.status, "completed");
|
||||||
assert.equal(output.body.output.text, "executor output");
|
assert.equal(output.body.output.text, "executor output");
|
||||||
|
assert.equal(output.body.truncation.truncated, false);
|
||||||
assert.ok(executorRequests.some((item) => item.method === "GET" && item.url.endsWith(`/jobs/${runningJob.body.job.id}/output`)));
|
assert.ok(executorRequests.some((item) => item.method === "GET" && item.url.endsWith(`/jobs/${runningJob.body.job.id}/output`)));
|
||||||
} finally {
|
} finally {
|
||||||
await new Promise((resolve, reject) => server.close((error) => (error ? reject(error) : resolve())));
|
await new Promise((resolve, reject) => server.close((error) => (error ? reject(error) : resolve())));
|
||||||
@@ -305,6 +306,64 @@ test("cloud api dispatches authorized device jobs to the internal device-pod exe
|
|||||||
}
|
}
|
||||||
});
|
});
|
||||||
|
|
||||||
|
test("cloud api bounds device-pod job output payloads", async () => {
|
||||||
|
const longText = "x".repeat(13000);
|
||||||
|
const executor = createServer(async (request, response) => {
|
||||||
|
const body = await requestJson(request);
|
||||||
|
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: body.traceId,
|
||||||
|
operationId: body.operationId,
|
||||||
|
job: { id: body.jobId, devicePodId: body.devicePodId, status: "completed", intent: body.intent },
|
||||||
|
output: { text: longText, nested: { kept: true } },
|
||||||
|
text: longText
|
||||||
|
}));
|
||||||
|
});
|
||||||
|
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, 200);
|
||||||
|
const output = await getJson(port, `/v1/device-pods/device-pod-71-freq/jobs/${job.body.job.id}/output`, aliceLogin.cookie);
|
||||||
|
assert.equal(output.status, 200);
|
||||||
|
assert.equal(output.body.bytes, 12000);
|
||||||
|
assert.equal(output.body.text.length, 12000);
|
||||||
|
assert.equal(output.body.output.text.length, 12000);
|
||||||
|
assert.equal(output.body.truncation.truncated, true);
|
||||||
|
assert.equal(output.body.truncation.originalBytes, 13000);
|
||||||
|
assert.equal(output.body.output.omitted.reason, "device_job_output_truncated");
|
||||||
|
assert.equal(JSON.stringify(output.body).includes(longText), false);
|
||||||
|
} 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 internal device-pod gateway dispatch is service-only and fail-closed without online gateway", async () => {
|
test("cloud api internal device-pod gateway dispatch is service-only and fail-closed without online gateway", async () => {
|
||||||
const server = createCloudApiServer({
|
const server = createCloudApiServer({
|
||||||
env: { HWLAB_ACCESS_CONTROL_REQUIRED: "1" },
|
env: { HWLAB_ACCESS_CONTROL_REQUIRED: "1" },
|
||||||
|
|||||||
@@ -117,6 +117,7 @@ const MUTATING_INTENTS = new Set([
|
|||||||
]);
|
]);
|
||||||
const DEFAULT_DEVICE_LEASE_TTL_SECONDS = 15 * 60;
|
const DEFAULT_DEVICE_LEASE_TTL_SECONDS = 15 * 60;
|
||||||
const MAX_DEVICE_LEASE_TTL_SECONDS = 60 * 60;
|
const MAX_DEVICE_LEASE_TTL_SECONDS = 60 * 60;
|
||||||
|
const DEVICE_JOB_OUTPUT_MAX_BYTES = 12000;
|
||||||
|
|
||||||
export function createAccessController(options = {}) {
|
export function createAccessController(options = {}) {
|
||||||
return new AccessController({
|
return new AccessController({
|
||||||
@@ -865,13 +866,13 @@ class AccessController {
|
|||||||
}
|
}
|
||||||
|
|
||||||
jobOutputPayload(job, pod) {
|
jobOutputPayload(job, pod) {
|
||||||
const text = typeof job.output?.text === "string" ? job.output.text : JSON.stringify(job.output ?? {});
|
const bounded = boundedOutput(job.output, DEVICE_JOB_OUTPUT_MAX_BYTES);
|
||||||
return {
|
return {
|
||||||
...this.jobPayload(job, pod),
|
...this.jobPayload(job, pod),
|
||||||
output: job.output ?? {},
|
output: bounded.output,
|
||||||
text,
|
text: bounded.text,
|
||||||
bytes: Buffer.byteLength(text, "utf8"),
|
bytes: bounded.bytes,
|
||||||
truncation: { maxBytes: 12000, truncated: false }
|
truncation: bounded.truncation
|
||||||
};
|
};
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -1063,6 +1064,20 @@ function devicePodExecutorBlocker(summary) { return { code: "device_pod_executor
|
|||||||
function gatewayDispatchBlocker(summary) { return { code: "gateway_dispatch_unavailable", layer: "device-pod", retryable: true, summary, userMessage: "Device Pod 已授权,但当前没有可用 gateway/device-host-cli 执行通道。" }; }
|
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 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 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 boundedOutput(output, maxBytes = DEVICE_JOB_OUTPUT_MAX_BYTES) {
|
||||||
|
const source = normalizeObject(output);
|
||||||
|
const text = typeof source.text === "string" ? source.text : stableJson(source);
|
||||||
|
const buffer = Buffer.from(text, "utf8");
|
||||||
|
const clipped = buffer.length > maxBytes;
|
||||||
|
const boundedText = clipped ? buffer.subarray(0, maxBytes).toString("utf8") : text;
|
||||||
|
const bounded = { ...source, text: boundedText };
|
||||||
|
if (clipped) {
|
||||||
|
delete bounded.executor;
|
||||||
|
delete bounded.output;
|
||||||
|
bounded.omitted = { reason: "device_job_output_truncated", originalBytes: buffer.length };
|
||||||
|
}
|
||||||
|
return { output: bounded, text: boundedText, bytes: Math.min(buffer.length, maxBytes), truncation: { maxBytes, truncated: clipped, originalBytes: buffer.length } };
|
||||||
|
}
|
||||||
function freshness(observedAt, blocker) { return { observedAt, ageMs: 0, stale: Boolean(blocker), source: blocker ? "blocked" : "cloud-api" }; }
|
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 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); }
|
function redactedUser(user) { return publicActor(user); }
|
||||||
|
|||||||
Reference in New Issue
Block a user