From 2da7f2dd5ac7f778e56c2c81f9b4683555a3419d Mon Sep 17 00:00:00 2001 From: hkimw <54717101+hkimw@users.noreply.github.com> Date: Wed, 6 May 2026 18:57:07 +0900 Subject: [PATCH 1/2] feat(kv260): add read-only status surface for IDE --- docs/EDITOR_BRIDGE_CONTRACT.md | 7 + docs/KV260_READ_ONLY_STATUS_SURFACE.md | 40 ++ .../lab-trace-manifest.example.json | 25 ++ .../launcher-npu-status.example.json | 7 + editors/vscode-prototype/README.md | 7 + editors/vscode-prototype/package.json | 7 +- editors/vscode-prototype/src/config.mjs | 5 + editors/vscode-prototype/src/extension.mjs | 31 ++ .../src/kv260-status-panel.mjs | 353 ++++++++++++++++++ .../test/extension-entrypoint.test.mjs | 19 + .../test/extension-host/smoke-suite.cjs | 17 + .../test/extension-manifest.test.mjs | 2 + .../test/kv260-status-panel.test.mjs | 113 ++++++ .../test/static-boundary.test.mjs | 1 + pyproject.toml | 1 + src/pccx_ide_cli/cli.py | 52 +++ src/pccx_ide_cli/kv260_status.py | 331 ++++++++++++++++ tests/test_kv260_status.py | 122 ++++++ 18 files changed, 1139 insertions(+), 1 deletion(-) create mode 100644 docs/KV260_READ_ONLY_STATUS_SURFACE.md create mode 100644 docs/examples/kv260-status/lab-trace-manifest.example.json create mode 100644 docs/examples/kv260-status/launcher-npu-status.example.json create mode 100644 editors/vscode-prototype/src/kv260-status-panel.mjs create mode 100644 editors/vscode-prototype/test/kv260-status-panel.test.mjs create mode 100644 src/pccx_ide_cli/kv260_status.py create mode 100644 tests/test_kv260_status.py diff --git a/docs/EDITOR_BRIDGE_CONTRACT.md b/docs/EDITOR_BRIDGE_CONTRACT.md index 0d679e3..557eedc 100644 --- a/docs/EDITOR_BRIDGE_CONTRACT.md +++ b/docs/EDITOR_BRIDGE_CONTRACT.md @@ -332,3 +332,10 @@ xsim path, and text surface sketches is documented in - No MCP server implementation in this repo today. - No pccx-llm-launcher runtime call yet; future integration requires an explicit reviewed contract. +- The KV260 status surface is a read-only local display over launcher NPU + status data and pccx-lab trace manifest JSON. The VS Code prototype command + `pccxSystemVerilog.showKv260StatusPanel` and CLI command `sv-ide + kv260-status` render fixture or explicitly supplied local JSON data only; + they do not invoke the launcher, invoke pccx-lab, run shell commands, open + SSH, control KV260 hardware, or write back status. The surface is documented + in [`KV260_READ_ONLY_STATUS_SURFACE.md`](./KV260_READ_ONLY_STATUS_SURFACE.md). diff --git a/docs/KV260_READ_ONLY_STATUS_SURFACE.md b/docs/KV260_READ_ONLY_STATUS_SURFACE.md new file mode 100644 index 0000000..025cf8a --- /dev/null +++ b/docs/KV260_READ_ONLY_STATUS_SURFACE.md @@ -0,0 +1,40 @@ +# KV260 Read-Only Status Surface + +This page documents the local IDE status surface for KV260 data. The surface +parses existing JSON data and renders a checklist before a future run path is +considered. It does not open SSH, run board commands, invoke the launcher, +invoke pccx-lab, load a bitstream, access AXI, scan networks, mutate files, or +write back status. + +## Inputs + +- `LauncherStatusReader` consumes the launcher `NPUStatus` shape from + `pccxai/pccx-llm-launcher#70`: `bitstream_loaded`, `bitstream_uuid`, + `axi_base_addr`, `axi_stat_register_value`, and `last_error`. +- `LabTraceReader` parses the lab `TraceManifest` JSON shape from + `pccxai/pccx-lab#160` for file-replay trace manifests. + +The launcher type is mirrored locally because this repository does not import +the launcher contract package. + +## Rendered Surface + +`Kv260StatusPanel` renders launcher status, lab manifest metadata, and a +`PreflightProposal` checklist: + +- bitstream loaded +- AXI reachable +- manifest available + +The checklist is display-only. A blocked item is evidence that the IDE should +keep any future KV260 run path gated until lower layers provide reviewed data. + +## CLI + +```bash +sv-ide kv260-status +``` + +The command reads the bundled tiny fixtures by default and prints the same +status panel as text. Optional local JSON paths can be supplied with +`--launcher-status` and `--trace-manifest`. diff --git a/docs/examples/kv260-status/lab-trace-manifest.example.json b/docs/examples/kv260-status/lab-trace-manifest.example.json new file mode 100644 index 0000000..a55f100 --- /dev/null +++ b/docs/examples/kv260-status/lab-trace-manifest.example.json @@ -0,0 +1,25 @@ +{ + "schema_version": "pccx.lab.kv260.trace-manifest.v0", + "bitstream_uuid": "00000000-0000-0000-0000-000000000160", + "axi_base": "0x00000000a0000000", + "isa_version": "pccx_v002", + "frame_count": 1, + "checksums": [ + { + "algorithm": "sha256", + "value": "fixture-only-not-a-hardware-capture", + "frame_idx": null + } + ], + "runbook_ref": "pccxai/pccx-lab#160", + "source_kind": "file_replay", + "frames": [ + { + "frame_idx": 0, + "axi_stat_register_value": 1, + "engine_completion_mask": 3, + "cycle_count": 128, + "result_payload": "a55a0001" + } + ] +} diff --git a/docs/examples/kv260-status/launcher-npu-status.example.json b/docs/examples/kv260-status/launcher-npu-status.example.json new file mode 100644 index 0000000..f093c6d --- /dev/null +++ b/docs/examples/kv260-status/launcher-npu-status.example.json @@ -0,0 +1,7 @@ +{ + "bitstream_loaded": false, + "bitstream_uuid": null, + "axi_base_addr": null, + "axi_stat_register_value": null, + "last_error": "fixture only; lower-layer evidence not supplied" +} diff --git a/editors/vscode-prototype/README.md b/editors/vscode-prototype/README.md index 5ebd2d1..ff86bc5 100644 --- a/editors/vscode-prototype/README.md +++ b/editors/vscode-prototype/README.md @@ -147,6 +147,7 @@ The contributed commands are: - `pccxSystemVerilog.showContextBundleAudit` - `pccxSystemVerilog.showPccxLabBackendStatus` - `pccxSystemVerilog.showDiagnosticsHandoffSummary` +- `pccxSystemVerilog.showKv260StatusPanel` The prototype-only settings are: @@ -559,6 +560,7 @@ node editors/vscode-prototype/test/runtime-readiness-consumer.test.mjs node editors/vscode-prototype/test/runtime-readiness-status-surface.test.mjs node editors/vscode-prototype/test/device-session-status-consumer.test.mjs node editors/vscode-prototype/test/device-session-status-surface.test.mjs +node editors/vscode-prototype/test/kv260-status-panel.test.mjs node editors/vscode-prototype/test/local-workflow-status.test.mjs node editors/vscode-prototype/test/context-bundle-audit.test.mjs node editors/vscode-prototype/test/validation-result-summary.test.mjs @@ -597,3 +599,8 @@ node editors/vscode-prototype/src/adapter.mjs diagnostics \ node editors/vscode-prototype/src/adapter.mjs navigation \ docs/examples/editor-bridge/declarations.example.json ``` + +The KV260 status surface is a read-only panel over local launcher NPU status +and pccx-lab trace manifest data. `pccxSystemVerilog.showKv260StatusPanel` +renders the bundled tiny fixture data and does not invoke the launcher, +pccx-lab, shell, SSH, or hardware paths. diff --git a/editors/vscode-prototype/package.json b/editors/vscode-prototype/package.json index b9e0650..8d4d713 100644 --- a/editors/vscode-prototype/package.json +++ b/editors/vscode-prototype/package.json @@ -34,7 +34,8 @@ "onCommand:pccxSystemVerilog.showLocalWorkflowStatus", "onCommand:pccxSystemVerilog.showContextBundleAudit", "onCommand:pccxSystemVerilog.showPccxLabBackendStatus", - "onCommand:pccxSystemVerilog.showDiagnosticsHandoffSummary" + "onCommand:pccxSystemVerilog.showDiagnosticsHandoffSummary", + "onCommand:pccxSystemVerilog.showKv260StatusPanel" ], "contributes": { "commands": [ @@ -125,6 +126,10 @@ { "command": "pccxSystemVerilog.showDiagnosticsHandoffSummary", "title": "PCCX SystemVerilog: Show Diagnostics Handoff Summary (Experimental)" + }, + { + "command": "pccxSystemVerilog.showKv260StatusPanel", + "title": "PCCX SystemVerilog: Show KV260 Status Panel (Experimental)" } ], "configuration": { diff --git a/editors/vscode-prototype/src/config.mjs b/editors/vscode-prototype/src/config.mjs index 4b15ccb..0f0ae2c 100644 --- a/editors/vscode-prototype/src/config.mjs +++ b/editors/vscode-prototype/src/config.mjs @@ -56,11 +56,16 @@ export const DIAGNOSTICS_HANDOFF_COMMAND_IDS = Object.freeze([ "pccxSystemVerilog.showDiagnosticsHandoffSummary", ]); +export const KV260_STATUS_COMMAND_IDS = Object.freeze([ + "pccxSystemVerilog.showKv260StatusPanel", +]); + export const COMMAND_IDS = Object.freeze([ ...FACADE_COMMAND_IDS, ...WORKFLOW_COMMAND_IDS, ...PCCX_LAB_COMMAND_IDS, ...DIAGNOSTICS_HANDOFF_COMMAND_IDS, + ...KV260_STATUS_COMMAND_IDS, ]); export const MODES = Object.freeze(["checkedExample", "liveWorkspace"]); diff --git a/editors/vscode-prototype/src/extension.mjs b/editors/vscode-prototype/src/extension.mjs index cb1eec5..bf516c8 100644 --- a/editors/vscode-prototype/src/extension.mjs +++ b/editors/vscode-prototype/src/extension.mjs @@ -72,6 +72,10 @@ import { createDiagnosticsHandoffStatusSurface, formatDiagnosticsHandoffStatusSurface, } from "./diagnostics-handoff-status-surface.mjs"; +import { + createKv260StatusPanel, + formatKv260StatusPanel, +} from "./kv260-status-panel.mjs"; const EXTENSION_ROOT = resolve(dirname(fileURLToPath(import.meta.url)), ".."); const DEFAULT_DIAGNOSTIC_FILE_ROOT = resolve(EXTENSION_ROOT, "../.."); @@ -113,6 +117,8 @@ export const PCCX_LAB_BACKEND_STATUS_COMMAND = "pccxSystemVerilog.showPccxLabBackendStatus"; export const SHOW_DIAGNOSTICS_HANDOFF_SUMMARY_COMMAND = "pccxSystemVerilog.showDiagnosticsHandoffSummary"; +export const SHOW_KV260_STATUS_PANEL_COMMAND = + "pccxSystemVerilog.showKv260StatusPanel"; const NAVIGATION_LOCATION_COMMAND_IDS = Object.freeze([ CHECKED_EXAMPLE_NAVIGATION_COMMAND, @@ -428,6 +434,9 @@ function appendCommandOutput(outputChannel, commandId, result) { if (result.kind === "diagnostics-handoff-status") { outputChannel.appendLine(formatDiagnosticsHandoffStatusSurface(result.surface)); } + if (result.kind === "kv260-status-panel") { + outputChannel.appendLine(formatKv260StatusPanel(result.panel)); + } if (result.status?.kind === "pccx-lab-backend-status") { outputChannel.appendLine(JSON.stringify(result.status, null, 2)); } @@ -1106,6 +1115,28 @@ export function createCommandHandler(commandId, vscodeApi, runtime = {}) { return result; } + if (commandId === SHOW_KV260_STATUS_PANEL_COMMAND) { + let result; + try { + const panel = createKv260StatusPanel(runtime.kv260StatusInputs); + result = { + ok: true, + commandId, + kind: "kv260-status-panel", + panel, + }; + vscodeApi?.window?.showInformationMessage?.( + `KV260 status surface: ${panel.lab.frameCount} trace frame(s).`, + result, + ); + } catch (error) { + result = { ok: false, commandId, error: error.message }; + vscodeApi?.window?.showWarningMessage?.(result.error, result); + } + appendCommandOutput(runtime.outputChannel, commandId, result); + return result; + } + if (commandId === APPROVED_VALIDATION_RUNNER_COMMAND) { let result; try { diff --git a/editors/vscode-prototype/src/kv260-status-panel.mjs b/editors/vscode-prototype/src/kv260-status-panel.mjs new file mode 100644 index 0000000..703f576 --- /dev/null +++ b/editors/vscode-prototype/src/kv260-status-panel.mjs @@ -0,0 +1,353 @@ +// SPDX-License-Identifier: Apache-2.0 +// Copyright 2026 pccxai + +export const LAUNCHER_NPU_STATUS_MIRROR_VERSION = + "pccx.ide.launcher-npu-status.local-mirror.v0"; +export const TRACE_MANIFEST_SCHEMA_VERSION = "pccx.lab.kv260.trace-manifest.v0"; +export const KV260_STATUS_PANEL_VERSION = "pccx.ide.kv260-status-panel.v0"; + +export const DEFAULT_LAUNCHER_NPU_STATUS = Object.freeze({ + bitstream_loaded: false, + bitstream_uuid: null, + axi_base_addr: null, + axi_stat_register_value: null, + last_error: "fixture only; lower-layer evidence not supplied", +}); + +export const DEFAULT_TRACE_MANIFEST = Object.freeze({ + schema_version: TRACE_MANIFEST_SCHEMA_VERSION, + bitstream_uuid: "00000000-0000-0000-0000-000000000160", + axi_base: "0x00000000a0000000", + isa_version: "pccx_v002", + frame_count: 1, + checksums: Object.freeze([ + Object.freeze({ + algorithm: "sha256", + value: "fixture-only-not-a-hardware-capture", + frame_idx: null, + }), + ]), + runbook_ref: "pccxai/pccx-lab#160", + source_kind: "file_replay", + frames: Object.freeze([ + Object.freeze({ + frame_idx: 0, + axi_stat_register_value: 1, + engine_completion_mask: 3, + cycle_count: 128, + result_payload: "a55a0001", + }), + ]), +}); + +const SECRET_ASSIGNMENT_PATTERN = + /\b(?:api[_-]?key|authorization|bearer|client[_-]?secret|password|private[_-]?key|secret|token)\b\s*[:=]/i; +const PRIVATE_PATH_PATTERN = /(?:\/home\/[^/\s]+|\/Users\/[^/\s]+|[A-Za-z]:\\Users\\)/; +const MODEL_ARTIFACT_PATTERN = + /\.(?:gguf|safetensors|ckpt|pt|pth|onnx|xclbin|bit)(?:\s|$|["'])/i; +const UNSUPPORTED_MARKER_PARTS = Object.freeze([ + ["production", "ready"], + ["marketplace", "ready"], + ["stable", "api"], + ["stable", "abi"], + ["kv260 inference ", "works"], + ["gemma 3n e4b runs on ", "kv260"], + ["20 tok/s ", "achieved"], + ["timing ", "closed"], +]); + +function clone(value) { + return JSON.parse(JSON.stringify(value)); +} + +function isObject(value) { + return value && typeof value === "object" && !Array.isArray(value); +} + +function addError(errors, path, message) { + errors.push(`${path}: ${message}`); +} + +function assertSafeText(value, errors) { + let text = ""; + try { + text = JSON.stringify(value ?? {}); + } catch { + addError(errors, "input", "must be JSON-serializable"); + return; + } + const lower = text.toLowerCase(); + if ( + SECRET_ASSIGNMENT_PATTERN.test(text) || + PRIVATE_PATH_PATTERN.test(text) || + MODEL_ARTIFACT_PATTERN.test(text) + ) { + addError(errors, "input", "must not include secrets, private paths, or model artifact paths"); + } + for (const parts of UNSUPPORTED_MARKER_PARTS) { + if (lower.includes(parts.join(""))) { + addError(errors, "input", "must not include unsupported runtime or readiness claims"); + } + } +} + +function stringField(value, path, errors, maxCharacters = 500) { + if (typeof value !== "string" || value.trim().length === 0) { + addError(errors, path, "must be a non-empty string"); + return ""; + } + if ( + value.length > maxCharacters || + value.includes("\0") || + value.includes("\n") || + value.includes("\r") + ) { + addError(errors, path, `must be a single-line string up to ${maxCharacters} characters`); + } + return value; +} + +function optionalStringField(value, path, errors, maxCharacters = 500) { + if (value == null) { + return null; + } + return stringField(value, path, errors, maxCharacters); +} + +function boolField(value, path, errors) { + if (typeof value !== "boolean") { + addError(errors, path, "must be a boolean"); + return false; + } + return value; +} + +function optionalIntegerField(value, path, errors) { + if (value == null) { + return null; + } + if (!Number.isSafeInteger(value) || value < 0) { + addError(errors, path, "must be null or a non-negative integer"); + return null; + } + return value; +} + +function integerField(value, path, errors) { + if (!Number.isSafeInteger(value) || value < 0) { + addError(errors, path, "must be a non-negative integer"); + return 0; + } + return value; +} + +function arrayField(value, path, errors) { + if (!Array.isArray(value)) { + addError(errors, path, "must be an array"); + return []; + } + return value; +} + +function normalizeTraceFrame(value, index, errors) { + const frame = isObject(value) ? value : {}; + if (!isObject(value)) { + addError(errors, `frames[${index}]`, "must be an object"); + } + const resultPayload = frame.result_payload == null + ? null + : stringField(frame.result_payload, `frames[${index}].result_payload`, errors, 10000); + const error = frame.error == null ? null : frame.error; + if (error != null && !isObject(error)) { + addError(errors, `frames[${index}].error`, "must be null or an object"); + } + return { + frame_idx: integerField(frame.frame_idx, `frames[${index}].frame_idx`, errors), + axi_stat_register_value: integerField( + frame.axi_stat_register_value, + `frames[${index}].axi_stat_register_value`, + errors, + ), + engine_completion_mask: integerField( + frame.engine_completion_mask, + `frames[${index}].engine_completion_mask`, + errors, + ), + cycle_count: integerField(frame.cycle_count, `frames[${index}].cycle_count`, errors), + result_payload: resultPayload, + error, + }; +} + +export class LauncherStatusReader { + static consume(status = DEFAULT_LAUNCHER_NPU_STATUS) { + const errors = []; + if (!isObject(status)) { + throw new Error("launcher NPU status must be an object"); + } + assertSafeText(status, errors); + const normalized = { + bitstream_loaded: boolField(status.bitstream_loaded, "bitstream_loaded", errors), + bitstream_uuid: optionalStringField(status.bitstream_uuid, "bitstream_uuid", errors, 160), + axi_base_addr: optionalIntegerField(status.axi_base_addr, "axi_base_addr", errors), + axi_stat_register_value: optionalIntegerField( + status.axi_stat_register_value, + "axi_stat_register_value", + errors, + ), + last_error: optionalStringField(status.last_error, "last_error", errors, 500), + }; + if (errors.length > 0) { + throw new Error(errors.join("; ")); + } + return Object.freeze(normalized); + } +} + +export class LabTraceReader { + static consume(manifest = DEFAULT_TRACE_MANIFEST) { + const errors = []; + if (!isObject(manifest)) { + throw new Error("trace manifest must be an object"); + } + assertSafeText(manifest, errors); + const schemaVersion = stringField(manifest.schema_version, "schema_version", errors); + if (schemaVersion !== TRACE_MANIFEST_SCHEMA_VERSION) { + addError(errors, "schema_version", `must be ${TRACE_MANIFEST_SCHEMA_VERSION}`); + } + const sourceKind = stringField(manifest.source_kind, "source_kind", errors); + if (sourceKind !== "file_replay") { + addError(errors, "source_kind", "must be file_replay"); + } + const frames = arrayField(manifest.frames, "frames", errors) + .map((frame, index) => normalizeTraceFrame(frame, index, errors)); + const frameCount = integerField(manifest.frame_count, "frame_count", errors); + if (frameCount !== frames.length) { + addError(errors, "frame_count", "must match inline frames"); + } + const normalized = { + schema_version: schemaVersion, + bitstream_uuid: stringField(manifest.bitstream_uuid, "bitstream_uuid", errors, 160), + axi_base: stringField(manifest.axi_base, "axi_base", errors, 80), + isa_version: stringField(manifest.isa_version, "isa_version", errors, 80), + frame_count: frameCount, + checksums: arrayField(manifest.checksums, "checksums", errors).map(clone), + runbook_ref: stringField(manifest.runbook_ref, "runbook_ref", errors, 240), + source_kind: sourceKind, + frames, + }; + if (errors.length > 0) { + throw new Error(errors.join("; ")); + } + return Object.freeze(normalized); + } +} + +function hexOrUnavailable(value) { + return value == null ? "unavailable" : `0x${value.toString(16)}`; +} + +function yesNo(value) { + return value ? "yes" : "no"; +} + +export function createPreflightProposal(launcherStatus, traceManifest) { + const axiPresent = + launcherStatus.axi_base_addr != null && launcherStatus.axi_stat_register_value != null; + return Object.freeze({ + kind: "kv260-preflight-proposal", + items: Object.freeze([ + Object.freeze({ + itemId: "bitstream_loaded", + label: "bitstream loaded", + satisfied: launcherStatus.bitstream_loaded, + evidence: launcherStatus.bitstream_uuid || "launcher status reports no bitstream UUID", + }), + Object.freeze({ + itemId: "axi_reachable", + label: "AXI reachable", + satisfied: axiPresent, + evidence: hexOrUnavailable(launcherStatus.axi_stat_register_value), + }), + Object.freeze({ + itemId: "manifest_available", + label: "manifest available", + satisfied: traceManifest.frame_count >= 0, + evidence: `${traceManifest.schema_version}; ${traceManifest.frame_count} frame(s)`, + }), + ]), + }); +} + +export function createKv260StatusPanel(inputs = {}) { + const launcherStatus = LauncherStatusReader.consume(inputs.launcherStatus); + const traceManifest = LabTraceReader.consume(inputs.traceManifest); + return Object.freeze({ + version: KV260_STATUS_PANEL_VERSION, + kind: "kv260-status-panel", + source: Object.freeze({ + launcherTypeMirror: LAUNCHER_NPU_STATUS_MIRROR_VERSION, + labManifestParser: "real", + adapterOutput: true, + executesLauncher: false, + executesPccxLab: false, + rawManifestParsedByUi: false, + }), + launcher: Object.freeze(launcherStatus), + lab: Object.freeze({ + schemaVersion: traceManifest.schema_version, + bitstreamUuid: traceManifest.bitstream_uuid, + axiBase: traceManifest.axi_base, + isaVersion: traceManifest.isa_version, + frameCount: traceManifest.frame_count, + sourceKind: traceManifest.source_kind, + runbookRef: traceManifest.runbook_ref, + }), + preflight: createPreflightProposal(launcherStatus, traceManifest), + safety: Object.freeze({ + dataOnly: true, + readOnly: true, + localOnly: true, + launcherExecution: false, + pccxLabExecution: false, + shellExecution: false, + sshExecution: false, + kv260Control: false, + providerCalls: false, + networkCalls: false, + telemetry: false, + automaticUpload: false, + writeBack: false, + }), + }); +} + +export function kv260StatusPanelJson(inputs = {}) { + return `${JSON.stringify(createKv260StatusPanel(inputs), null, 2)}\n`; +} + +export function formatKv260StatusPanel(panel = createKv260StatusPanel()) { + const lines = [ + "KV260 Status Surface", + `version: ${panel.version}`, + `launcherMirror: ${panel.source.launcherTypeMirror}`, + `launcher.bitstreamLoaded: ${yesNo(panel.launcher.bitstream_loaded)}`, + `launcher.bitstreamUuid: ${panel.launcher.bitstream_uuid || "unavailable"}`, + `launcher.axiBaseAddr: ${hexOrUnavailable(panel.launcher.axi_base_addr)}`, + `launcher.axiStatus: ${hexOrUnavailable(panel.launcher.axi_stat_register_value)}`, + `launcher.lastError: ${panel.launcher.last_error || "none"}`, + `lab.schema: ${panel.lab.schemaVersion}`, + `lab.sourceKind: ${panel.lab.sourceKind}`, + `lab.frames: ${panel.lab.frameCount}`, + `lab.axiBase: ${panel.lab.axiBase}`, + `lab.isaVersion: ${panel.lab.isaVersion}`, + "preflight:", + ]; + for (const item of panel.preflight.items) { + const state = item.satisfied ? "pass" : "blocked"; + lines.push(`- ${item.label}: ${state} (${item.evidence})`); + } + lines.push("execution: no launcher, no pccx-lab, no shell, no SSH, no KV260 control"); + lines.push("writeBack: no"); + return `${lines.join("\n")}\n`; +} diff --git a/editors/vscode-prototype/test/extension-entrypoint.test.mjs b/editors/vscode-prototype/test/extension-entrypoint.test.mjs index d6e770b..2825131 100644 --- a/editors/vscode-prototype/test/extension-entrypoint.test.mjs +++ b/editors/vscode-prototype/test/extension-entrypoint.test.mjs @@ -17,6 +17,7 @@ import { PCCX_LAB_BACKEND_STATUS_COMMAND, SHOW_CONTEXT_BUNDLE_AUDIT_COMMAND, SHOW_DIAGNOSTICS_HANDOFF_SUMMARY_COMMAND, + SHOW_KV260_STATUS_PANEL_COMMAND, SHOW_LOCAL_WORKFLOW_STATUS_COMMAND, SHOW_PATCH_PROPOSAL_PREVIEW_COMMAND, SHOW_RECENT_VALIDATION_RESULTS_COMMAND, @@ -1548,6 +1549,23 @@ async function testPccxLabBackendStatusCommandReturnsStatusOnly() { assert.ok(result.status.futureSafetyRequirements.includes("fixed args")); } +async function testKv260StatusPanelCommandReturnsDataOnlySurface() { + const handler = createCommandHandler(SHOW_KV260_STATUS_PANEL_COMMAND, null, {}); + + const result = await handler(); + + assert.equal(result.ok, true); + assert.equal(result.commandId, SHOW_KV260_STATUS_PANEL_COMMAND); + assert.equal(result.kind, "kv260-status-panel"); + assert.equal(result.panel.kind, "kv260-status-panel"); + assert.equal(result.panel.source.launcherTypeMirror, "pccx.ide.launcher-npu-status.local-mirror.v0"); + assert.equal(result.panel.source.labManifestParser, "real"); + assert.equal(result.panel.safety.launcherExecution, false); + assert.equal(result.panel.safety.pccxLabExecution, false); + assert.equal(result.panel.safety.shellExecution, false); + assert.equal(result.panel.safety.sshExecution, false); +} + testKnownFacadeArgs(); testUnknownCommandsRejected(); testCheckedExampleDiagnosticsCommandStaysExampleMode(); @@ -1576,5 +1594,6 @@ await testLocalWorkflowStatusCommandReturnsFixtureOnlyBoundaryState(); await testContextBundleAuditCommandReturnsBoundedAudit(); await testDiagnosticsHandoffSummaryCommandReturnsDataOnlySurface(); await testPccxLabBackendStatusCommandReturnsStatusOnly(); +await testKv260StatusPanelCommandReturnsDataOnlySurface(); console.log("vscode extension entrypoint tests ok"); diff --git a/editors/vscode-prototype/test/extension-host/smoke-suite.cjs b/editors/vscode-prototype/test/extension-host/smoke-suite.cjs index 878bd78..e0a646b 100644 --- a/editors/vscode-prototype/test/extension-host/smoke-suite.cjs +++ b/editors/vscode-prototype/test/extension-host/smoke-suite.cjs @@ -81,6 +81,7 @@ async function run() { "pccxSystemVerilog.showContextBundleAudit", "pccxSystemVerilog.showPccxLabBackendStatus", "pccxSystemVerilog.showDiagnosticsHandoffSummary", + "pccxSystemVerilog.showKv260StatusPanel", ]); const manifest = readManifest(); @@ -595,6 +596,22 @@ async function run() { assert.equal(diagnosticsHandoffStatus.surface.safety.mcpCalls, false); assert.equal(diagnosticsHandoffStatus.surface.safety.lspImplemented, false); + const kv260Status = await vscode.commands.executeCommand( + "pccxSystemVerilog.showKv260StatusPanel", + ); + assert.equal(kv260Status.ok, true); + assert.equal(kv260Status.kind, "kv260-status-panel"); + assert.equal(kv260Status.panel.kind, "kv260-status-panel"); + assert.equal( + kv260Status.panel.source.launcherTypeMirror, + "pccx.ide.launcher-npu-status.local-mirror.v0", + ); + assert.equal(kv260Status.panel.source.labManifestParser, "real"); + assert.equal(kv260Status.panel.safety.launcherExecution, false); + assert.equal(kv260Status.panel.safety.pccxLabExecution, false); + assert.equal(kv260Status.panel.safety.shellExecution, false); + assert.equal(kv260Status.panel.safety.sshExecution, false); + const extensionModule = await importExtensionEntrypoint(); assert.equal(typeof extensionModule.deactivate, "function"); extensionModule.deactivate(); diff --git a/editors/vscode-prototype/test/extension-manifest.test.mjs b/editors/vscode-prototype/test/extension-manifest.test.mjs index 7959d4c..6304074 100644 --- a/editors/vscode-prototype/test/extension-manifest.test.mjs +++ b/editors/vscode-prototype/test/extension-manifest.test.mjs @@ -31,6 +31,7 @@ const COMMAND_IDS = [ "pccxSystemVerilog.showContextBundleAudit", "pccxSystemVerilog.showPccxLabBackendStatus", "pccxSystemVerilog.showDiagnosticsHandoffSummary", + "pccxSystemVerilog.showKv260StatusPanel", ]; async function readText(path) { @@ -139,6 +140,7 @@ async function testDocsKeepExperimentalScope() { assert.match(combined, /allowlisted proposal IDs/i); assert.match(combined, /pccx-lab backend status/i); assert.match(combined, /showDiagnosticsHandoffSummary/); + assert.match(combined, /KV260 status surface/i); assert.match(combined, /no provider\/runtime calls/i); assert.match(combined, /no MCP server implementation/i); } diff --git a/editors/vscode-prototype/test/kv260-status-panel.test.mjs b/editors/vscode-prototype/test/kv260-status-panel.test.mjs new file mode 100644 index 0000000..6442df5 --- /dev/null +++ b/editors/vscode-prototype/test/kv260-status-panel.test.mjs @@ -0,0 +1,113 @@ +// SPDX-License-Identifier: Apache-2.0 +// Copyright 2026 pccxai + +import assert from "node:assert/strict"; +import { readFile } from "node:fs/promises"; +import { dirname, resolve } from "node:path"; +import { fileURLToPath } from "node:url"; + +import { + KV260_STATUS_PANEL_VERSION, + LAUNCHER_NPU_STATUS_MIRROR_VERSION, + LabTraceReader, + LauncherStatusReader, + createKv260StatusPanel, + formatKv260StatusPanel, + kv260StatusPanelJson, +} from "../src/kv260-status-panel.mjs"; + +const ROOT = resolve(dirname(fileURLToPath(import.meta.url)), "../../.."); +const LAUNCHER_FIXTURE = resolve( + ROOT, + "docs/examples/kv260-status/launcher-npu-status.example.json", +); +const TRACE_FIXTURE = resolve( + ROOT, + "docs/examples/kv260-status/lab-trace-manifest.example.json", +); + +async function readJson(path) { + return JSON.parse(await readFile(path, "utf8")); +} + +async function testReadersParseFixtures() { + const launcher = LauncherStatusReader.consume(await readJson(LAUNCHER_FIXTURE)); + const trace = LabTraceReader.consume(await readJson(TRACE_FIXTURE)); + + assert.equal(launcher.bitstream_loaded, false); + assert.equal(launcher.bitstream_uuid, null); + assert.equal(trace.schema_version, "pccx.lab.kv260.trace-manifest.v0"); + assert.equal(trace.frame_count, 1); + assert.equal(trace.frames[0].result_payload, "a55a0001"); +} + +async function testPanelRendersPreflightAndSafety() { + const panel = createKv260StatusPanel({ + launcherStatus: await readJson(LAUNCHER_FIXTURE), + traceManifest: await readJson(TRACE_FIXTURE), + }); + const text = formatKv260StatusPanel(panel); + const jsonText = kv260StatusPanelJson({ + launcherStatus: await readJson(LAUNCHER_FIXTURE), + traceManifest: await readJson(TRACE_FIXTURE), + }); + + assert.equal(panel.version, KV260_STATUS_PANEL_VERSION); + assert.equal(panel.source.launcherTypeMirror, LAUNCHER_NPU_STATUS_MIRROR_VERSION); + assert.equal(panel.source.labManifestParser, "real"); + assert.equal(panel.safety.readOnly, true); + assert.equal(panel.safety.launcherExecution, false); + assert.equal(panel.safety.pccxLabExecution, false); + assert.equal(panel.safety.sshExecution, false); + assert.deepEqual(panel.preflight.items.map((item) => item.itemId), [ + "bitstream_loaded", + "axi_reachable", + "manifest_available", + ]); + assert.match(text, /KV260 Status Surface/); + assert.match(text, /launcherMirror: pccx\.ide\.launcher-npu-status\.local-mirror\.v0/); + assert.match(text, /execution: no launcher, no pccx-lab, no shell, no SSH, no KV260 control/); + assert.equal(JSON.parse(jsonText).kind, "kv260-status-panel"); +} + +async function testRejectsInvalidOrUnsafeInputs() { + assert.throws( + () => LauncherStatusReader.consume({ bitstream_loaded: "false" }), + /bitstream_loaded/, + ); + + const invalid = await readJson(TRACE_FIXTURE); + invalid.source_kind = "ssh_log_tail"; + assert.throws( + () => LabTraceReader.consume(invalid), + /source_kind: must be file_replay/, + ); + + const unsafe = await readJson(LAUNCHER_FIXTURE); + unsafe.last_error = "/home/user/private.log"; + assert.throws( + () => LauncherStatusReader.consume(unsafe), + /private paths/, + ); +} + +async function testModuleSourceHasNoExecutionTerms() { + const source = await readFile( + resolve(ROOT, "editors/vscode-prototype/src/kv260-status-panel.mjs"), + "utf8", + ); + + assert.doesNotMatch(source, /node:child_process|\bexecFile\b|\bspawn\s*\(|\bexec\s*\(/); + assert.doesNotMatch(source, /node:fs|readFile\s*\(|readdir\s*\(|opendir\s*\(|stat\s*\(/); + assert.doesNotMatch(source, /\bwriteFile\s*\(|\bappendFile\s*\(|\bunlink\s*\(|\brm\s*\(/); + assert.doesNotMatch(source, /\bfetch\s*\(|XMLHttpRequest|WebSocket|node:https|node:http|node:net|node:tls/); + assert.doesNotMatch(source, /\b(?:openai|anthropic|gemini)\b/i); + assert.doesNotMatch(source, /modelcontextprotocol|McpServer|vscode-languageclient|LanguageClient/); +} + +await testReadersParseFixtures(); +await testPanelRendersPreflightAndSafety(); +await testRejectsInvalidOrUnsafeInputs(); +await testModuleSourceHasNoExecutionTerms(); + +console.log("vscode kv260 status panel tests ok"); diff --git a/editors/vscode-prototype/test/static-boundary.test.mjs b/editors/vscode-prototype/test/static-boundary.test.mjs index 2cf09a3..8211a79 100644 --- a/editors/vscode-prototype/test/static-boundary.test.mjs +++ b/editors/vscode-prototype/test/static-boundary.test.mjs @@ -151,6 +151,7 @@ async function testProposalAndStatusModulesAreDataOnly() { resolve(EXTENSION_ROOT, "src/launcher-status-contract.mjs"), resolve(EXTENSION_ROOT, "src/diagnostics-handoff-consumer.mjs"), resolve(EXTENSION_ROOT, "src/diagnostics-handoff-status-surface.mjs"), + resolve(EXTENSION_ROOT, "src/kv260-status-panel.mjs"), resolve(EXTENSION_ROOT, "src/runtime-readiness-consumer.mjs"), resolve(EXTENSION_ROOT, "src/runtime-readiness-status-surface.mjs"), resolve(EXTENSION_ROOT, "src/local-workflow-status.mjs"), diff --git a/pyproject.toml b/pyproject.toml index aa60aea..3548bbe 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -22,6 +22,7 @@ test = ["pytest"] [project.scripts] pccx-ide = "pccx_ide_cli.__main__:main" +sv-ide = "pccx_ide_cli.__main__:main" [tool.hatch.build.targets.wheel] packages = ["src/pccx_ide_cli"] diff --git a/src/pccx_ide_cli/cli.py b/src/pccx_ide_cli/cli.py index 25d3c38..f953d2f 100644 --- a/src/pccx_ide_cli/cli.py +++ b/src/pccx_ide_cli/cli.py @@ -1127,6 +1127,29 @@ def _build_parser() -> argparse.ArgumentParser: help="Output format (default: json).", ) + kv260_status = sub.add_parser( + "kv260-status", + help="Render a read-only KV260 launcher/lab status surface.", + ) + kv260_status.add_argument( + "--launcher-status", + type=Path, + default=None, + help="Path to launcher NPUStatus JSON; defaults to the bundled fixture.", + ) + kv260_status.add_argument( + "--trace-manifest", + type=Path, + default=None, + help="Path to lab TraceManifest JSON; defaults to the bundled fixture.", + ) + kv260_status.add_argument( + "--format", + choices=("text", "json"), + default="text", + help="Output format (default: text).", + ) + xsim_log = sub.add_parser( "xsim-log", help="Parse an existing xsim-style log file into diagnostics-like output.", @@ -1203,6 +1226,35 @@ def main(argv: Sequence[str] | None = None) -> int: sys.stdout.write(schema_path.read_text(encoding="utf-8")) return 0 + if args.command == "kv260-status": + from .kv260_status import ( + LabTraceReader, + LauncherStatusReader, + create_kv260_status_panel, + default_launcher_status_path, + default_trace_manifest_path, + format_kv260_status_panel, + panel_to_dict, + ) + + repo_root = Path(__file__).resolve().parent.parent.parent + launcher_path = args.launcher_status or default_launcher_status_path(repo_root) + trace_path = args.trace_manifest or default_trace_manifest_path(repo_root) + try: + launcher_status = LauncherStatusReader.from_json_file(launcher_path) + trace_manifest = LabTraceReader.from_json_file(trace_path) + panel = create_kv260_status_panel(launcher_status, trace_manifest) + except (OSError, ValueError, json.JSONDecodeError) as error: + sys.stderr.write(f"error: cannot render kv260 status: {error}\n") + return 2 + + if args.format == "json": + json.dump(panel_to_dict(panel), sys.stdout, indent=2, sort_keys=True) + sys.stdout.write("\n") + else: + sys.stdout.write(format_kv260_status_panel(panel)) + return 0 + if args.command == "check": if args.backend == "pccx-lab": from .pccx_lab_backend import run as _run_pccx_lab diff --git a/src/pccx_ide_cli/kv260_status.py b/src/pccx_ide_cli/kv260_status.py new file mode 100644 index 0000000..146cc3f --- /dev/null +++ b/src/pccx_ide_cli/kv260_status.py @@ -0,0 +1,331 @@ +# SPDX-License-Identifier: Apache-2.0 +# Copyright 2026 pccxai + +from __future__ import annotations + +import json +from dataclasses import dataclass +from pathlib import Path +from typing import Any, Mapping + + +LAUNCHER_NPU_STATUS_MIRROR_VERSION = "pccx.ide.launcher-npu-status.local-mirror.v0" +TRACE_MANIFEST_SCHEMA_VERSION = "pccx.lab.kv260.trace-manifest.v0" +KV260_STATUS_PANEL_VERSION = "pccx.ide.kv260-status-panel.v0" + + +@dataclass(frozen=True) +class NPUStatus: + bitstream_loaded: bool + bitstream_uuid: str | None + axi_base_addr: int | None + axi_stat_register_value: int | None + last_error: str | None + + +@dataclass(frozen=True) +class TraceFrame: + frame_idx: int + axi_stat_register_value: int + engine_completion_mask: int + cycle_count: int + result_payload: str | None + error: Mapping[str, Any] | None + + +@dataclass(frozen=True) +class TraceManifest: + schema_version: str + bitstream_uuid: str + axi_base: str + isa_version: str + frame_count: int + checksums: tuple[Mapping[str, Any], ...] + runbook_ref: str + source_kind: str + frames: tuple[TraceFrame, ...] + + +@dataclass(frozen=True) +class PreflightItem: + item_id: str + label: str + satisfied: bool + evidence: str + + +@dataclass(frozen=True) +class PreflightProposal: + kind: str + items: tuple[PreflightItem, ...] + + +@dataclass(frozen=True) +class Kv260StatusPanel: + version: str + launcher_status: NPUStatus + trace_manifest: TraceManifest + preflight: PreflightProposal + + +class LauncherStatusReader: + """Read-only local mirror of launcher#70 `NPUStatus` type shape.""" + + @classmethod + def from_json_text(cls, text: str) -> NPUStatus: + return cls.from_object(json.loads(text)) + + @classmethod + def from_json_file(cls, path: Path) -> NPUStatus: + return cls.from_json_text(path.read_text(encoding="utf-8")) + + @staticmethod + def from_object(value: Mapping[str, Any]) -> NPUStatus: + if not isinstance(value, Mapping): + raise ValueError("launcher NPU status must be a JSON object") + bitstream_loaded = _bool_field(value, "bitstream_loaded") + bitstream_uuid = _optional_string_field(value, "bitstream_uuid") + axi_base_addr = _optional_int_field(value, "axi_base_addr") + axi_stat_register_value = _optional_int_field(value, "axi_stat_register_value") + last_error = _optional_string_field(value, "last_error") + return NPUStatus( + bitstream_loaded=bitstream_loaded, + bitstream_uuid=bitstream_uuid, + axi_base_addr=axi_base_addr, + axi_stat_register_value=axi_stat_register_value, + last_error=last_error, + ) + + +class LabTraceReader: + """Read-only parser for lab#160 `TraceManifest` JSON fixtures.""" + + @classmethod + def from_json_text(cls, text: str) -> TraceManifest: + return cls.from_object(json.loads(text)) + + @classmethod + def from_json_file(cls, path: Path) -> TraceManifest: + return cls.from_json_text(path.read_text(encoding="utf-8")) + + @staticmethod + def from_object(value: Mapping[str, Any]) -> TraceManifest: + if not isinstance(value, Mapping): + raise ValueError("trace manifest must be a JSON object") + schema_version = _string_field(value, "schema_version") + if schema_version != TRACE_MANIFEST_SCHEMA_VERSION: + raise ValueError(f"unsupported trace manifest schema_version: {schema_version}") + source_kind = _string_field(value, "source_kind") + if source_kind != "file_replay": + raise ValueError("only file_replay trace manifests are displayable") + frames = tuple(_trace_frame(frame, index) for index, frame in enumerate(_list_field(value, "frames"))) + frame_count = _int_field(value, "frame_count") + if frame_count != len(frames): + raise ValueError("trace manifest frame_count must match inline frames") + return TraceManifest( + schema_version=schema_version, + bitstream_uuid=_string_field(value, "bitstream_uuid"), + axi_base=_string_field(value, "axi_base"), + isa_version=_string_field(value, "isa_version"), + frame_count=frame_count, + checksums=tuple(_mapping_item(item, "checksums", index) for index, item in enumerate(_list_field(value, "checksums"))), + runbook_ref=_string_field(value, "runbook_ref"), + source_kind=source_kind, + frames=frames, + ) + + +def create_preflight_proposal(status: NPUStatus, manifest: TraceManifest) -> PreflightProposal: + axi_present = status.axi_base_addr is not None and status.axi_stat_register_value is not None + return PreflightProposal( + kind="kv260-preflight-proposal", + items=( + PreflightItem( + item_id="bitstream_loaded", + label="bitstream loaded", + satisfied=status.bitstream_loaded, + evidence=status.bitstream_uuid or "launcher status reports no bitstream UUID", + ), + PreflightItem( + item_id="axi_reachable", + label="AXI reachable", + satisfied=axi_present, + evidence=_hex_or_unavailable(status.axi_stat_register_value), + ), + PreflightItem( + item_id="manifest_available", + label="manifest available", + satisfied=manifest.frame_count >= 0, + evidence=f"{manifest.schema_version}; {manifest.frame_count} frame(s)", + ), + ), + ) + + +def create_kv260_status_panel(status: NPUStatus, manifest: TraceManifest) -> Kv260StatusPanel: + return Kv260StatusPanel( + version=KV260_STATUS_PANEL_VERSION, + launcher_status=status, + trace_manifest=manifest, + preflight=create_preflight_proposal(status, manifest), + ) + + +def panel_to_dict(panel: Kv260StatusPanel) -> dict[str, Any]: + return { + "version": panel.version, + "kind": "kv260-status-panel", + "launcher": { + "mirror_version": LAUNCHER_NPU_STATUS_MIRROR_VERSION, + "bitstream_loaded": panel.launcher_status.bitstream_loaded, + "bitstream_uuid": panel.launcher_status.bitstream_uuid, + "axi_base_addr": panel.launcher_status.axi_base_addr, + "axi_stat_register_value": panel.launcher_status.axi_stat_register_value, + "last_error": panel.launcher_status.last_error, + }, + "lab": { + "schema_version": panel.trace_manifest.schema_version, + "bitstream_uuid": panel.trace_manifest.bitstream_uuid, + "axi_base": panel.trace_manifest.axi_base, + "isa_version": panel.trace_manifest.isa_version, + "frame_count": panel.trace_manifest.frame_count, + "source_kind": panel.trace_manifest.source_kind, + "runbook_ref": panel.trace_manifest.runbook_ref, + }, + "preflight": { + "kind": panel.preflight.kind, + "items": [ + { + "item_id": item.item_id, + "label": item.label, + "satisfied": item.satisfied, + "evidence": item.evidence, + } + for item in panel.preflight.items + ], + }, + "safety": { + "read_only": True, + "launcher_execution": False, + "pccx_lab_execution": False, + "shell_execution": False, + "ssh_execution": False, + "kv260_control": False, + "write_back": False, + }, + } + + +def format_kv260_status_panel(panel: Kv260StatusPanel) -> str: + status = panel.launcher_status + manifest = panel.trace_manifest + lines = [ + "KV260 Status Surface", + f"version: {panel.version}", + f"launcherMirror: {LAUNCHER_NPU_STATUS_MIRROR_VERSION}", + f"launcher.bitstreamLoaded: {_yes_no(status.bitstream_loaded)}", + f"launcher.bitstreamUuid: {status.bitstream_uuid or 'unavailable'}", + f"launcher.axiBaseAddr: {_hex_or_unavailable(status.axi_base_addr)}", + f"launcher.axiStatus: {_hex_or_unavailable(status.axi_stat_register_value)}", + f"launcher.lastError: {status.last_error or 'none'}", + f"lab.schema: {manifest.schema_version}", + f"lab.sourceKind: {manifest.source_kind}", + f"lab.frames: {manifest.frame_count}", + f"lab.axiBase: {manifest.axi_base}", + f"lab.isaVersion: {manifest.isa_version}", + "preflight:", + ] + for item in panel.preflight.items: + state = "pass" if item.satisfied else "blocked" + lines.append(f"- {item.label}: {state} ({item.evidence})") + lines.extend([ + "execution: no launcher, no pccx-lab, no shell, no SSH, no KV260 control", + "writeBack: no", + ]) + return "\n".join(lines) + "\n" + + +def default_launcher_status_path(repo_root: Path) -> Path: + return repo_root / "docs" / "examples" / "kv260-status" / "launcher-npu-status.example.json" + + +def default_trace_manifest_path(repo_root: Path) -> Path: + return repo_root / "docs" / "examples" / "kv260-status" / "lab-trace-manifest.example.json" + + +def _bool_field(value: Mapping[str, Any], key: str) -> bool: + field = value.get(key) + if not isinstance(field, bool): + raise ValueError(f"{key} must be a boolean") + return field + + +def _int_field(value: Mapping[str, Any], key: str) -> int: + field = value.get(key) + if not isinstance(field, int) or isinstance(field, bool) or field < 0: + raise ValueError(f"{key} must be a non-negative integer") + return field + + +def _optional_int_field(value: Mapping[str, Any], key: str) -> int | None: + field = value.get(key) + if field is None: + return None + if not isinstance(field, int) or isinstance(field, bool) or field < 0: + raise ValueError(f"{key} must be null or a non-negative integer") + return field + + +def _string_field(value: Mapping[str, Any], key: str) -> str: + field = value.get(key) + if not isinstance(field, str) or not field.strip() or "\n" in field or "\r" in field: + raise ValueError(f"{key} must be a non-empty single-line string") + return field + + +def _optional_string_field(value: Mapping[str, Any], key: str) -> str | None: + field = value.get(key) + if field is None: + return None + if not isinstance(field, str) or "\n" in field or "\r" in field: + raise ValueError(f"{key} must be null or a single-line string") + return field + + +def _list_field(value: Mapping[str, Any], key: str) -> list[Any]: + field = value.get(key) + if not isinstance(field, list): + raise ValueError(f"{key} must be an array") + return field + + +def _mapping_item(value: Any, key: str, index: int) -> Mapping[str, Any]: + if not isinstance(value, Mapping): + raise ValueError(f"{key}[{index}] must be an object") + return value + + +def _trace_frame(value: Any, index: int) -> TraceFrame: + item = _mapping_item(value, "frames", index) + result_payload = item.get("result_payload") + if result_payload is not None and not isinstance(result_payload, str): + raise ValueError(f"frames[{index}].result_payload must be null or a string") + error = item.get("error") + if error is not None and not isinstance(error, Mapping): + raise ValueError(f"frames[{index}].error must be null or an object") + return TraceFrame( + frame_idx=_int_field(item, "frame_idx"), + axi_stat_register_value=_int_field(item, "axi_stat_register_value"), + engine_completion_mask=_int_field(item, "engine_completion_mask"), + cycle_count=_int_field(item, "cycle_count"), + result_payload=result_payload, + error=error, + ) + + +def _hex_or_unavailable(value: int | None) -> str: + return "unavailable" if value is None else f"0x{value:x}" + + +def _yes_no(value: bool) -> str: + return "yes" if value else "no" diff --git a/tests/test_kv260_status.py b/tests/test_kv260_status.py new file mode 100644 index 0000000..995d59f --- /dev/null +++ b/tests/test_kv260_status.py @@ -0,0 +1,122 @@ +# SPDX-License-Identifier: Apache-2.0 +# Copyright 2026 pccxai + +from __future__ import annotations + +import json +import os +import subprocess +import sys +from pathlib import Path + +import pytest + +REPO_ROOT = Path(__file__).resolve().parent.parent +SRC = REPO_ROOT / "src" +sys.path.insert(0, str(SRC)) + +from pccx_ide_cli.kv260_status import ( + LabTraceReader, + LauncherStatusReader, + create_kv260_status_panel, + format_kv260_status_panel, + panel_to_dict, +) + + +LAUNCHER_FIXTURE = REPO_ROOT / "docs/examples/kv260-status/launcher-npu-status.example.json" +TRACE_FIXTURE = REPO_ROOT / "docs/examples/kv260-status/lab-trace-manifest.example.json" + + +def _run_cli(*args: str) -> subprocess.CompletedProcess: + env = { + "PYTHONPATH": str(SRC), + "PATH": os.environ.get("PATH", ""), + } + return subprocess.run( + [sys.executable, "-m", "pccx_ide_cli", *args], + cwd=REPO_ROOT, + env=env, + capture_output=True, + text=True, + check=False, + ) + + +def test_readers_parse_tiny_fixtures_as_data_only() -> None: + launcher = LauncherStatusReader.from_json_file(LAUNCHER_FIXTURE) + manifest = LabTraceReader.from_json_file(TRACE_FIXTURE) + + assert launcher.bitstream_loaded is False + assert launcher.bitstream_uuid is None + assert manifest.schema_version == "pccx.lab.kv260.trace-manifest.v0" + assert manifest.frame_count == 1 + assert manifest.frames[0].result_payload == "a55a0001" + + +def test_status_panel_and_preflight_are_deterministic() -> None: + launcher = LauncherStatusReader.from_json_file(LAUNCHER_FIXTURE) + manifest = LabTraceReader.from_json_file(TRACE_FIXTURE) + panel = create_kv260_status_panel(launcher, manifest) + payload = panel_to_dict(panel) + text = format_kv260_status_panel(panel) + + assert payload["kind"] == "kv260-status-panel" + assert payload["launcher"]["mirror_version"] == "pccx.ide.launcher-npu-status.local-mirror.v0" + assert payload["safety"]["launcher_execution"] is False + assert payload["safety"]["pccx_lab_execution"] is False + assert payload["safety"]["ssh_execution"] is False + assert [item["item_id"] for item in payload["preflight"]["items"]] == [ + "bitstream_loaded", + "axi_reachable", + "manifest_available", + ] + assert "KV260 Status Surface" in text + assert "execution: no launcher, no pccx-lab, no shell, no SSH, no KV260 control" in text + + +def test_cli_kv260_status_prints_text_and_json() -> None: + text_result = _run_cli("kv260-status") + assert text_result.returncode == 0, text_result.stderr + assert "KV260 Status Surface" in text_result.stdout + assert "launcherMirror: pccx.ide.launcher-npu-status.local-mirror.v0" in text_result.stdout + assert "preflight:" in text_result.stdout + + json_result = _run_cli("kv260-status", "--format", "json") + assert json_result.returncode == 0, json_result.stderr + payload = json.loads(json_result.stdout) + assert payload["kind"] == "kv260-status-panel" + assert payload["lab"]["source_kind"] == "file_replay" + + +def test_readers_reject_unsafe_or_invalid_shapes() -> None: + with pytest.raises(ValueError, match="bitstream_loaded"): + LauncherStatusReader.from_object({"bitstream_loaded": "false"}) + + invalid_manifest = json.loads(TRACE_FIXTURE.read_text(encoding="utf-8")) + invalid_manifest["source_kind"] = "ssh_log_tail" + with pytest.raises(ValueError, match="file_replay"): + LabTraceReader.from_object(invalid_manifest) + + +def test_touched_sources_do_not_add_execution_paths() -> None: + source = (SRC / "pccx_ide_cli" / "kv260_status.py").read_text(encoding="utf-8") + combined = "\n".join([ + source, + (REPO_ROOT / "docs/KV260_READ_ONLY_STATUS_SURFACE.md").read_text(encoding="utf-8"), + ]) + forbidden_source_terms = [ + "subprocess", + "os.system", + "popen", + "socket", + "paramiko", + "requests", + "urllib", + "write_text", + "open(", + ] + for term in forbidden_source_terms: + assert term not in source, term + assert "/home/" not in combined + assert "password:" not in combined.lower() From 790dc8d812280ed4866c1af89fe86cc7df5f1950 Mon Sep 17 00:00:00 2001 From: hkimw <54717101+hkimw@users.noreply.github.com> Date: Wed, 6 May 2026 22:44:56 +0900 Subject: [PATCH 2/2] feat(kv260): wire preflight to serial probe --- docs/KV260_READ_ONLY_STATUS_SURFACE.md | 15 +- .../launcher-npu-status.example.json | 11 +- .../src/kv260-status-panel.mjs | 149 ++++++++++++-- .../test/kv260-status-panel.test.mjs | 61 +++++- src/pccx_ide_cli/kv260_status.py | 183 ++++++++++++++++-- tests/test_kv260_status.py | 60 +++++- 6 files changed, 432 insertions(+), 47 deletions(-) diff --git a/docs/KV260_READ_ONLY_STATUS_SURFACE.md b/docs/KV260_READ_ONLY_STATUS_SURFACE.md index 025cf8a..55b0a57 100644 --- a/docs/KV260_READ_ONLY_STATUS_SURFACE.md +++ b/docs/KV260_READ_ONLY_STATUS_SURFACE.md @@ -11,6 +11,12 @@ write back status. - `LauncherStatusReader` consumes the launcher `NPUStatus` shape from `pccxai/pccx-llm-launcher#70`: `bitstream_loaded`, `bitstream_uuid`, `axi_base_addr`, `axi_stat_register_value`, and `last_error`. +- The same reader accepts the launcher serial preflight snapshot from + `pccxai/pccx-llm-launcher#72`: selected tty port, login result, truncated + kernel uname display, XRT presence, and last preflight timestamp. When the + snapshot is absent, blocked, or has no tty/login data, the panel reports + `preflight not run` instead of reading environment variables or opening a + port. - `LabTraceReader` parses the lab `TraceManifest` JSON shape from `pccxai/pccx-lab#160` for file-replay trace manifests. @@ -22,12 +28,15 @@ the launcher contract package. `Kv260StatusPanel` renders launcher status, lab manifest metadata, and a `PreflightProposal` checklist: -- bitstream loaded -- AXI reachable -- manifest available +- serial tty port +- serial login +- XRT present +- serial preflight timestamp The checklist is display-only. A blocked item is evidence that the IDE should keep any future KV260 run path gated until lower layers provide reviewed data. +The IDE surface does not run the launcher serial backend, SSH, shell commands, +or board commands; it only renders JSON already produced by the launcher side. ## CLI diff --git a/docs/examples/kv260-status/launcher-npu-status.example.json b/docs/examples/kv260-status/launcher-npu-status.example.json index f093c6d..609102d 100644 --- a/docs/examples/kv260-status/launcher-npu-status.example.json +++ b/docs/examples/kv260-status/launcher-npu-status.example.json @@ -3,5 +3,14 @@ "bitstream_uuid": null, "axi_base_addr": null, "axi_stat_register_value": null, - "last_error": "fixture only; lower-layer evidence not supplied" + "last_error": "fixture only; lower-layer evidence not supplied", + "serial_probe": { + "schema_version": "pccx.launcher.kv260-serial-preflight.v0", + "status": "not_run", + "tty_port": null, + "login_ok": null, + "kernel_uname": null, + "xrt_present": null, + "last_preflight_at": null + } } diff --git a/editors/vscode-prototype/src/kv260-status-panel.mjs b/editors/vscode-prototype/src/kv260-status-panel.mjs index 703f576..b0c2f4c 100644 --- a/editors/vscode-prototype/src/kv260-status-panel.mjs +++ b/editors/vscode-prototype/src/kv260-status-panel.mjs @@ -3,15 +3,28 @@ export const LAUNCHER_NPU_STATUS_MIRROR_VERSION = "pccx.ide.launcher-npu-status.local-mirror.v0"; +export const LAUNCHER_SERIAL_PREFLIGHT_STATUS_VERSION = + "pccx.launcher.kv260-serial-preflight.v0"; export const TRACE_MANIFEST_SCHEMA_VERSION = "pccx.lab.kv260.trace-manifest.v0"; export const KV260_STATUS_PANEL_VERSION = "pccx.ide.kv260-status-panel.v0"; +export const DEFAULT_SERIAL_PREFLIGHT_STATUS = Object.freeze({ + schema_version: LAUNCHER_SERIAL_PREFLIGHT_STATUS_VERSION, + status: "not_run", + tty_port: null, + login_ok: null, + kernel_uname: null, + xrt_present: null, + last_preflight_at: null, +}); + export const DEFAULT_LAUNCHER_NPU_STATUS = Object.freeze({ bitstream_loaded: false, bitstream_uuid: null, axi_base_addr: null, axi_stat_register_value: null, last_error: "fixture only; lower-layer evidence not supplied", + serial_probe: DEFAULT_SERIAL_PREFLIGHT_STATUS, }); export const DEFAULT_TRACE_MANIFEST = Object.freeze({ @@ -122,6 +135,13 @@ function boolField(value, path, errors) { return value; } +function optionalBoolField(value, path, errors) { + if (value == null) { + return null; + } + return boolField(value, path, errors); +} + function optionalIntegerField(value, path, errors) { if (value == null) { return null; @@ -141,6 +161,15 @@ function integerField(value, path, errors) { return value; } +function firstPresent(value, keys) { + for (const key of keys) { + if (Object.hasOwn(value, key)) { + return value[key]; + } + } + return undefined; +} + function arrayField(value, path, errors) { if (!Array.isArray(value)) { addError(errors, path, "must be an array"); @@ -149,6 +178,45 @@ function arrayField(value, path, errors) { return value; } +function normalizeSerialPreflightStatus(value, errors) { + const probe = value ?? DEFAULT_SERIAL_PREFLIGHT_STATUS; + if (!isObject(probe)) { + addError(errors, "serial_probe", "must be an object"); + return { ...DEFAULT_SERIAL_PREFLIGHT_STATUS }; + } + const schemaVersion = probe.schema_version ?? LAUNCHER_SERIAL_PREFLIGHT_STATUS_VERSION; + if (schemaVersion !== LAUNCHER_SERIAL_PREFLIGHT_STATUS_VERSION) { + addError( + errors, + "serial_probe.schema_version", + `must be ${LAUNCHER_SERIAL_PREFLIGHT_STATUS_VERSION}`, + ); + } + const status = probe.status ?? "not_run"; + if (!["available", "blocked", "not_run"].includes(status)) { + addError(errors, "serial_probe.status", "must be available, blocked, or not_run"); + } + return { + schema_version: schemaVersion, + status, + tty_port: optionalStringField(probe.tty_port, "serial_probe.tty_port", errors, 240), + login_ok: optionalBoolField(probe.login_ok, "serial_probe.login_ok", errors), + kernel_uname: optionalStringField( + probe.kernel_uname, + "serial_probe.kernel_uname", + errors, + 500, + ), + xrt_present: optionalBoolField(probe.xrt_present, "serial_probe.xrt_present", errors), + last_preflight_at: optionalStringField( + firstPresent(probe, ["last_preflight_at", "checked_at"]), + "serial_probe.last_preflight_at", + errors, + 120, + ), + }; +} + function normalizeTraceFrame(value, index, errors) { const frame = isObject(value) ? value : {}; if (!isObject(value)) { @@ -196,6 +264,7 @@ export class LauncherStatusReader { errors, ), last_error: optionalStringField(status.last_error, "last_error", errors, 500), + serial_probe: normalizeSerialPreflightStatus(status.serial_probe, errors), }; if (errors.length > 0) { throw new Error(errors.join("; ")); @@ -251,29 +320,57 @@ function yesNo(value) { return value ? "yes" : "no"; } -export function createPreflightProposal(launcherStatus, traceManifest) { - const axiPresent = - launcherStatus.axi_base_addr != null && launcherStatus.axi_stat_register_value != null; +function probeEvidence(probe, value, unavailable = "preflight not run") { + if (probe.status === "not_run") { + return unavailable; + } + if (value == null || value === "") { + return "unavailable"; + } + return value; +} + +function boolProbeItem(itemId, label, probe, value, trueText, falseText) { + const notRun = probe.status === "not_run"; + return Object.freeze({ + itemId, + label, + state: notRun ? "not_run" : value ? "pass" : "blocked", + satisfied: value === true, + evidence: notRun ? "preflight not run" : value ? trueText : falseText, + }); +} + +function truncateText(value, maxCharacters = 96) { + if (!value) { + return "preflight not run"; + } + if (value.length <= maxCharacters) { + return value; + } + return `${value.slice(0, maxCharacters - 3)}...`; +} + +export function createPreflightProposal(launcherStatus, _traceManifest) { + const probe = launcherStatus.serial_probe; return Object.freeze({ kind: "kv260-preflight-proposal", items: Object.freeze([ Object.freeze({ - itemId: "bitstream_loaded", - label: "bitstream loaded", - satisfied: launcherStatus.bitstream_loaded, - evidence: launcherStatus.bitstream_uuid || "launcher status reports no bitstream UUID", - }), - Object.freeze({ - itemId: "axi_reachable", - label: "AXI reachable", - satisfied: axiPresent, - evidence: hexOrUnavailable(launcherStatus.axi_stat_register_value), + itemId: "serial_tty_port", + label: "serial tty port", + state: probe.status === "not_run" ? "not_run" : probe.tty_port ? "pass" : "blocked", + satisfied: Boolean(probe.tty_port), + evidence: probeEvidence(probe, probe.tty_port), }), + boolProbeItem("serial_login", "serial login", probe, probe.login_ok, "login OK", "login not OK"), + boolProbeItem("serial_xrt", "XRT present", probe, probe.xrt_present, "xrt_present true", "xrt_present false"), Object.freeze({ - itemId: "manifest_available", - label: "manifest available", - satisfied: traceManifest.frame_count >= 0, - evidence: `${traceManifest.schema_version}; ${traceManifest.frame_count} frame(s)`, + itemId: "serial_probe_timestamp", + label: "serial preflight timestamp", + state: probe.status === "not_run" ? "not_run" : probe.last_preflight_at ? "pass" : "blocked", + satisfied: Boolean(probe.last_preflight_at), + evidence: probeEvidence(probe, probe.last_preflight_at), }), ]), }); @@ -294,6 +391,16 @@ export function createKv260StatusPanel(inputs = {}) { rawManifestParsedByUi: false, }), launcher: Object.freeze(launcherStatus), + serialProbe: Object.freeze({ + schemaVersion: launcherStatus.serial_probe.schema_version, + status: launcherStatus.serial_probe.status, + ttyPort: launcherStatus.serial_probe.tty_port, + loginOk: launcherStatus.serial_probe.login_ok, + kernelUname: launcherStatus.serial_probe.kernel_uname, + kernelUnameDisplay: truncateText(launcherStatus.serial_probe.kernel_uname), + xrtPresent: launcherStatus.serial_probe.xrt_present, + lastPreflightAt: launcherStatus.serial_probe.last_preflight_at, + }), lab: Object.freeze({ schemaVersion: traceManifest.schema_version, bitstreamUuid: traceManifest.bitstream_uuid, @@ -336,6 +443,12 @@ export function formatKv260StatusPanel(panel = createKv260StatusPanel()) { `launcher.axiBaseAddr: ${hexOrUnavailable(panel.launcher.axi_base_addr)}`, `launcher.axiStatus: ${hexOrUnavailable(panel.launcher.axi_stat_register_value)}`, `launcher.lastError: ${panel.launcher.last_error || "none"}`, + `serial.ttyPort: ${panel.serialProbe.ttyPort || "preflight not run"}`, + `serial.kernelUname: ${panel.serialProbe.kernelUnameDisplay}`, + `serial.xrtPresent: ${ + panel.serialProbe.xrtPresent == null ? "preflight not run" : yesNo(panel.serialProbe.xrtPresent) + }`, + `serial.lastPreflightAt: ${panel.serialProbe.lastPreflightAt || "preflight not run"}`, `lab.schema: ${panel.lab.schemaVersion}`, `lab.sourceKind: ${panel.lab.sourceKind}`, `lab.frames: ${panel.lab.frameCount}`, @@ -344,7 +457,7 @@ export function formatKv260StatusPanel(panel = createKv260StatusPanel()) { "preflight:", ]; for (const item of panel.preflight.items) { - const state = item.satisfied ? "pass" : "blocked"; + const state = item.state ?? (item.satisfied ? "pass" : "blocked"); lines.push(`- ${item.label}: ${state} (${item.evidence})`); } lines.push("execution: no launcher, no pccx-lab, no shell, no SSH, no KV260 control"); diff --git a/editors/vscode-prototype/test/kv260-status-panel.test.mjs b/editors/vscode-prototype/test/kv260-status-panel.test.mjs index 6442df5..a147287 100644 --- a/editors/vscode-prototype/test/kv260-status-panel.test.mjs +++ b/editors/vscode-prototype/test/kv260-status-panel.test.mjs @@ -8,6 +8,7 @@ import { fileURLToPath } from "node:url"; import { KV260_STATUS_PANEL_VERSION, + LAUNCHER_SERIAL_PREFLIGHT_STATUS_VERSION, LAUNCHER_NPU_STATUS_MIRROR_VERSION, LabTraceReader, LauncherStatusReader, @@ -36,14 +37,25 @@ async function testReadersParseFixtures() { assert.equal(launcher.bitstream_loaded, false); assert.equal(launcher.bitstream_uuid, null); + assert.equal(launcher.serial_probe.status, "not_run"); assert.equal(trace.schema_version, "pccx.lab.kv260.trace-manifest.v0"); assert.equal(trace.frame_count, 1); assert.equal(trace.frames[0].result_payload, "a55a0001"); } async function testPanelRendersPreflightAndSafety() { + const launcherStatus = await readJson(LAUNCHER_FIXTURE); + launcherStatus.serial_probe = { + schema_version: LAUNCHER_SERIAL_PREFLIGHT_STATUS_VERSION, + status: "available", + tty_port: "/dev/ttyUSB0", + login_ok: true, + kernel_uname: "Linux kv260 6.6.0-test #1 SMP PREEMPT aarch64 GNU/Linux", + xrt_present: true, + last_preflight_at: "2026-05-06T09:00:00Z", + }; const panel = createKv260StatusPanel({ - launcherStatus: await readJson(LAUNCHER_FIXTURE), + launcherStatus, traceManifest: await readJson(TRACE_FIXTURE), }); const text = formatKv260StatusPanel(panel); @@ -59,17 +71,56 @@ async function testPanelRendersPreflightAndSafety() { assert.equal(panel.safety.launcherExecution, false); assert.equal(panel.safety.pccxLabExecution, false); assert.equal(panel.safety.sshExecution, false); + assert.equal(panel.serialProbe.ttyPort, "/dev/ttyUSB0"); + assert.equal(panel.serialProbe.xrtPresent, true); + assert.equal(panel.serialProbe.lastPreflightAt, "2026-05-06T09:00:00Z"); assert.deepEqual(panel.preflight.items.map((item) => item.itemId), [ - "bitstream_loaded", - "axi_reachable", - "manifest_available", + "serial_tty_port", + "serial_login", + "serial_xrt", + "serial_probe_timestamp", ]); assert.match(text, /KV260 Status Surface/); + assert.match(text, /serial\.ttyPort: \/dev\/ttyUSB0/); + assert.match(text, /serial\.kernelUname: Linux kv260/); + assert.match(text, /serial\.xrtPresent: yes/); + assert.match(text, /serial\.lastPreflightAt: 2026-05-06T09:00:00Z/); assert.match(text, /launcherMirror: pccx\.ide\.launcher-npu-status\.local-mirror\.v0/); assert.match(text, /execution: no launcher, no pccx-lab, no shell, no SSH, no KV260 control/); assert.equal(JSON.parse(jsonText).kind, "kv260-status-panel"); } +async function testPreflightNotRunIsGracefulDefault() { + const panel = createKv260StatusPanel({ + launcherStatus: await readJson(LAUNCHER_FIXTURE), + traceManifest: await readJson(TRACE_FIXTURE), + }); + const text = formatKv260StatusPanel(panel); + + assert.equal(panel.serialProbe.status, "not_run"); + assert.ok(panel.preflight.items.every((item) => item.state === "not_run")); + assert.match(text, /serial\.ttyPort: preflight not run/); + assert.match(text, /serial\.kernelUname: preflight not run/); + assert.match(text, /serial\.xrtPresent: preflight not run/); +} + +async function testLiveSerialProbeTypeOnlySkipWithoutData() { + const raw = process.env.PCCX_KV260_SERIAL_PREFLIGHT_JSON; + if (!raw) { + console.log("skip: no live KV260 serial preflight JSON"); + return; + } + const launcherStatus = await readJson(LAUNCHER_FIXTURE); + launcherStatus.serial_probe = JSON.parse(raw); + const panel = createKv260StatusPanel({ + launcherStatus, + traceManifest: await readJson(TRACE_FIXTURE), + }); + + assert.equal(panel.serialProbe.schemaVersion, LAUNCHER_SERIAL_PREFLIGHT_STATUS_VERSION); + assert.ok(["available", "blocked", "not_run"].includes(panel.serialProbe.status)); +} + async function testRejectsInvalidOrUnsafeInputs() { assert.throws( () => LauncherStatusReader.consume({ bitstream_loaded: "false" }), @@ -107,6 +158,8 @@ async function testModuleSourceHasNoExecutionTerms() { await testReadersParseFixtures(); await testPanelRendersPreflightAndSafety(); +await testPreflightNotRunIsGracefulDefault(); +await testLiveSerialProbeTypeOnlySkipWithoutData(); await testRejectsInvalidOrUnsafeInputs(); await testModuleSourceHasNoExecutionTerms(); diff --git a/src/pccx_ide_cli/kv260_status.py b/src/pccx_ide_cli/kv260_status.py index 146cc3f..cdb576b 100644 --- a/src/pccx_ide_cli/kv260_status.py +++ b/src/pccx_ide_cli/kv260_status.py @@ -10,10 +10,22 @@ LAUNCHER_NPU_STATUS_MIRROR_VERSION = "pccx.ide.launcher-npu-status.local-mirror.v0" +LAUNCHER_SERIAL_PREFLIGHT_STATUS_VERSION = "pccx.launcher.kv260-serial-preflight.v0" TRACE_MANIFEST_SCHEMA_VERSION = "pccx.lab.kv260.trace-manifest.v0" KV260_STATUS_PANEL_VERSION = "pccx.ide.kv260-status-panel.v0" +@dataclass(frozen=True) +class SerialPreflightStatus: + schema_version: str + status: str + tty_port: str | None + login_ok: bool | None + kernel_uname: str | None + xrt_present: bool | None + last_preflight_at: str | None + + @dataclass(frozen=True) class NPUStatus: bitstream_loaded: bool @@ -21,6 +33,7 @@ class NPUStatus: axi_base_addr: int | None axi_stat_register_value: int | None last_error: str | None + serial_probe: SerialPreflightStatus @dataclass(frozen=True) @@ -50,6 +63,7 @@ class TraceManifest: class PreflightItem: item_id: str label: str + state: str satisfied: bool evidence: str @@ -69,7 +83,7 @@ class Kv260StatusPanel: class LauncherStatusReader: - """Read-only local mirror of launcher#70 `NPUStatus` type shape.""" + """Read-only local mirror of launcher NPU and serial preflight status.""" @classmethod def from_json_text(cls, text: str) -> NPUStatus: @@ -88,12 +102,14 @@ def from_object(value: Mapping[str, Any]) -> NPUStatus: axi_base_addr = _optional_int_field(value, "axi_base_addr") axi_stat_register_value = _optional_int_field(value, "axi_stat_register_value") last_error = _optional_string_field(value, "last_error") + serial_probe = _serial_preflight_status(value.get("serial_probe")) return NPUStatus( bitstream_loaded=bitstream_loaded, bitstream_uuid=bitstream_uuid, axi_base_addr=axi_base_addr, axi_stat_register_value=axi_stat_register_value, last_error=last_error, + serial_probe=serial_probe, ) @@ -136,27 +152,40 @@ def from_object(value: Mapping[str, Any]) -> TraceManifest: def create_preflight_proposal(status: NPUStatus, manifest: TraceManifest) -> PreflightProposal: - axi_present = status.axi_base_addr is not None and status.axi_stat_register_value is not None + del manifest + probe = status.serial_probe return PreflightProposal( kind="kv260-preflight-proposal", items=( PreflightItem( - item_id="bitstream_loaded", - label="bitstream loaded", - satisfied=status.bitstream_loaded, - evidence=status.bitstream_uuid or "launcher status reports no bitstream UUID", + item_id="serial_tty_port", + label="serial tty port", + state=_probe_value_state(probe, probe.tty_port is not None), + satisfied=probe.tty_port is not None, + evidence=_probe_evidence(probe, probe.tty_port), ), - PreflightItem( - item_id="axi_reachable", - label="AXI reachable", - satisfied=axi_present, - evidence=_hex_or_unavailable(status.axi_stat_register_value), + _bool_probe_item( + "serial_login", + "serial login", + probe, + probe.login_ok, + "login OK", + "login not OK", + ), + _bool_probe_item( + "serial_xrt", + "XRT present", + probe, + probe.xrt_present, + "xrt_present true", + "xrt_present false", ), PreflightItem( - item_id="manifest_available", - label="manifest available", - satisfied=manifest.frame_count >= 0, - evidence=f"{manifest.schema_version}; {manifest.frame_count} frame(s)", + item_id="serial_probe_timestamp", + label="serial preflight timestamp", + state=_probe_value_state(probe, probe.last_preflight_at is not None), + satisfied=probe.last_preflight_at is not None, + evidence=_probe_evidence(probe, probe.last_preflight_at), ), ), ) @@ -182,6 +211,25 @@ def panel_to_dict(panel: Kv260StatusPanel) -> dict[str, Any]: "axi_base_addr": panel.launcher_status.axi_base_addr, "axi_stat_register_value": panel.launcher_status.axi_stat_register_value, "last_error": panel.launcher_status.last_error, + "serial_probe": { + "schema_version": panel.launcher_status.serial_probe.schema_version, + "status": panel.launcher_status.serial_probe.status, + "tty_port": panel.launcher_status.serial_probe.tty_port, + "login_ok": panel.launcher_status.serial_probe.login_ok, + "kernel_uname": panel.launcher_status.serial_probe.kernel_uname, + "xrt_present": panel.launcher_status.serial_probe.xrt_present, + "last_preflight_at": panel.launcher_status.serial_probe.last_preflight_at, + }, + }, + "serial_probe": { + "schema_version": panel.launcher_status.serial_probe.schema_version, + "status": panel.launcher_status.serial_probe.status, + "tty_port": panel.launcher_status.serial_probe.tty_port, + "login_ok": panel.launcher_status.serial_probe.login_ok, + "kernel_uname": panel.launcher_status.serial_probe.kernel_uname, + "kernel_uname_display": _truncate_uname(panel.launcher_status.serial_probe.kernel_uname), + "xrt_present": panel.launcher_status.serial_probe.xrt_present, + "last_preflight_at": panel.launcher_status.serial_probe.last_preflight_at, }, "lab": { "schema_version": panel.trace_manifest.schema_version, @@ -198,6 +246,7 @@ def panel_to_dict(panel: Kv260StatusPanel) -> dict[str, Any]: { "item_id": item.item_id, "label": item.label, + "state": item.state, "satisfied": item.satisfied, "evidence": item.evidence, } @@ -218,6 +267,7 @@ def panel_to_dict(panel: Kv260StatusPanel) -> dict[str, Any]: def format_kv260_status_panel(panel: Kv260StatusPanel) -> str: status = panel.launcher_status + probe = status.serial_probe manifest = panel.trace_manifest lines = [ "KV260 Status Surface", @@ -228,6 +278,10 @@ def format_kv260_status_panel(panel: Kv260StatusPanel) -> str: f"launcher.axiBaseAddr: {_hex_or_unavailable(status.axi_base_addr)}", f"launcher.axiStatus: {_hex_or_unavailable(status.axi_stat_register_value)}", f"launcher.lastError: {status.last_error or 'none'}", + f"serial.ttyPort: {probe.tty_port or 'preflight not run'}", + f"serial.kernelUname: {_truncate_uname(probe.kernel_uname)}", + f"serial.xrtPresent: {_optional_yes_no(probe.xrt_present)}", + f"serial.lastPreflightAt: {probe.last_preflight_at or 'preflight not run'}", f"lab.schema: {manifest.schema_version}", f"lab.sourceKind: {manifest.source_kind}", f"lab.frames: {manifest.frame_count}", @@ -236,8 +290,7 @@ def format_kv260_status_panel(panel: Kv260StatusPanel) -> str: "preflight:", ] for item in panel.preflight.items: - state = "pass" if item.satisfied else "blocked" - lines.append(f"- {item.label}: {state} ({item.evidence})") + lines.append(f"- {item.label}: {item.state} ({item.evidence})") lines.extend([ "execution: no launcher, no pccx-lab, no shell, no SSH, no KV260 control", "writeBack: no", @@ -260,6 +313,15 @@ def _bool_field(value: Mapping[str, Any], key: str) -> bool: return field +def _optional_bool_field(value: Mapping[str, Any], key: str) -> bool | None: + field = value.get(key) + if field is None: + return None + if not isinstance(field, bool): + raise ValueError(f"{key} must be null or a boolean") + return field + + def _int_field(value: Mapping[str, Any], key: str) -> int: field = value.get(key) if not isinstance(field, int) or isinstance(field, bool) or field < 0: @@ -292,6 +354,42 @@ def _optional_string_field(value: Mapping[str, Any], key: str) -> str | None: return field +def _serial_preflight_status(value: Any) -> SerialPreflightStatus: + if value is None: + value = { + "schema_version": LAUNCHER_SERIAL_PREFLIGHT_STATUS_VERSION, + "status": "not_run", + "tty_port": None, + "login_ok": None, + "kernel_uname": None, + "xrt_present": None, + "last_preflight_at": None, + } + if not isinstance(value, Mapping): + raise ValueError("serial_probe must be a JSON object") + schema_version = value.get("schema_version") or LAUNCHER_SERIAL_PREFLIGHT_STATUS_VERSION + if schema_version != LAUNCHER_SERIAL_PREFLIGHT_STATUS_VERSION: + raise ValueError( + "serial_probe.schema_version must be " + f"{LAUNCHER_SERIAL_PREFLIGHT_STATUS_VERSION}" + ) + status = value.get("status") or "not_run" + if status not in {"available", "blocked", "not_run"}: + raise ValueError("serial_probe.status must be available, blocked, or not_run") + return SerialPreflightStatus( + schema_version=schema_version, + status=status, + tty_port=_optional_string_field(value, "tty_port"), + login_ok=_optional_bool_field(value, "login_ok"), + kernel_uname=_optional_string_field(value, "kernel_uname"), + xrt_present=_optional_bool_field(value, "xrt_present"), + last_preflight_at=_optional_string_field( + {"last_preflight_at": value.get("last_preflight_at", value.get("checked_at"))}, + "last_preflight_at", + ), + ) + + def _list_field(value: Mapping[str, Any], key: str) -> list[Any]: field = value.get(key) if not isinstance(field, list): @@ -329,3 +427,54 @@ def _hex_or_unavailable(value: int | None) -> str: def _yes_no(value: bool) -> str: return "yes" if value else "no" + + +def _optional_yes_no(value: bool | None) -> str: + if value is None: + return "preflight not run" + return _yes_no(value) + + +def _probe_value_state(probe: SerialPreflightStatus, present: bool) -> str: + if probe.status == "not_run": + return "not_run" + return "pass" if present else "blocked" + + +def _probe_evidence(probe: SerialPreflightStatus, value: str | None) -> str: + if probe.status == "not_run": + return "preflight not run" + return value or "unavailable" + + +def _bool_probe_item( + item_id: str, + label: str, + probe: SerialPreflightStatus, + value: bool | None, + true_text: str, + false_text: str, +) -> PreflightItem: + if probe.status == "not_run": + return PreflightItem( + item_id=item_id, + label=label, + state="not_run", + satisfied=False, + evidence="preflight not run", + ) + return PreflightItem( + item_id=item_id, + label=label, + state="pass" if value else "blocked", + satisfied=value is True, + evidence=true_text if value else false_text, + ) + + +def _truncate_uname(value: str | None, max_characters: int = 96) -> str: + if not value: + return "preflight not run" + if len(value) <= max_characters: + return value + return value[: max_characters - 3] + "..." diff --git a/tests/test_kv260_status.py b/tests/test_kv260_status.py index 995d59f..2289a1d 100644 --- a/tests/test_kv260_status.py +++ b/tests/test_kv260_status.py @@ -16,6 +16,7 @@ sys.path.insert(0, str(SRC)) from pccx_ide_cli.kv260_status import ( + LAUNCHER_SERIAL_PREFLIGHT_STATUS_VERSION, LabTraceReader, LauncherStatusReader, create_kv260_status_panel, @@ -49,13 +50,24 @@ def test_readers_parse_tiny_fixtures_as_data_only() -> None: assert launcher.bitstream_loaded is False assert launcher.bitstream_uuid is None + assert launcher.serial_probe.status == "not_run" assert manifest.schema_version == "pccx.lab.kv260.trace-manifest.v0" assert manifest.frame_count == 1 assert manifest.frames[0].result_payload == "a55a0001" def test_status_panel_and_preflight_are_deterministic() -> None: - launcher = LauncherStatusReader.from_json_file(LAUNCHER_FIXTURE) + launcher_data = json.loads(LAUNCHER_FIXTURE.read_text(encoding="utf-8")) + launcher_data["serial_probe"] = { + "schema_version": LAUNCHER_SERIAL_PREFLIGHT_STATUS_VERSION, + "status": "available", + "tty_port": "/dev/ttyUSB0", + "login_ok": True, + "kernel_uname": "Linux kv260 6.6.0-test #1 SMP PREEMPT aarch64 GNU/Linux", + "xrt_present": True, + "last_preflight_at": "2026-05-06T09:00:00Z", + } + launcher = LauncherStatusReader.from_object(launcher_data) manifest = LabTraceReader.from_json_file(TRACE_FIXTURE) panel = create_kv260_status_panel(launcher, manifest) payload = panel_to_dict(panel) @@ -66,27 +78,67 @@ def test_status_panel_and_preflight_are_deterministic() -> None: assert payload["safety"]["launcher_execution"] is False assert payload["safety"]["pccx_lab_execution"] is False assert payload["safety"]["ssh_execution"] is False + assert payload["serial_probe"]["tty_port"] == "/dev/ttyUSB0" + assert payload["serial_probe"]["xrt_present"] is True + assert payload["serial_probe"]["last_preflight_at"] == "2026-05-06T09:00:00Z" assert [item["item_id"] for item in payload["preflight"]["items"]] == [ - "bitstream_loaded", - "axi_reachable", - "manifest_available", + "serial_tty_port", + "serial_login", + "serial_xrt", + "serial_probe_timestamp", ] assert "KV260 Status Surface" in text + assert "serial.ttyPort: /dev/ttyUSB0" in text + assert "serial.kernelUname: Linux kv260" in text + assert "serial.xrtPresent: yes" in text + assert "serial.lastPreflightAt: 2026-05-06T09:00:00Z" in text assert "execution: no launcher, no pccx-lab, no shell, no SSH, no KV260 control" in text +def test_preflight_not_run_is_graceful_default() -> None: + launcher = LauncherStatusReader.from_json_file(LAUNCHER_FIXTURE) + manifest = LabTraceReader.from_json_file(TRACE_FIXTURE) + panel = create_kv260_status_panel(launcher, manifest) + payload = panel_to_dict(panel) + text = format_kv260_status_panel(panel) + + assert payload["serial_probe"]["status"] == "not_run" + assert {item["state"] for item in payload["preflight"]["items"]} == {"not_run"} + assert "serial.ttyPort: preflight not run" in text + assert "serial.kernelUname: preflight not run" in text + assert "serial.xrtPresent: preflight not run" in text + + +def test_live_serial_probe_type_only_skips_without_data() -> None: + raw = os.environ.get("PCCX_KV260_SERIAL_PREFLIGHT_JSON") + if not raw: + pytest.skip("no live KV260 serial preflight JSON") + launcher_data = json.loads(LAUNCHER_FIXTURE.read_text(encoding="utf-8")) + launcher_data["serial_probe"] = json.loads(raw) + launcher = LauncherStatusReader.from_object(launcher_data) + manifest = LabTraceReader.from_json_file(TRACE_FIXTURE) + panel = create_kv260_status_panel(launcher, manifest) + + assert panel.launcher_status.serial_probe.schema_version == ( + LAUNCHER_SERIAL_PREFLIGHT_STATUS_VERSION + ) + assert panel.launcher_status.serial_probe.status in {"available", "blocked", "not_run"} + + def test_cli_kv260_status_prints_text_and_json() -> None: text_result = _run_cli("kv260-status") assert text_result.returncode == 0, text_result.stderr assert "KV260 Status Surface" in text_result.stdout assert "launcherMirror: pccx.ide.launcher-npu-status.local-mirror.v0" in text_result.stdout assert "preflight:" in text_result.stdout + assert "serial.ttyPort: preflight not run" in text_result.stdout json_result = _run_cli("kv260-status", "--format", "json") assert json_result.returncode == 0, json_result.stderr payload = json.loads(json_result.stdout) assert payload["kind"] == "kv260-status-panel" assert payload["lab"]["source_kind"] == "file_replay" + assert payload["serial_probe"]["status"] == "not_run" def test_readers_reject_unsafe_or_invalid_shapes() -> None: