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

fix: freeze migration artifact manifests

Browse Source
This commit is contained in:
AgentRun Codex
2026-07-13 00:32:36 +00:00
parent 3a63da0fa8
commit 036b29b59b
2 changed files with 17 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scripts/src/monitor-sqlite-migration.ts
+12 -9
View File
@@ -8,13 +8,15 @@ import { Database } from "bun:sqlite";
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";
export const MONITOR_SQLITE_MIGRATION_SCHEMA = "2026-07-12-p17-3";
export interface MonitorArtifactSource { readonly ownerPvc: string; readonly mountPath: string; readonly hostPath?: string; }
function freezeArtifactManifests(snapshot: Pick<MonitorSqliteSnapshot, "source" | "tables">): readonly { readonly runId: string; readonly locators: readonly MonitorArtifactLocator[] }[] { const rows = tableRows(snapshot.tables, ["runs", "run"]); const locatorRows = tableRows(snapshot.tables, ["artifact_locators", "locators"]); return rows.map((row) => { const runId = stringField(row, ["run_id", "id"]); const rowLocators = locatorRows.filter((item) => stringField(item, ["run_id"]) === runId).map((item) => locatorFromRow(item, snapshot)); const locators = rowLocators.length > 0 ? rowLocators : artifactLocatorsFromRun(row, snapshot); const expectedCount = numberField(row, ["artifact_count", "artifactCount"]); if (expectedCount !== null && expectedCount !== locators.length) throw artifactCountMismatch(runId, stringField(row, ["state_dir", "stateDir"]), expectedCount, locators.length); verifyReportLocator(row, locators); return { runId, locators }; }); }
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;
readonly tables: readonly MonitorSqliteTable[];
readonly artifactManifests: readonly { readonly runId: string; readonly locators: readonly MonitorArtifactLocator[] }[];
readonly artifactLocator: { readonly path: string; readonly sha256: string; readonly sizeBytes: number; readonly kind: "legacy-sqlite" };
readonly manifestContentHash: string;
}
@@ -22,6 +24,7 @@ export interface MonitorMigrationReconciliation { readonly runs: number; readonl
export async function verifyMonitorSqliteSnapshotAgainstStore(snapshot: MonitorSqliteSnapshot, store: MonitorCentralStore): Promise<Record<string, unknown>> {
const snapshotVerification = verifyMonitorSqliteSnapshot(snapshot);
verifyFrozenArtifactManifests(snapshot);
const expected = canonicalIngests(snapshot);
const expectedMetadata = canonicalRuntimeMetadata(snapshot);
const checks: Record<string, unknown>[] = [];
@@ -71,7 +74,8 @@ export function exportMonitorSqliteSnapshot(source: MonitorSqliteSource, outputP
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 artifactManifests = freezeArtifactManifests({ source: { ...source, path }, tables });
const unsigned = { schema: MONITOR_SQLITE_MIGRATION_SCHEMA, createdAt: new Date().toISOString(), source: { ...source, path }, fingerprint: `sha256:${sha256(immutableBytes)}`, integrity, byteLength: immutableBytes.byteLength, tables, artifactManifests, 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");
@@ -93,6 +97,7 @@ export function verifyMonitorSqliteSnapshot(snapshot: MonitorSqliteSnapshot): Re
export async function importMonitorSqliteSnapshot(snapshot: MonitorSqliteSnapshot, store: MonitorCentralStore): Promise<Record<string, unknown>> {
const verified = verifyMonitorSqliteSnapshot(snapshot);
if (verified.ok !== true) throw Object.assign(new Error("snapshot verification failed"), { code: "monitor-migration-snapshot-invalid", verification: verified });
verifyFrozenArtifactManifests(snapshot);
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) };
@@ -112,27 +117,23 @@ function canonicalIngests(snapshot: MonitorSqliteSnapshot): readonly ReturnType<
const findings = tableRows(snapshot.tables, ["findings", "finding"]);
const metadata = tableRows(snapshot.tables, ["metadata", "run_metadata"]);
const payloads = tableRows(snapshot.tables, ["payloads", "run_payloads"]);
const locators = tableRows(snapshot.tables, ["artifact_locators", "locators"]);
return rows.map((row) => {
const runId = stringField(row, ["run_id", "id"]);
const rowFindings = findings.filter((item) => stringField(item, ["run_id"]) === runId).map(normalizeLegacyFinding);
const report = metadataValue(metadata, `run.report.${runId}`);
const rowLocators = locators.filter((item) => stringField(item, ["run_id"]) === runId).map((item) => locatorFromRow(item, snapshot));
const storedPayload = jsonField(payloads.find((item) => stringField(item, ["run_id"]) === runId), ["payload", "value_json"]);
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);
const artifactLocators = rowLocators.length > 0 ? rowLocators : artifactLocatorsFromRun(row, snapshot);
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", diagnostic: { runId, stateDir: stringField(row, ["state_dir", "stateDir"]), expectedCount: artifactCount, observedCount: artifactLocators.length, valuesRedacted: true } });
verifyReportLocator(row, artifactLocators);
const artifactLocators = snapshot.artifactManifests.find((item) => item.runId === runId)?.locators;
if (artifactLocators === undefined) throw Object.assign(new Error(`snapshot artifact manifest is missing ${runId}`), { code: "monitor-migration-artifact-manifest-missing" });
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>, 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 locatorFromRow(row: Record<string, unknown>, snapshot: Pick<MonitorSqliteSnapshot, "source">): 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 artifactLocatorsFromRun(row: Record<string, unknown>, snapshot: MonitorSqliteSnapshot): readonly MonitorArtifactLocator[] { const stateDir = stringField(row, ["state_dir", "stateDir"]); if (!stateDir) return []; const source = artifactSourceForStateDir(snapshot.source, stateDir); const directory = artifactPath(source.mountPath, stateDir, ""); if (!existsSync(directory) || !statSync(directory).isDirectory()) throw Object.assign(new Error("legacy artifact stateDir is unavailable"), { code: "monitor-migration-artifact-path-invalid" }); return artifactFiles(directory).map((path) => { const bytes = readFileSync(path); const relativePath = path.slice(directory.length + 1).replaceAll("\\", "/"); return { ownerNode: snapshot.source.node, ownerLane: snapshot.source.lane, ownerPvc: source.ownerPvc, stateDir, relativePath, sha256: `sha256:${sha256(bytes)}`, sizeBytes: bytes.byteLength, kind: relativePath === "analysis/report.json" ? "report-json" : "artifact", reachable: true }; }); }
function verifyReportLocator(row: Record<string, unknown>, locators: readonly MonitorArtifactLocator[]): void { const expectedHash = stringField(row, ["report_json_sha256", "reportJsonSha256"]); if (!expectedHash) return; const report = locators.find((locator) => locator.relativePath === "analysis/report.json"); if (!report || report.sha256 !== expectedHash) throw Object.assign(new Error("legacy report hash does not match artifact"), { code: "monitor-migration-report-locator-invalid" }); }
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"]); }
@@ -147,6 +148,8 @@ function artifactSourceFor(source: MonitorSqliteSource, ownerPvc: string): Monit
function artifactSourceForStateDir(source: MonitorSqliteSource, stateDir: string): MonitorArtifactSource { const directory = resolve(stateDir); const matches = source.artifactSources.filter((item) => directory === resolve(item.mountPath) || directory.startsWith(`${resolve(item.mountPath)}/`)).sort((left, right) => resolve(right.mountPath).length - resolve(left.mountPath).length); if (matches.length === 0) throw Object.assign(new Error(`legacy artifact stateDir is not registry-mounted: ${stateDir}`), { code: "monitor-migration-artifact-source-missing" }); return matches[0]!; }
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 artifactFiles(directory: string): readonly string[] { return readdirSync(directory, { withFileTypes: true }).flatMap((entry) => { const path = join(directory, entry.name); if (entry.isDirectory()) return artifactFiles(path); return entry.isFile() ? [path] : []; }).sort(); }
function verifyFrozenArtifactManifests(snapshot: MonitorSqliteSnapshot): void { for (const manifest of snapshot.artifactManifests) for (const locator of manifest.locators) { const source = artifactSourceFor(snapshot.source, locator.ownerPvc); const path = artifactPath(source.mountPath, locator.stateDir, locator.relativePath); if (!existsSync(path) || !statSync(path).isFile()) throw Object.assign(new Error(`frozen artifact is unavailable for ${manifest.runId}`), { code: "monitor-migration-frozen-artifact-invalid", diagnostic: { runId: manifest.runId, stateDir: locator.stateDir, expectedCount: manifest.locators.length, observedCount: null, valuesRedacted: true } }); const bytes = readFileSync(path); if (`sha256:${sha256(bytes)}` !== locator.sha256 || bytes.byteLength !== locator.sizeBytes) throw Object.assign(new Error(`frozen artifact changed for ${manifest.runId}`), { code: "monitor-migration-frozen-artifact-invalid", diagnostic: { runId: manifest.runId, stateDir: locator.stateDir, expectedCount: manifest.locators.length, observedCount: manifest.locators.length, valuesRedacted: true } }); } }
function artifactCountMismatch(runId: string, stateDir: string, expectedCount: number, observedCount: number): Error { return Object.assign(new Error(`legacy artifact locator count mismatch for ${runId}`), { code: "monitor-migration-artifact-count-mismatch", diagnostic: { runId, stateDir, expectedCount, observedCount, valuesRedacted: true } }); }
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; }
scripts/src/monitor-terminal-ingest-migration.test.ts
+5 -3
View File
@@ -189,8 +189,11 @@ test("workspace artifact source verifies locator identity hash and size", async
assert.equal(run?.artifactLocators.length, 5);
assert.deepEqual(run?.artifactLocators.map((locator) => ({ ownerPvc: locator.ownerPvc, stateDir: locator.stateDir, relativePath: locator.relativePath, sha256: locator.sha256, sizeBytes: locator.sizeBytes })), artifacts.map(([relativePath, bytes]) => ({ ownerPvc: "legacy-web-observe-workspace-nc01", stateDir, relativePath, sha256: `sha256:${createHash("sha256").update(bytes).digest("hex")}`, sizeBytes: bytes.byteLength })).sort((left, right) => left.relativePath.localeCompare(right.relativePath)));
assert.equal((await verifyMonitorSqliteSnapshotAgainstStore(snapshot, store)).ok, true);
writeFileSync(join(stateDir, "artifacts", "request.txt"), "changed\n");
await assert.rejects(() => importMonitorSqliteSnapshot(snapshot, createInMemoryMonitorCentralStore()), (error: unknown) => (error as { code?: string }).code === "monitor-migration-frozen-artifact-invalid");
writeFileSync(join(stateDir, "artifacts", "request.txt"), artifacts[2][1]);
rmSync(join(stateDir, "artifacts", "response.txt"));
await assert.rejects(() => importMonitorSqliteSnapshot(snapshot, createInMemoryMonitorCentralStore()), (error: unknown) => { const migrationError = error as { code?: string; diagnostic?: Record<string, unknown> }; return migrationError.code === "monitor-migration-artifact-count-mismatch" && migrationError.diagnostic?.runId === "run-1" && migrationError.diagnostic?.stateDir === stateDir && migrationError.diagnostic?.expectedCount === 5 && migrationError.diagnostic?.observedCount === 4; });
await assert.rejects(() => verifyMonitorSqliteSnapshotAgainstStore(snapshot, store), (error: unknown) => { const migrationError = error as { code?: string; diagnostic?: Record<string, unknown> }; return migrationError.code === "monitor-migration-frozen-artifact-invalid" && migrationError.diagnostic?.runId === "run-1" && migrationError.diagnostic?.stateDir === stateDir && migrationError.diagnostic?.expectedCount === 5; });
} finally {
rmSync(directory, { recursive: true, force: true });
}
@@ -206,8 +209,7 @@ test("artifact count without a registry-backed locator fails closed", async () =
database.run("INSERT INTO runs VALUES (?, ?, ?)", ["run-1", 1, "2026-07-12T00:00:00.000Z"]);
database.close();
const source = { sentinelId: "fixture", node: "NC01", lane: "v03", path: sqlitePath, runtimeConfigRef: "fixture#runtime", stateRoot: directory, ownerPvc: "fixture-pvc", artifactSources: [{ ownerPvc: "fixture-pvc", mountPath: directory }] };
const snapshot = exportMonitorSqliteSnapshot(source, snapshotPath);
await assert.rejects(() => importMonitorSqliteSnapshot(snapshot, createInMemoryMonitorCentralStore()), (error: unknown) => (error as { code?: string }).code === "monitor-migration-artifact-count-mismatch");
assert.throws(() => exportMonitorSqliteSnapshot(source, snapshotPath), (error: unknown) => (error as { code?: string }).code === "monitor-migration-artifact-count-mismatch");
} finally {
rmSync(directory, { recursive: true, force: true });
}