mirrors/pikasTech-unidesk
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: close Release A monitor migration blockers

Browse Source
This commit is contained in:
AgentRun Codex
2026-07-12 23:24:41 +00:00
parent e3d21ed4e1
commit c101b2543e
9 changed files with 162 additions and 64 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scripts/src/monitor-sqlite-migration.ts
+16 -8
View File
@@ -5,7 +5,7 @@ import { closeSync, existsSync, openSync, readFileSync, statSync, writeFileSync
import { dirname, join, resolve } from "node:path";
import { Database } from "bun:sqlite";
import type { MonitorArtifactLocator, MonitorCentralIngest, MonitorCentralStore } from "./monitor-central-persistence";
import type { MonitorArtifactLocator, MonitorCentralIngest, MonitorCentralStore, MonitorRuntimeMetadata } from "./monitor-central-persistence";
import { monitorCentralStableJson, normalizeMonitorCentralIngest } from "./monitor-central-persistence";
export const MONITOR_SQLITE_MIGRATION_SCHEMA = "2026-07-12-p17-2";
@@ -22,13 +22,13 @@ export interface MonitorMigrationReconciliation { readonly runs: number; readonl
export async function verifyMonitorSqliteSnapshotAgainstStore(snapshot: MonitorSqliteSnapshot, store: MonitorCentralStore): Promise<Record<string, unknown>> {
const snapshotVerification = verifyMonitorSqliteSnapshot(snapshot);
const expected = canonicalIngests(snapshot);
const expectedMetadata = canonicalRuntimeMetadata(snapshot);
const checks: Record<string, unknown>[] = [];
for (const input of expected) {
const scope = { sentinelId: input.sentinelId, node: input.node, lane: input.lane };
const actual = await store.run(scope, input.runId);
checks.push({ name: "run", runId: input.runId, ok: actual?.payloadHash === input.payloadHash && actual?.status === input.status && monitorCentralStableJson(actual?.payload) === monitorCentralStableJson(input.payload), valuesRedacted: true });
checks.push({ name: "run", runId: input.runId, ok: actual?.payloadHash === input.payloadHash && actual?.status === input.status && actual?.updatedAt === input.updatedAt && monitorCentralStableJson(actual?.payload) === monitorCentralStableJson(input.payload), valuesRedacted: true });
checks.push({ name: "finding", runId: input.runId, ok: monitorCentralStableJson(actual?.findings) === monitorCentralStableJson(input.findings), valuesRedacted: true });
checks.push({ name: "metadata", runId: input.runId, ok: monitorCentralStableJson(actual?.payload) === monitorCentralStableJson(input.payload), valuesRedacted: true });
checks.push({ name: "artifactLocator", runId: input.runId, ok: monitorCentralStableJson(actual?.artifactLocators) === monitorCentralStableJson(input.artifactLocators), valuesRedacted: true });
const views = isRecord(input.payload.views) ? input.payload.views : {};
for (const [name, value] of Object.entries(views)) {
@@ -39,14 +39,18 @@ export async function verifyMonitorSqliteSnapshotAgainstStore(snapshot: MonitorS
const latestFor = (items: readonly typeof expected) => [...items].sort((left, right) => right.updatedAt.localeCompare(left.updatedAt) || right.runId.localeCompare(left.runId))[0] ?? null;
const globalLatest = latestFor(expected);
const global = await store.overview({});
checks.push({ name: "globalLatest", ok: globalLatest === null ? global.latest === null : isRecord(global.latest) && global.latest.runId === globalLatest.runId && global.latest.updatedAt === globalLatest.updatedAt, valuesRedacted: true });
checks.push({ name: "globalOverview", ok: global.runCount === expected.length && (globalLatest === null ? global.latest === null : sameRun(global.latest, globalLatest)), valuesRedacted: true });
const scopes = [...new Map(expected.map((item) => [`${item.sentinelId}/${item.node}/${item.lane}`, { sentinelId: item.sentinelId, node: item.node, lane: item.lane }])).values()];
for (const scope of scopes) {
const latest = latestFor(expected.filter((item) => item.sentinelId === scope.sentinelId && item.node === scope.node && item.lane === scope.lane));
const overview = await store.overview(scope);
checks.push({ name: "scopedLatest", scope, ok: latest === null ? overview.latest === null : isRecord(overview.latest) && overview.latest.runId === latest.runId && overview.latest.updatedAt === latest.updatedAt, valuesRedacted: true });
const expectedCount = expected.filter((item) => item.sentinelId === scope.sentinelId && item.node === scope.node && item.lane === scope.lane).length;
checks.push({ name: "scopedOverview", scope, ok: overview.runCount === expectedCount && (latest === null ? overview.latest === null : sameRun(overview.latest, latest)), valuesRedacted: true });
const actualMetadata = await store.runtimeMetadata(scope);
const scopedMetadata = expectedMetadata.filter((item) => item.sentinelId === scope.sentinelId && item.node === scope.node && item.lane === scope.lane);
checks.push({ name: "runtimeMetadata", scope, ok: monitorCentralStableJson(actualMetadata) === monitorCentralStableJson(scopedMetadata), sourceCount: scopedMetadata.length, targetCount: actualMetadata.length, valuesRedacted: true });
}
return { ok: snapshotVerification.ok === true && checks.every((item) => item.ok === true), command: "web-probe sentinel migration verify", source: "resolved-active-registry", snapshot: snapshotVerification, reconciliation: snapshotVerification.reconciliation, checks, valuesRedacted: true };
return { ok: snapshotVerification.ok === true && checks.every((item) => item.ok === true), command: "web-probe sentinel migration verify", source: "resolved-active-registry", snapshot: snapshotVerification, reconciliation: { source: reconcileTables(snapshot.tables), target: { runs: expected.length, runtimeMetadata: expectedMetadata.length } }, checks, valuesRedacted: true };
}
export function planMonitorSqliteSources(sources: readonly MonitorSqliteSource[]): Record<string, unknown> {
@@ -87,11 +91,12 @@ export async function importMonitorSqliteSnapshot(snapshot: MonitorSqliteSnapsho
const verified = verifyMonitorSqliteSnapshot(snapshot);
if (verified.ok !== true) throw Object.assign(new Error("snapshot verification failed"), { code: "monitor-migration-snapshot-invalid", verification: verified });
const ingests = canonicalIngests(snapshot);
const runtimeMetadata = canonicalRuntimeMetadata(snapshot);
const manifest = { manifestId: snapshot.manifestContentHash, sourceFingerprint: snapshot.fingerprint, rowCount: snapshot.tables.reduce((total, table) => total + table.rowCount, 0) };
const batch = await store.importSource(manifest, ingests);
const batch = await store.importSource(manifest, ingests, runtimeMetadata);
const dispositions = batch.runs;
const reconciliation = reconcileTables(snapshot.tables);
return { ok: true, command: "web-probe sentinel migration import", source: snapshot.source, manifest, disposition: batch.disposition, imported: { inserted: dispositions.filter((item) => item.disposition === "inserted").length, idempotent: dispositions.filter((item) => item.disposition === "idempotent").length }, reconciliation, payloadHashCount: new Set(ingests.map((item) => item.payloadHash)).size, valuesRedacted: true };
return { ok: true, command: "web-probe sentinel migration import", source: snapshot.source, manifest, disposition: batch.disposition, imported: { inserted: dispositions.filter((item) => item.disposition === "inserted").length, idempotent: dispositions.filter((item) => item.disposition === "idempotent").length }, reconciliation: { source: reconciliation, target: { runs: ingests.length, runtimeMetadata: runtimeMetadata.length } }, payloadHashCount: new Set(ingests.map((item) => item.payloadHash)).size, valuesRedacted: true };
}
export function reconcileTables(tables: readonly MonitorSqliteTable[]): MonitorMigrationReconciliation {
@@ -118,6 +123,8 @@ function canonicalIngests(snapshot: MonitorSqliteSnapshot): readonly ReturnType<
return normalizeMonitorCentralIngest({ sentinelId: stringField(row, ["sentinel_id", "sentinelId"]) || snapshot.source.sentinelId, node: stringField(row, ["node"]) || snapshot.source.node, lane: stringField(row, ["lane"]) || snapshot.source.lane, runId, updatedAt: stringField(row, ["updated_at", "updatedAt", "created_at", "createdAt"]) || snapshot.createdAt, status: stringField(row, ["status"]) || "legacy", payload, findings: canonicalFindings, artifactLocators: reportLocator ? [...rowLocators, reportLocator] : rowLocators, timeline: [] });
});
}
function canonicalRuntimeMetadata(snapshot: MonitorSqliteSnapshot): readonly MonitorRuntimeMetadata[] { return tableRows(snapshot.tables, ["metadata", "run_metadata"]).map((row) => ({ sentinelId: snapshot.source.sentinelId, node: snapshot.source.node, lane: snapshot.source.lane, key: stringField(row, ["key"]), value: jsonValue(row, ["value_json", "value"]), updatedAt: stringField(row, ["updated_at", "updatedAt", "created_at", "createdAt"]) || snapshot.createdAt })).filter((item) => item.key.length > 0).sort((left, right) => left.key.localeCompare(right.key)); }
function sameRun(value: unknown, expected: { readonly sentinelId: string; readonly node: string; readonly lane: string; readonly runId: string; readonly updatedAt: string }): boolean { return isRecord(value) && value.sentinelId === expected.sentinelId && value.node === expected.node && value.lane === expected.lane && value.runId === expected.runId && value.updatedAt === expected.updatedAt; }
function locatorFromRow(row: Record<string, unknown>, source: MonitorSqliteSnapshot): MonitorArtifactLocator { const ownerPvc = stringField(row, ["owner_pvc", "ownerPvc"]); const relativePath = stringField(row, ["relative_path", "relativePath"]); const hash = stringField(row, ["sha256"]); const sizeBytes = numberField(row, ["size_bytes", "sizeBytes"]); if (!ownerPvc || !relativePath || !hash || sizeBytes === null) throw Object.assign(new Error("legacy artifact locator is incomplete"), { code: "monitor-migration-locator-invalid" }); return { ownerNode: stringField(row, ["owner_node", "ownerNode"]) || source.source.node, ownerLane: stringField(row, ["owner_lane", "ownerLane"]) || source.source.lane, ownerPvc, stateDir: stringField(row, ["state_dir", "stateDir"]), relativePath, sha256: hash, sizeBytes, kind: stringField(row, ["kind"]) || "artifact", reachable: row.reachable !== false }; }
function reportLocatorFromRun(row: Record<string, unknown>, snapshot: MonitorSqliteSnapshot): MonitorArtifactLocator | null { const stateDir = stringField(row, ["state_dir", "stateDir"]); const expectedHash = stringField(row, ["report_json_sha256", "reportJsonSha256"]); if (!stateDir && !expectedHash) return null; if (!stateDir || !expectedHash) throw Object.assign(new Error("legacy report locator is incomplete"), { code: "monitor-migration-report-locator-invalid" }); const path = join(snapshot.source.stateRoot, stateDir, "analysis", "report.json"); const bytes = readFileSync(path); const actualHash = `sha256:${sha256(bytes)}`; if (actualHash !== expectedHash) throw Object.assign(new Error("legacy report hash does not match artifact"), { code: "monitor-migration-report-locator-invalid" }); return { ownerNode: snapshot.source.node, ownerLane: snapshot.source.lane, ownerPvc: snapshot.source.ownerPvc, stateDir, relativePath: "analysis/report.json", sha256: actualHash, sizeBytes: bytes.byteLength, kind: "report-json", reachable: true }; }
function metadataValue(rows: readonly Record<string, unknown>[], key: string): Record<string, unknown> | null { return jsonField(rows.find((row) => stringField(row, ["key"]) === key), ["value_json", "value"]); }
@@ -130,6 +137,7 @@ function writeSnapshot(path: string, content: string): void { const parent = dir
function quoteIdentifier(value: string): string { return `"${value.replaceAll("\"", "\"\"")}"`; }
function normalizeRow(value: unknown): Record<string, unknown> { const row = value as Record<string, unknown>; return Object.fromEntries(Object.entries(row).map(([key, item]) => [key, item instanceof Uint8Array ? Buffer.from(item).toString("base64") : item])); }
function jsonField(row: Record<string, unknown> | undefined, fields: readonly string[]): Record<string, unknown> | null { if (!row) return null; const value = fields.map((field) => row[field]).find((item) => item !== undefined); if (typeof value === "string") try { const parsed = JSON.parse(value); return parsed && typeof parsed === "object" && !Array.isArray(parsed) ? parsed as Record<string, unknown> : null; } catch { return null; } return value && typeof value === "object" && !Array.isArray(value) ? value as Record<string, unknown> : null; }
function jsonValue(row: Record<string, unknown> | undefined, fields: readonly string[]): unknown { if (!row) return null; const value = fields.map((field) => row[field]).find((item) => item !== undefined); if (typeof value !== "string") return value ?? null; try { return JSON.parse(value); } catch { return value; } }
function stringField(row: Record<string, unknown> | undefined, fields: readonly string[]): string { const value = row === undefined ? null : fields.map((field) => row[field]).find((item) => typeof item === "string" && item.length > 0); return typeof value === "string" ? value : ""; }
function numberField(row: Record<string, unknown>, fields: readonly string[]): number | null { const value = fields.map((field) => row[field]).find((item) => typeof item === "number"); return typeof value === "number" && Number.isSafeInteger(value) && value >= 0 ? value : null; }
function sha256(value: Uint8Array): string { return createHash("sha256").update(value).digest("hex"); }