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

fix: preserve immutable monitor migration sources

Browse Source
This commit is contained in:
AgentRun Codex
2026-07-13 00:12:35 +00:00
parent 1a3e95e8d9
commit 9a32ead76f
9 changed files with 130 additions and 29 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scripts/src/monitor-sqlite-migration.ts
+15 -6
View File
@@ -9,7 +9,8 @@ import type { MonitorArtifactLocator, MonitorCentralIngest, MonitorCentralStore,
import { monitorCentralStableJson, normalizeMonitorCentralIngest } from "./monitor-central-persistence";
export const MONITOR_SQLITE_MIGRATION_SCHEMA = "2026-07-12-p17-2";
export interface MonitorSqliteSource { readonly sentinelId: string; readonly node: string; readonly lane: string; readonly path: string; readonly runtimeConfigRef: string; readonly stateRoot: string; readonly ownerPvc: string; }
export interface MonitorArtifactSource { readonly ownerPvc: string; readonly mountPath: string; readonly hostPath?: string; }
export interface MonitorSqliteSource { readonly sentinelId: string; readonly node: string; readonly lane: string; readonly path: string; readonly runtimeConfigRef: string; readonly stateRoot: string; readonly ownerPvc: string; readonly artifactSources: readonly MonitorArtifactSource[]; }
export interface MonitorSqliteTable { readonly name: string; readonly sql: string; readonly rowCount: number; readonly rows: readonly Record<string, unknown>[]; }
export interface MonitorSqliteSnapshot {
readonly schema: string; readonly createdAt: string; readonly source: MonitorSqliteSource; readonly fingerprint: string; readonly integrity: string; readonly byteLength: number;
@@ -59,7 +60,6 @@ export function planMonitorSqliteSources(sources: readonly MonitorSqliteSource[]
export function exportMonitorSqliteSnapshot(source: MonitorSqliteSource, outputPath: string): MonitorSqliteSnapshot {
const path = assertReadOnlySqliteSource(source.path);
const sourceBytes = readFileSync(path);
const database = new Database(path, { readonly: true });
let transactionOpen = false;
try {
@@ -68,7 +68,10 @@ export function exportMonitorSqliteSnapshot(source: MonitorSqliteSource, outputP
const integrity = String(database.query("PRAGMA integrity_check").get()?.integrity_check ?? "unknown");
if (integrity !== "ok") throw new Error(`legacy SQLite integrity_check failed: ${integrity}`);
const tables = (database.query("SELECT name, sql FROM sqlite_master WHERE type = 'table' AND name NOT LIKE 'sqlite_%' ORDER BY name").all() as { name: string; sql: string }[]).map((table) => ({ name: table.name, sql: table.sql, rowCount: Number(database.query(`SELECT COUNT(*) AS count FROM ${quoteIdentifier(table.name)}`).get()?.count ?? 0), rows: database.query(`SELECT * FROM ${quoteIdentifier(table.name)}`).all().map(normalizeRow) }));
const unsigned = { schema: MONITOR_SQLITE_MIGRATION_SCHEMA, createdAt: new Date().toISOString(), source: { ...source, path }, fingerprint: `sha256:${sha256(sourceBytes)}`, integrity, byteLength: sourceBytes.byteLength, tables, artifactLocator: { path, sha256: `sha256:${sha256(sourceBytes)}`, sizeBytes: sourceBytes.byteLength, kind: "legacy-sqlite" } };
const immutablePath = sqliteSnapshotPath(outputPath);
const immutableBytes = (database as unknown as { serialize(): Uint8Array }).serialize();
writeFileSync(immutablePath, immutableBytes);
const unsigned = { schema: MONITOR_SQLITE_MIGRATION_SCHEMA, createdAt: new Date().toISOString(), source: { ...source, path }, fingerprint: `sha256:${sha256(immutableBytes)}`, integrity, byteLength: immutableBytes.byteLength, tables, artifactLocator: { path: immutablePath, sha256: `sha256:${sha256(immutableBytes)}`, sizeBytes: immutableBytes.byteLength, kind: "legacy-sqlite" } };
const snapshot: MonitorSqliteSnapshot = { ...unsigned, manifestContentHash: monitorSnapshotContentHash(unsigned) };
writeSnapshot(outputPath, `${monitorCentralStableJson(snapshot)}\n`);
database.exec("COMMIT");
@@ -120,14 +123,17 @@ function canonicalIngests(snapshot: MonitorSqliteSnapshot): readonly ReturnType<
const payload = { ...(storedPayload ?? {}), ...(report ?? {}), legacy: { run: row, sourceFingerprint: snapshot.fingerprint } };
const reportFindings = Array.isArray(report?.findings) ? report.findings.filter(isRecord) : [];
const canonicalFindings = dedupeFindings(reportFindings, rowFindings);
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: [] });
const artifactLocators = reportLocator ? [...rowLocators, reportLocator] : rowLocators;
const artifactCount = numberField(row, ["artifact_count", "artifactCount"]);
if (artifactCount !== null && artifactCount !== artifactLocators.length) throw Object.assign(new Error(`legacy artifact locator count mismatch for ${runId}`), { code: "monitor-migration-artifact-count-mismatch" });
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, timeline: [] });
});
}
function targetReconciliation(ingests: readonly ReturnType<typeof normalizeMonitorCentralIngest>[], runtimeMetadata: readonly MonitorRuntimeMetadata[]): Record<string, number> { return { runs: ingests.length, findings: ingests.reduce((total, item) => total + item.findings.length, 0), payloads: ingests.length, artifactLocators: ingests.reduce((total, item) => total + item.artifactLocators.length, 0), views: ingests.reduce((total, item) => total + (isRecord(item.payload.views) ? Object.keys(item.payload.views).length : 0), 0), runtimeMetadata: runtimeMetadata.length }; }
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 locatorFromRow(row: Record<string, unknown>, snapshot: MonitorSqliteSnapshot): MonitorArtifactLocator { const ownerPvc = stringField(row, ["owner_pvc", "ownerPvc"]); const relativePath = stringField(row, ["relative_path", "relativePath"]); const expectedHash = stringField(row, ["sha256"]); const expectedSize = numberField(row, ["size_bytes", "sizeBytes"]); const stateDir = stringField(row, ["state_dir", "stateDir"]); if (!ownerPvc || !stateDir || !relativePath || !expectedHash || expectedSize === null) throw Object.assign(new Error("legacy artifact locator is incomplete"), { code: "monitor-migration-locator-invalid" }); const source = artifactSourceFor(snapshot.source, ownerPvc); const path = artifactPath(source.mountPath, stateDir, relativePath); const bytes = readFileSync(path); const actualHash = `sha256:${sha256(bytes)}`; if (actualHash !== expectedHash || bytes.byteLength !== expectedSize) throw Object.assign(new Error("legacy artifact locator does not match source artifact"), { code: "monitor-migration-locator-invalid" }); return { ownerNode: stringField(row, ["owner_node", "ownerNode"]) || snapshot.source.node, ownerLane: stringField(row, ["owner_lane", "ownerLane"]) || snapshot.source.lane, ownerPvc, stateDir, relativePath, sha256: actualHash, sizeBytes: bytes.byteLength, 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 ownerPvc = stringField(row, ["owner_pvc", "ownerPvc", "artifact_owner_pvc", "artifactOwnerPvc"]) || snapshot.source.ownerPvc; const source = artifactSourceFor(snapshot.source, ownerPvc); const path = artifactPath(source.mountPath, 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, 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"]); }
function normalizeLegacyFinding(row: Record<string, unknown>): Record<string, unknown> { return jsonField(row, ["finding", "value_json"]) ?? { id: stringField(row, ["finding_id", "id"]), severity: stringField(row, ["severity"]), count: numberField(row, ["count"]) ?? 1, summary: stringField(row, ["summary"]), reportJsonSha256: stringField(row, ["report_json_sha256", "reportJsonSha256"]), createdAt: stringField(row, ["created_at", "createdAt"]) }; }
function dedupeFindings(primary: readonly Record<string, unknown>[], secondary: readonly Record<string, unknown>[]): readonly Record<string, unknown>[] { const seen = new Set<string>(); return [...primary, ...secondary].filter((finding) => { const id = stringField(finding, ["id"]); const key = id || monitorCentralStableJson(finding); if (seen.has(key)) return false; seen.add(key); return true; }); }
@@ -135,6 +141,9 @@ function tableRows(tables: readonly MonitorSqliteTable[], names: readonly string
function sourcePlan(source: MonitorSqliteSource): Record<string, unknown> { const path = resolve(source.path); const state = existsSync(path) ? statSync(path) : null; return { sentinelId: source.sentinelId, node: source.node, lane: source.lane, path, runtimeConfigRef: source.runtimeConfigRef, exists: state !== null, isFile: state?.isFile() ?? false, byteLength: state?.size ?? null, mutation: false, valuesRedacted: true }; }
function assertReadOnlySqliteSource(value: string): string { const path = resolve(value); const state = statSync(path); if (!state.isFile()) throw new Error(`legacy SQLite source is not a file: ${path}`); const descriptor = openSync(path, "r"); closeSync(descriptor); return path; }
function writeSnapshot(path: string, content: string): void { const parent = dirname(resolve(path)); if (!existsSync(parent)) throw new Error(`snapshot output directory does not exist: ${parent}`); writeFileSync(path, content); }
function sqliteSnapshotPath(path: string): string { const resolved = resolve(path); return resolved.endsWith(".json") ? `${resolved.slice(0, -5)}.sqlite` : `${resolved}.sqlite`; }
function artifactSourceFor(source: MonitorSqliteSource, ownerPvc: string): MonitorArtifactSource { const artifactSource = source.artifactSources.find((item) => item.ownerPvc === ownerPvc); if (!artifactSource) throw Object.assign(new Error(`legacy artifact owner is not registry-mounted: ${ownerPvc}`), { code: "monitor-migration-artifact-source-missing" }); return artifactSource; }
function artifactPath(mountPath: string, stateDir: string, relativePath: string): string { const root = resolve(mountPath); const directory = stateDir.startsWith("/") ? resolve(stateDir) : resolve(root, stateDir); if (directory !== root && !directory.startsWith(`${root}/`)) throw Object.assign(new Error("legacy artifact stateDir escapes registry mount"), { code: "monitor-migration-artifact-path-invalid" }); return join(directory, relativePath); }
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; }