Merge pull request #2590 from pikasTech/feat/2586-caserun-read-model
Pipelines as Code CI / hwlab-nc01-v03-ci-poll- Success

feat: 对齐 CaseRun 同源 read model
This commit is contained in:
Lyon
2026-07-17 02:38:01 +08:00
committed by GitHub
6 changed files with 186 additions and 43 deletions
+72 -5
View File
@@ -60,10 +60,15 @@ test("cloud-api exposes web CaseRun cases, run status, events and aggregate", as
assert.equal(startResponse.status, 202);
assert.ok(["queued", "running"].includes(startPayload.status));
assert.match(startPayload.runId, /^web-d601-f103-v2-compile-/u);
assert.equal(startPayload.runtimeApiUrl, serverUrl(server));
assert.equal(startPayload.statusUrl, `/v1/caserun/runs/${startPayload.runId}`);
const runPayload = await waitForRun(server, startPayload.runId, "completed");
assert.equal(runPayload.summary.jobId, "job-test");
assert.equal(runPayload.terminal, true);
assert.equal(runPayload.facts.jobId, "job-test");
assert.equal(runPayload.facts.artifactCount, 2);
assert.equal(runPayload.references.events.count >= 3, true);
assert.match(runPayload.references.aggregate.sha256, /^[a-f0-9]{64}$/u);
assert.equal(runPayload.sourceAuthority.authority, "caserun-archived-facts");
assert.equal(seen[0].noCaseRepoRecord, true);
assert.equal(seen[0].caseRepo, root);
assert.equal(seen[0].apiUrl, "http://hwlab-cloud-api.hwlab-v03.svc.cluster.local:6667");
@@ -71,6 +76,9 @@ test("cloud-api exposes web CaseRun cases, run status, events and aggregate", as
const eventsResponse = await fetch(`${serverUrl(server)}/v1/caserun/runs/${startPayload.runId}/events`);
const eventsPayload = await eventsResponse.json();
assert.equal(eventsResponse.status, 200);
assert.equal(eventsPayload.runId, runPayload.runId);
assert.equal(eventsPayload.caseId, runPayload.caseId);
assert.equal(eventsPayload.sourceAuthority.authority, runPayload.sourceAuthority.authority);
assert.ok(eventsPayload.events.some((event: any) => event.stage === "build-completed"));
const aggregateResponse = await fetch(`${serverUrl(server)}/v1/caserun/runs/${startPayload.runId}/aggregate`);
@@ -78,6 +86,7 @@ test("cloud-api exposes web CaseRun cases, run status, events and aggregate", as
assert.equal(aggregateResponse.status, 200);
assert.equal(aggregatePayload.summary.jobId, "job-test");
assert.match(aggregatePayload.sha256, /^[a-f0-9]{64}$/u);
assert.equal(aggregatePayload.sourceAuthority.authority, runPayload.sourceAuthority.authority);
} finally {
await close(server);
await rm(root, { recursive: true, force: true });
@@ -119,9 +128,8 @@ test("cloud-api completes a hardware-free software smoke CaseRun", async () => {
const started = await startResponse.json();
assert.equal(startResponse.status, 202);
const completed = await waitForRun(server, started.runId, "completed");
assert.equal(completed.summary.mode, "software-smoke");
assert.equal(completed.summary.hardwareRequired, false);
assert.equal(completed.summary.artifactCount, 1);
assert.equal(completed.terminal, true);
assert.equal(completed.facts.artifactCount, 1);
const eventsResponse = await fetch(`${serverUrl(server)}/v1/caserun/runs/${started.runId}/events`);
const eventsPayload = await eventsResponse.json();
@@ -140,6 +148,65 @@ test("cloud-api completes a hardware-free software smoke CaseRun", async () => {
}
});
test("cloud-api projects queued, running, completed and failed records with one read-model shape", async () => {
const root = await mkdtemp(path.join(os.tmpdir(), "hwlab-caserun-read-model-"));
const stateRoot = path.join(root, "state");
const accessController = {
async ensureBootstrap() {},
async requireNavAccess() { return true; }
};
const server = createCloudApiServer({
accessController,
env: { PATH: process.env.PATH, HWLAB_CASERUN_CASE_REPO: root, HWLAB_CASERUN_STATE_DIR: stateRoot }
});
await mkdir(path.join(stateRoot, "records"), { recursive: true });
const statuses = ["queued", "running", "completed", "failed"];
for (const status of statuses) {
const runId = `read-model-${status}`;
const record = {
runId,
caseId: "read-model-case",
status,
stage: status === "running" ? "build" : status,
runDir: path.join(stateRoot, "runs", runId),
createdAt: "2026-07-16T00:00:00.000Z",
updatedAt: "2026-07-16T00:00:01.000Z",
caseAuthority: { kind: "case-repo" },
events: [],
result: status === "completed" ? {
summary: { jobId: "job-read-model", artifactCount: 1, artifactManifestPath: "runs/read-model/artifact-manifest.json", artifactManifestSha256: "a".repeat(64) },
evidence: { status: "recorded", replay: { contractVersion: "hwpod-case-validation-replay-v1" }, validation: { archivePath: "validation-observations.json" }, hwpod: { operation: { rawArtifactRef: { path: "raw.json" } } } },
run: { runId, caseId: "read-model-case" }
} : null,
blocker: status === "failed" ? { code: "hwpod_build_failed", summary: "HWPOD build failed", layer: "hwpod" } : null,
error: status === "failed" ? { code: "caserun_failed", message: "CaseRun failed." } : null
};
await mkdir(record.runDir, { recursive: true });
await writeFile(path.join(stateRoot, "records", `${runId}.json`), `${JSON.stringify(record)}\n`, "utf8");
}
await listen(server);
try {
for (const status of statuses) {
const response = await fetch(`${serverUrl(server)}/v1/caserun/runs/read-model-${status}`);
const payload = await response.json();
assert.equal(response.status, 200);
assert.deepEqual(Object.keys(payload).sort(), ["blocker", "caseId", "contractVersion", "createdAt", "error", "facts", "ok", "references", "runId", "sourceAuthority", "stage", "status", "terminal", "updatedAt"].sort());
assert.equal(payload.status, status);
assert.equal(payload.terminal, ["completed", "failed"].includes(status));
assert.equal(payload.references.status.href, `/v1/caserun/runs/read-model-${status}`);
}
const completed = await (await fetch(`${serverUrl(server)}/v1/caserun/runs/read-model-completed`)).json();
assert.equal(completed.references.manifest.sha256, "a".repeat(64));
assert.equal(completed.references.replay.ref, "validation-observations.json");
assert.equal(completed.references.hwpod.path, "raw.json");
const failed = await (await fetch(`${serverUrl(server)}/v1/caserun/runs/read-model-failed`)).json();
assert.equal(failed.blocker.code, "hwpod_build_failed");
} finally {
await close(server);
await rm(root, { recursive: true, force: true });
}
});
async function waitForRun(server: any, runId: string, status: string) {
const deadline = Date.now() + 3000;
let last: any = null;
+97 -11
View File
@@ -76,7 +76,7 @@ export async function handleCaseRunHttp(request: IncomingMessage, response: Serv
remember(record);
appendEvent(record, "queued", "queued", { caseId });
startBackgroundCaseRun(record, { ...options, env, cwd, stateRoot, caseRepo, caseDefinition });
sendJson(response, 202, publicRecord(record));
sendJson(response, 202, acceptedRecord(record));
return;
}
@@ -87,8 +87,18 @@ export async function handleCaseRunHttp(request: IncomingMessage, response: Serv
const suffix = runMatch[2] ?? "";
const record = await findRecord(stateRoot, runId);
if (!record) return sendError(response, 404, "caserun_run_not_found", "CaseRun run was not found.", { runId });
if (!suffix && request.method === "GET") return sendJson(response, 200, publicRecord(record));
if (suffix === "events" && request.method === "GET") return sendJson(response, 200, { ok: true, contractVersion: CONTRACT_VERSION, runId, events: record.events ?? [] });
if (!suffix && request.method === "GET") return sendJson(response, 200, await projectRunReadModel(record));
if (suffix === "events" && request.method === "GET") return sendJson(response, 200, {
ok: true,
contractVersion: CONTRACT_VERSION,
runId,
caseId: record.caseId,
status: record.status,
stage: record.stage,
terminal: terminalStatus(record.status),
sourceAuthority: sourceAuthorityForRecord(record),
events: record.events ?? []
});
if (suffix === "aggregate" && request.method === "GET") return sendJson(response, 200, await aggregateForRecord(record));
if (suffix === "cancel" && request.method === "POST") return sendError(response, 409, "caserun_cancel_not_supported", "This compile-only CaseRun worker cannot be cancelled after dispatch.", { runId, status: record.status });
return sendError(response, 405, "method_not_allowed", suffix === "cancel" ? "POST required." : "GET required.");
@@ -103,6 +113,7 @@ function startBackgroundCaseRun(record: any, input: any) {
record.status = "failed";
record.stage = "failed";
record.error = { code: error?.code ?? error?.name ?? "caserun_failed", message: String(error?.message ?? error), details: error?.details ?? null };
record.blocker = typedBlocker(error?.blocker ?? error?.details?.blocker ?? error?.details ?? null);
record.updatedAt = new Date().toISOString();
await saveRecord(input.stateRoot, record).catch(() => undefined);
});
@@ -303,6 +314,9 @@ async function aggregateForRecord(record: any) {
runId: record.runId,
caseId: record.caseId,
status: record.status,
stage: record.stage,
terminal: terminalStatus(record.status),
sourceAuthority: sourceAuthorityForRecord(record),
sha256: sha256Json(source),
summary,
evidence,
@@ -331,7 +345,18 @@ function summarizeEvidence(evidence: any) {
};
}
function publicRecord(record: any) {
async function projectRunReadModel(record: any) {
const aggregate = await aggregateForRecord(record);
const summary = aggregate.summary && typeof aggregate.summary === "object" ? aggregate.summary : {};
const evidence = aggregate.evidence && typeof aggregate.evidence === "object" ? aggregate.evidence : {};
const archivedRun = aggregate.run && typeof aggregate.run === "object" ? aggregate.run : {};
const blocker = typedBlocker(record.blocker ?? archivedRun.blocker ?? evidence.validation?.blocker ?? evidence.hwpod?.buildWaitBlocker ?? evidence.hwpod?.operation?.blocker ?? record.error?.details?.blocker ?? null);
const manifestPath = firstText(summary.artifactManifestPath, evidence.artifactManifest?.path, archivedRun.artifactManifestPath);
const manifestSha256 = firstSha256(summary.artifactManifestSha256, evidence.artifactManifest?.sha256, archivedRun.artifactManifestSha256);
const replay = evidence.replay && typeof evidence.replay === "object" ? evidence.replay : null;
const replayPath = firstText(evidence.validation?.archivePath, archivedRun.replayPath);
const trace = firstObject(summary.trace, summary.traceLookup, evidence.agentTrace, evidence.traceLookup, archivedRun.agentTrace);
const hwpod = firstObject(evidence.hwpod?.operation?.rawArtifactRef, evidence.hwpod?.operation, evidence.keilJob, archivedRun.hwpod);
return {
ok: true,
contractVersion: CONTRACT_VERSION,
@@ -339,18 +364,79 @@ function publicRecord(record: any) {
caseId: record.caseId,
status: record.status,
stage: record.stage,
runDir: record.runDir,
terminal: terminalStatus(record.status),
blocker,
createdAt: record.createdAt,
updatedAt: record.updatedAt,
runtimeApiUrl: record.runtimeApiUrl,
caseAuthority: record.caseAuthority,
eventCount: Array.isArray(record.events) ? record.events.length : 0,
lastEvent: Array.isArray(record.events) ? record.events.at(-1) ?? null : null,
summary: record.result?.summary ?? summarizeEvidence(record.result?.evidence),
error: record.error ?? null
facts: {
jobId: firstText(summary.jobId, evidence.keilJob?.jobId, evidence.hwpod?.operation?.jobId) || null,
artifactCount: Number(summary.artifactCount ?? (Array.isArray(evidence.artifacts) ? evidence.artifacts.length : 0)),
evidenceStatus: firstText(evidence.status, summary.status) || null
},
references: {
status: { href: `/v1/caserun/runs/${encodeURIComponent(record.runId)}` },
events: { href: `/v1/caserun/runs/${encodeURIComponent(record.runId)}/events`, count: Array.isArray(record.events) ? record.events.length : 0 },
aggregate: { href: `/v1/caserun/runs/${encodeURIComponent(record.runId)}/aggregate`, sha256: aggregate.sha256 },
manifest: manifestPath ? { ref: manifestPath, sha256: manifestSha256 } : null,
replay: replay || replayPath ? { ref: replayPath || null, relationship: replay } : null,
trace,
hwpod
},
sourceAuthority: sourceAuthorityForRecord(record),
error: record.error ? { code: record.error.code ?? "caserun_failed", message: record.error.message ?? "CaseRun failed." } : null
};
}
function acceptedRecord(record: any) {
return {
ok: true,
contractVersion: CONTRACT_VERSION,
runId: record.runId,
caseId: record.caseId,
status: record.status,
statusUrl: `/v1/caserun/runs/${encodeURIComponent(record.runId)}`
};
}
function sourceAuthorityForRecord(record: any) {
return {
projector: "hwlab-cloud-api",
authority: "caserun-archived-facts",
runRecord: "cloud-api-caserun-record",
archive: "caserun-run-directory",
execution: "hwpod-agentrun-caserun",
case: record.caseAuthority ?? null
};
}
function terminalStatus(status: unknown) {
return ["completed", "failed", "cancelled", "canceled", "timed_out", "timeout"].includes(String(status ?? "").toLowerCase());
}
function typedBlocker(value: any) {
if (!value || typeof value !== "object" || Array.isArray(value)) return null;
const code = firstText(value.code);
if (!code) return null;
return {
code,
summary: firstText(value.summary, value.message) || code,
layer: firstText(value.layer) || null,
details: value.details && typeof value.details === "object" ? value.details : null
};
}
function firstObject(...values: any[]) {
return values.find((value) => value && typeof value === "object" && !Array.isArray(value)) ?? null;
}
function firstText(...values: unknown[]) {
return values.map((value) => String(value ?? "").trim()).find(Boolean) ?? "";
}
function firstSha256(...values: unknown[]) {
return values.map((value) => String(value ?? "").trim().toLowerCase()).find((value) => /^[a-f0-9]{64}$/u.test(value)) ?? null;
}
function runtimeApiUrlFromRequest(request: IncomingMessage, env: Record<string, string | undefined>) {
const explicit = String(env.HWLAB_CASERUN_RUNTIME_API_URL ?? "").trim();
if (explicit) return explicit.replace(/\/+$/u, "");
+2 -3
View File
@@ -1,5 +1,5 @@
import { fetchJson } from "./client";
import type { ApiResult, CaseRunAggregateResponse, CaseRunCasesResponse, CaseRunEventsResponse, CaseRunRunResponse } from "@/types";
import type { ApiResult, CaseRunCasesResponse, CaseRunEventsResponse, CaseRunRunResponse } from "@/types";
export const caserunAPI = {
cases: (): Promise<ApiResult<CaseRunCasesResponse>> => fetchJson("/v1/caserun/cases", { timeoutMs: 12000, timeoutName: "CaseRun cases" }),
@@ -10,6 +10,5 @@ export const caserunAPI = {
timeoutName: "CaseRun start"
}),
run: (runId: string): Promise<ApiResult<CaseRunRunResponse>> => fetchJson(`/v1/caserun/runs/${encodeURIComponent(runId)}`, { timeoutMs: 12000, timeoutName: "CaseRun run" }),
events: (runId: string): Promise<ApiResult<CaseRunEventsResponse>> => fetchJson(`/v1/caserun/runs/${encodeURIComponent(runId)}/events`, { timeoutMs: 12000, timeoutName: "CaseRun events" }),
aggregate: (runId: string): Promise<ApiResult<CaseRunAggregateResponse>> => fetchJson(`/v1/caserun/runs/${encodeURIComponent(runId)}/aggregate`, { timeoutMs: 12000, timeoutName: "CaseRun aggregate" })
events: (runId: string): Promise<ApiResult<CaseRunEventsResponse>> => fetchJson(`/v1/caserun/runs/${encodeURIComponent(runId)}/events`, { timeoutMs: 12000, timeoutName: "CaseRun events" })
};
@@ -16,11 +16,10 @@ const statusTone = computed(() => {
return "pending";
});
const eventLines = computed(() => caserun.events.map((event) => `[${formatTimestamp(event.at)}] ${event.stage} ${event.status}${event.payload ? ` ${JSON.stringify(event.payload)}` : ""}`));
const artifactCount = computed(() => {
const artifacts = caserun.aggregate?.summary?.artifacts;
return Array.isArray(artifacts) ? artifacts.length : Number(caserun.aggregate?.summary?.artifactCount ?? 0);
});
const jobId = computed(() => String(caserun.aggregate?.summary?.jobId ?? caserun.currentRun?.summary?.jobId ?? ""));
const artifactCount = computed(() => Number(caserun.currentRun?.facts?.artifactCount ?? 0));
const jobId = computed(() => String(caserun.currentRun?.facts?.jobId ?? ""));
const aggregateHash = computed(() => caserun.currentRun?.references?.aggregate?.sha256 ?? "");
const blockerText = computed(() => caserun.currentRun?.blocker ? `${caserun.currentRun.blocker.code}: ${caserun.currentRun.blocker.summary}` : "");
onMounted(async () => {
await caserun.refreshCases();
@@ -73,11 +72,11 @@ async function startRun(): Promise<void> {
<span>events</span>
<span>{{ caserun.events.length }}</span>
</header>
<pre id="caserun-event-text">{{ eventLines.join('\n') || caserun.error || '等待 CaseRun。' }}</pre>
<pre id="caserun-event-text">{{ eventLines.join('\n') || blockerText || caserun.error || '等待 CaseRun。' }}</pre>
</section>
<section class="caserun-aggregate" id="caserun-aggregate" aria-label="CaseRun aggregate">
<strong>{{ caserun.aggregate?.sha256 ? caserun.aggregate.sha256.slice(0, 16) : 'aggregate pending' }}</strong>
<span>{{ caserun.aggregate?.status || caserun.currentRun?.status || 'idle' }}</span>
<strong>{{ aggregateHash ? aggregateHash.slice(0, 16) : 'aggregate pending' }}</strong>
<span>{{ caserun.currentRun?.status || 'idle' }}</span>
</section>
</aside>
</template>
+4 -13
View File
@@ -1,13 +1,12 @@
import { computed, ref } from "vue";
import { defineStore } from "pinia";
import { caserunAPI } from "@/api";
import type { CaseRunAggregateResponse, CaseRunCaseSummary, CaseRunEvent, CaseRunRunResponse } from "@/types";
import type { CaseRunCaseSummary, CaseRunEvent, CaseRunRunResponse } from "@/types";
export const useCaseRunStore = defineStore("caserun", () => {
const cases = ref<CaseRunCaseSummary[]>([]);
const currentRun = ref<CaseRunRunResponse | null>(null);
const events = ref<CaseRunEvent[]>([]);
const aggregate = ref<CaseRunAggregateResponse | null>(null);
const loading = ref(false);
const starting = ref(false);
const error = ref<string | null>(null);
@@ -25,7 +24,6 @@ export const useCaseRunStore = defineStore("caserun", () => {
async function startRun(caseId: string): Promise<CaseRunRunResponse | null> {
starting.value = true;
error.value = null;
aggregate.value = null;
events.value = [];
const result = await caserunAPI.startRun(caseId);
starting.value = false;
@@ -34,7 +32,7 @@ export const useCaseRunStore = defineStore("caserun", () => {
return null;
}
currentRun.value = result.data;
await refreshEvents(result.data.runId ?? "");
await refreshRun(result.data.runId ?? "");
return result.data;
}
@@ -47,20 +45,13 @@ export const useCaseRunStore = defineStore("caserun", () => {
}
currentRun.value = result.data;
await refreshEvents(runId);
if (result.data.status === "completed") await refreshAggregate(runId);
}
async function refreshEvents(runId = currentRun.value?.runId ?? ""): Promise<void> {
if (!runId) return;
const result = await caserunAPI.events(runId);
if (result.ok) events.value = result.data?.events ?? [];
if (result.ok && result.data?.runId === runId) events.value = result.data.events ?? [];
}
async function refreshAggregate(runId = currentRun.value?.runId ?? ""): Promise<void> {
if (!runId) return;
const result = await caserunAPI.aggregate(runId);
if (result.ok) aggregate.value = result.data;
}
return { cases, currentRun, events, aggregate, loading, starting, error, selectedCaseId, refreshCases, startRun, refreshRun, refreshEvents, refreshAggregate };
return { cases, currentRun, events, loading, starting, error, selectedCaseId, refreshCases, startRun, refreshRun, refreshEvents };
});
+4 -3
View File
@@ -518,9 +518,10 @@ export interface HwlabNodeUpdateMetadata {
export interface CaseRunCaseSummary { caseId: string; title?: string; mode?: string; available?: boolean; hwpodSpec?: string; subject?: Record<string, unknown> | null; expected?: Record<string, unknown> | null; runtime?: Record<string, unknown> | null; [key: string]: unknown }
export interface CaseRunCasesResponse { ok?: boolean; contractVersion?: string; count?: number; cases?: CaseRunCaseSummary[]; caseAuthority?: Record<string, unknown>; [key: string]: unknown }
export interface CaseRunEvent { at?: string; status?: string; stage?: string; payload?: Record<string, unknown>; [key: string]: unknown }
export interface CaseRunRunResponse { ok?: boolean; contractVersion?: string; runId?: string; caseId?: string; status?: string; stage?: string; runDir?: string; runtimeApiUrl?: string; createdAt?: string; updatedAt?: string; eventCount?: number; lastEvent?: CaseRunEvent | null; summary?: Record<string, unknown> | null; error?: Record<string, unknown> | null; [key: string]: unknown }
export interface CaseRunEventsResponse { ok?: boolean; contractVersion?: string; runId?: string; events?: CaseRunEvent[]; [key: string]: unknown }
export interface CaseRunAggregateResponse { ok?: boolean; contractVersion?: string; runId?: string; caseId?: string; status?: string; sha256?: string; summary?: Record<string, unknown> | null; evidence?: Record<string, unknown> | null; run?: Record<string, unknown> | null; [key: string]: unknown }
export interface CaseRunBlocker { code: string; summary: string; layer?: string | null; details?: Record<string, unknown> | null }
export interface CaseRunReference { href?: string; ref?: string | null; sha256?: string | null; count?: number; relationship?: Record<string, unknown> | null; [key: string]: unknown }
export interface CaseRunRunResponse { ok?: boolean; contractVersion?: string; runId?: string; caseId?: string; status?: string; stage?: string; terminal?: boolean; blocker?: CaseRunBlocker | null; createdAt?: string; updatedAt?: string; facts?: { jobId?: string | null; artifactCount?: number; evidenceStatus?: string | null }; references?: { status?: CaseRunReference; events?: CaseRunReference; aggregate?: CaseRunReference; manifest?: CaseRunReference | null; replay?: CaseRunReference | null; trace?: Record<string, unknown> | null; hwpod?: Record<string, unknown> | null }; sourceAuthority?: Record<string, unknown>; error?: { code?: string; message?: string } | null; statusUrl?: string; [key: string]: unknown }
export interface CaseRunEventsResponse { ok?: boolean; contractVersion?: string; runId?: string; caseId?: string; status?: string; stage?: string; terminal?: boolean; sourceAuthority?: Record<string, unknown>; events?: CaseRunEvent[]; [key: string]: unknown }
export interface ProviderProfilesResponse { ok?: boolean; status?: string; profiles?: Array<Record<string, unknown>>; [key: string]: unknown }
export interface ProviderProfileCatalogItem {
profile: string;