feat: add monitor terminal ingest migration CLI

scripts/src/hwlab-node-web-sentinel-cicd-shared.ts

@@ -16,6 +16,7 @@ export type WebProbeSentinelPublishAction = "publish-current";
 export type WebProbeSentinelMaintenanceAction = "status" | "start" | "stop";
 export type WebProbeSentinelDashboardAction = "verify" | "screenshot" | "trigger";
 export type WebProbeSentinelReportView = "summary" | "turn-summary" | "findings" | "trace-frame" | "auth-session-switch-summary";
+export type WebProbeSentinelMigrationAction = "plan" | "export" | "import" | "verify";
 export type WebProbeSentinelSourceAuthority = "git-mirror-snapshot" | "gitea-snapshot";
 
 export interface WebProbeSentinelSourceOverrideOptions {
@@ -26,6 +27,17 @@ export interface WebProbeSentinelSourceOverrideOptions {
 }
 
 export type WebProbeSentinelOptions =
+  | {
+    readonly kind: "migration";
+    readonly action: WebProbeSentinelMigrationAction;
+    readonly node: string;
+    readonly lane: string;
+    readonly sentinelId: string | null;
+    readonly sourcePath: string | null;
+    readonly snapshotPath: string | null;
+    readonly confirm: boolean;
+    readonly json: boolean;
+  }
   | {
     readonly kind: "config";
     readonly action: WebProbeSentinelConfigAction;

scripts/src/hwlab-node-web-sentinel-cicd.ts

@@ -36,6 +36,7 @@ import type { RenderedCliResult } from "./output";
 import { probeSentinelRuntimeHealthEndpoint, runSentinelDashboard, runSentinelErrorDetail, runSentinelInspect, runSentinelMaintenance, runSentinelReport, runSentinelValidate } from "./hwlab-node-web-sentinel-p5";
 import { runChildCli, sentinelP5Next } from "./hwlab-node-web-sentinel-p5-observe";
 import { emitWebProbeSentinelSpan, webProbeSentinelOtelSummary } from "./hwlab-node-web-sentinel-otel";
+import { exportMonitorSqliteSnapshot, planMonitorSqliteSources, verifyMonitorSqliteSnapshot, type MonitorSqliteSource } from "./monitor-sqlite-migration";
 import {
   arrayAt,
   arrayAtNullable,
@@ -161,6 +162,7 @@ const SPEC_REF = "PJ2026-01060508 Web哨兵 draft-2026-07-01-p16-cicd-source-sna
 export type SourceResolveMode = "cached" | "sync";
 
 export function runWebProbeSentinelCommand(spec: HwlabRuntimeLaneSpec, options: WebProbeSentinelOptions): RenderedCliResult {
+  if (options.kind === "migration") return runSentinelMigration(spec, options);
   if (options.kind === "config") return withWebProbeSentinelConfigRendered(webProbeSentinelConfigPlan(spec, options.action, options.sentinelId));
   const guardedAction = guardedSentinelDeliveryAction(options);
   if (guardedAction !== null) {
@@ -189,6 +191,37 @@ export function runWebProbeSentinelCommand(spec: HwlabRuntimeLaneSpec, options:
   return runSentinelReport(state, options);
 }
 
+function runSentinelMigration(spec: HwlabRuntimeLaneSpec, options: Extract<WebProbeSentinelOptions, { kind: "migration" }>): RenderedCliResult {
+  const sentinel = resolveWebProbeSentinel(spec, options.sentinelId);
+  const runtime = recordTarget(readWebProbeSentinelConfigRefTarget(spec, sentinel.configRefs.runtime), sentinel.configRefs.runtime);
+  const source: MonitorSqliteSource = { sentinelId: sentinel.id, node: spec.nodeId, lane: spec.lane, path: options.sourcePath ?? stringAt(runtime, "sqlite.path"), runtimeConfigRef: sentinel.configRefs.runtime };
+  const command = `web-probe sentinel migration ${options.action}`;
+  if (options.action === "plan") {
+    const projection = planMonitorSqliteSources([source]);
+    return rendered(true, command, renderMigrationCompact(projection), projection);
+  }
+  if (options.action === "export") {
+    if (options.snapshotPath === null) throw new Error("web-probe sentinel migration export requires --snapshot");
+    const snapshot = exportMonitorSqliteSnapshot(source, options.snapshotPath);
+    const verified = verifyMonitorSqliteSnapshot(snapshot);
+    const projection = { ...verified, command, snapshot: { fingerprint: snapshot.fingerprint, integrity: snapshot.integrity, createdAt: snapshot.createdAt }, valuesRedacted: true };
+    return rendered(verified.ok === true, command, renderMigrationCompact(projection), projection);
+  }
+  if (options.snapshotPath === null) throw new Error(`web-probe sentinel migration ${options.action} requires --snapshot`);
+  const snapshot = JSON.parse(readFileSync(options.snapshotPath, "utf8")) as Parameters<typeof verifyMonitorSqliteSnapshot>[0];
+  const verification = verifyMonitorSqliteSnapshot(snapshot);
+  if (options.action === "import") {
+    const projection = { ...verification, command, mutation: true, imported: false, blocker: { code: "monitor-migration-import-target-unconfigured", reason: "Host PostgreSQL target is intentionally not inferred; provide the controlled monitor import configuration before execution." }, valuesRedacted: true };
+    return rendered(false, command, renderMigrationCompact(projection), projection);
+  }
+  return rendered(verification.ok === true, command, renderMigrationCompact(verification), verification);
+}
+
+function renderMigrationCompact(value: Record<string, unknown>): string {
+  const reconciliation = record(value.reconciliation);
+  return [`ok=${value.ok === true}`, `command=${String(value.command ?? "web-probe sentinel migration")}`, `runs=${String(reconciliation.runs ?? "-")}`, `findings=${String(reconciliation.findings ?? "-")}`, `metadata=${String(reconciliation.metadata ?? "-")}`, `payloads=${String(reconciliation.payloads ?? "-")}`, `locators=${String(reconciliation.locators ?? "-")}`, `latest=${String(reconciliation.latest ?? "-")}`].join(" ");
+}
+
 function runSentinelImage(state: SentinelCicdState, options: Extract<WebProbeSentinelOptions, { kind: "image" }>): RenderedCliResult {
   const command = `web-probe sentinel image ${options.action}`;
   const deliveryAuthority = webProbeSentinelDeliveryAuthority(state.spec);

scripts/src/hwlab-node/web-probe-observe.ts

@@ -58,6 +58,7 @@ export function parseNodeWebProbeSentinelOptions(args: string[]): NodeWebProbeSe
     && sentinelActionRaw !== "dashboard"
     && sentinelActionRaw !== "error"
     && sentinelActionRaw !== "report"
+    && sentinelActionRaw !== "migration"
     && sentinelActionRaw !== "inspect-url"
     && sentinelActionRaw !== "inspect-id"
   ) {
@@ -87,6 +88,8 @@ export function parseNodeWebProbeSentinelOptions(args: string[]): NodeWebProbeSe
     "--source-stage-ref",
     "--source-mirror-commit",
     "--source-authority",
+    "--source-path",
+    "--snapshot",
   ]), new Set(["--dry-run", "--confirm", "--wait", "--local", "--rerun", "--manual-recovery", "--recovery", "--quick-verify", "--raw", "--full", "--latest", "--full-page", "--no-full-page", "--diagnose-monitor-web"]));
   const nodeOption = optionValue(args, "--node") ?? null;
   const laneOption = optionValue(args, "--lane") ?? null;
@@ -216,6 +219,13 @@ export function parseNodeWebProbeSentinelOptions(args: string[]): NodeWebProbeSe
     const errorId = requiredOption(args, "--error-id");
     if (!/^wbe_[a-f0-9]{20}$/u.test(errorId)) throw new Error(`web-probe sentinel error get --error-id must look like wbe_<20 hex>, got ${errorId}`);
     sentinel = { kind: "error", action: "get", node, lane, sentinelId, errorId, raw: args.includes("--raw"), full: args.includes("--full"), timeoutSeconds };
+  } else if (sentinelActionRaw === "migration") {
+    const migrationAction = args[1];
+    if (migrationAction !== "plan" && migrationAction !== "export" && migrationAction !== "import" && migrationAction !== "verify") {
+      throw new Error("web-probe sentinel migration usage: migration plan|export|import|verify --node NODE --lane vNN [--sentinel ID] [--source-path PATH] [--snapshot PATH] [--confirm] [--json]");
+    }
+    if ((migrationAction === "import" || migrationAction === "export") && !confirm) throw new Error(`web-probe sentinel migration ${migrationAction} requires --confirm`);
+    sentinel = { kind: "migration", action: migrationAction, node, lane, sentinelId, sourcePath: optionValue(args, "--source-path") ?? null, snapshotPath: optionValue(args, "--snapshot") ?? null, confirm, json: args.includes("--json") };
   } else if (sentinelActionRaw === "report") {
     const view = parseWebProbeSentinelReportView(optionValue(args, "--view") ?? "summary");
     const latest = args.includes("--latest");

scripts/src/monitor-sqlite-migration.ts

@@ -0,0 +1,146 @@
+// SPEC: PJ2026-01060508 Web哨兵 draft-2026-07-12-p17-monitor-central-persistence.
+// Responsibility: Read-only legacy SQLite snapshot export and bounded, explicit one-shot migration verification.
+import { createHash } from "node:crypto";
+import { closeSync, existsSync, openSync, readFileSync, statSync, writeFileSync } from "node:fs";
+import { dirname, resolve } from "node:path";
+import { Database } from "bun:sqlite";
+
+import { stableJson } from "./monitor-terminal-ingest";
+
+export const MONITOR_SQLITE_MIGRATION_SCHEMA = "2026-07-12-p17-1";
+
+export interface MonitorSqliteSource {
+  readonly sentinelId: string;
+  readonly node: string;
+  readonly lane: string;
+  readonly path: string;
+  readonly runtimeConfigRef: string;
+}
+
+export interface MonitorSqliteSnapshot {
+  readonly schema: string;
+  readonly createdAt: string;
+  readonly source: MonitorSqliteSource;
+  readonly fingerprint: string;
+  readonly integrity: string;
+  readonly byteLength: number;
+  readonly tables: readonly { readonly name: string; readonly sql: string; readonly rowCount: number; readonly rows: readonly Record<string, unknown>[] }[];
+  readonly artifactLocator: { readonly path: string; readonly sha256: string; readonly byteLength: number };
+}
+
+export interface MonitorMigrationReconciliation {
+  readonly runs: number;
+  readonly findings: number;
+  readonly metadata: number;
+  readonly payloads: number;
+  readonly locators: number;
+  readonly latest: number;
+}
+
+export function planMonitorSqliteSources(sources: readonly MonitorSqliteSource[]): Record<string, unknown> {
+  const rows = sources.map((source) => {
+    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,
+    };
+  });
+  return { ok: true, command: "web-probe sentinel migration plan", sources: rows, sourceCount: rows.length, valuesRedacted: true };
+}
+
+export function exportMonitorSqliteSnapshot(source: MonitorSqliteSource, outputPath: string): MonitorSqliteSnapshot {
+  const path = assertReadOnlySqliteSource(source.path);
+  const bytes = readFileSync(path);
+  const fingerprint = `sha256:${sha256(bytes)}`;
+  const database = new Database(path, { readonly: true });
+  try {
+    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 }[];
+    const snapshot: MonitorSqliteSnapshot = {
+      schema: MONITOR_SQLITE_MIGRATION_SCHEMA,
+      createdAt: new Date().toISOString(),
+      source: { ...source, path },
+      fingerprint,
+      integrity,
+      byteLength: bytes.byteLength,
+      tables: tables.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),
+      })),
+      artifactLocator: { path: resolve(outputPath), sha256: "", byteLength: 0 },
+    };
+    const locatorSha256 = `sha256:${sha256(Buffer.from(stableJson({ ...snapshot, artifactLocator: { ...snapshot.artifactLocator, sha256: "", byteLength: 0 } })))}`;
+    const materialized = `${stableJson({ ...snapshot, artifactLocator: { ...snapshot.artifactLocator, sha256: locatorSha256, byteLength: 0 } })}\n`;
+    writeSnapshot(outputPath, materialized);
+    const artifact = readFileSync(outputPath);
+    return { ...snapshot, artifactLocator: { path: resolve(outputPath), sha256: locatorSha256, byteLength: artifact.byteLength } };
+  } finally {
+    database.close();
+  }
+}
+
+export function verifyMonitorSqliteSnapshot(snapshot: MonitorSqliteSnapshot): Record<string, unknown> {
+  const expected = `sha256:${sha256(Buffer.from(stableJson({ ...snapshot, artifactLocator: { ...snapshot.artifactLocator, sha256: "", byteLength: 0 } })))} ` .trim();
+  const reconciliation = reconcileTables(snapshot.tables);
+  return {
+    ok: snapshot.schema === MONITOR_SQLITE_MIGRATION_SCHEMA && snapshot.integrity === "ok" && snapshot.artifactLocator.sha256 === expected,
+    command: "web-probe sentinel migration verify",
+    fingerprint: snapshot.fingerprint,
+    integrity: snapshot.integrity,
+    reconciliation,
+    artifactLocator: { ...snapshot.artifactLocator, valid: snapshot.artifactLocator.sha256 === expected },
+    valuesRedacted: true,
+  };
+}
+
+export function reconcileTables(tables: readonly MonitorSqliteSnapshot["tables"][number][]): MonitorMigrationReconciliation {
+  const count = (names: readonly string[]) => tables.filter((table) => names.includes(table.name)).reduce((total, table) => total + table.rowCount, 0);
+  return {
+    runs: count(["runs", "run"]),
+    findings: count(["findings", "finding"]),
+    metadata: count(["metadata", "run_metadata"]),
+    payloads: count(["payloads", "run_payloads"]),
+    locators: count(["artifact_locators", "locators"]),
+    latest: count(["latest", "latest_runs"]),
+  };
+}
+
+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 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 sha256(value: Uint8Array): string {
+  return createHash("sha256").update(value).digest("hex");
+}

scripts/src/monitor-terminal-ingest-migration.test.ts

@@ -0,0 +1,49 @@
+// SPEC: PJ2026-01060508 Web哨兵 draft-2026-07-12-p17-monitor-central-persistence.
+import assert from "node:assert/strict";
+import { mkdtempSync, readFileSync, rmSync, writeFileSync } from "node:fs";
+import { tmpdir } from "node:os";
+import { join } from "node:path";
+import test from "node:test";
+import { Database } from "bun:sqlite";
+
+import { exportMonitorSqliteSnapshot, planMonitorSqliteSources, verifyMonitorSqliteSnapshot } from "./monitor-sqlite-migration";
+import { buildMonitorTerminalIngest, MonitorIngestFailedError, submitMonitorTerminalIngest } from "./monitor-terminal-ingest";
+
+test("terminal ingest payload is immutable and retries bounded failures", async () => {
+  const report = { ok: false, findings: [{ id: "blocked" }], updatedAt: "2026-07-12T00:00:00.000Z" };
+  const ingest = buildMonitorTerminalIngest({ sentinelId: "fixture", node: "NC01", lane: "v03", runId: "run-1" }, report, [{ relativePath: "analysis/report.json", kind: "report", sha256: "sha256:report", sizeBytes: 42 }], { ownerNode: "NC01", ownerLane: "v03", ownerPvc: "fixture-pvc", stateDir: "/state" });
+  report.findings[0].id = "changed";
+  assert.equal((ingest.payload.report as { findings: { id: string }[] }).findings[0].id, "blocked");
+  assert.match(ingest.payloadHash ?? "", /^sha256:/u);
+  await assert.rejects(
+    () => submitMonitorTerminalIngest({ endpoint: "https://monitor.example.test/api/ingest", timeoutMs: 20, retry: { maxAttempts: 2, delayMs: 0 } }, ingest, async () => new Response("{}", { status: 503 })),
+    (error: unknown) => error instanceof MonitorIngestFailedError && error.details.attempts === 2,
+  );
+});
+
+test("SQLite snapshot reconciles the 228/242/234 fixture shape without production assumptions", () => {
+  const directory = mkdtempSync(join(tmpdir(), "monitor-migration-"));
+  try {
+    const sqlitePath = join(directory, "legacy.sqlite");
+    const snapshotPath = join(directory, "snapshot.json");
+    const database = new Database(sqlitePath);
+    database.run("CREATE TABLE runs (id TEXT)");
+    database.run("CREATE TABLE findings (id TEXT)");
+    database.run("CREATE TABLE metadata (id TEXT)");
+    for (let index = 0; index < 228; index += 1) database.run("INSERT INTO runs VALUES (?)", [`run-${index}`]);
+    for (let index = 0; index < 242; index += 1) database.run("INSERT INTO findings VALUES (?)", [`finding-${index}`]);
+    for (let index = 0; index < 234; index += 1) database.run("INSERT INTO metadata VALUES (?)", [`metadata-${index}`]);
+    database.close();
+    const source = { sentinelId: "fixture", node: "NC01", lane: "v03", path: sqlitePath, runtimeConfigRef: "fixture#runtime" };
+    const plan = planMonitorSqliteSources([source]);
+    assert.equal((plan.sources as { exists: boolean }[])[0].exists, true);
+    const snapshot = exportMonitorSqliteSnapshot(source, snapshotPath);
+    const verified = verifyMonitorSqliteSnapshot(snapshot);
+    assert.equal(verified.ok, true);
+    assert.deepEqual(verified.reconciliation, { runs: 228, findings: 242, metadata: 234, payloads: 0, locators: 0, latest: 0 });
+    const persisted = JSON.parse(readFileSync(snapshotPath, "utf8"));
+    assert.equal(persisted.schema, snapshot.schema);
+  } finally {
+    rmSync(directory, { recursive: true, force: true });
+  }
+});

scripts/src/monitor-terminal-ingest.ts

@@ -0,0 +1,131 @@
+// SPEC: PJ2026-01060508 Web哨兵 draft-2026-07-12-p17-monitor-central-persistence.
+// Responsibility: Build immutable terminal monitor payloads and submit them with YAML-declared bounded retries.
+import { createHash } from "node:crypto";
+
+import type { MonitorArtifactLocator, MonitorCentralIngest } from "./monitor-central-persistence";
+
+export interface MonitorIngestRetry {
+  readonly maxAttempts: number;
+  readonly delayMs: number;
+}
+
+export interface MonitorTerminalIngestConfig {
+  readonly endpoint: string;
+  readonly timeoutMs: number;
+  readonly retry: MonitorIngestRetry;
+}
+
+export interface MonitorTerminalIdentity {
+  readonly sentinelId: string;
+  readonly node: string;
+  readonly lane: string;
+  readonly runId: string;
+}
+
+export interface MonitorTerminalArtifact {
+  readonly relativePath: string;
+  readonly kind: string;
+  readonly sha256: string;
+  readonly sizeBytes: number;
+}
+
+export class MonitorIngestFailedError extends Error {
+  readonly code = "monitor-ingest-failed";
+
+  constructor(readonly details: Record<string, unknown>) {
+    super("Monitor terminal ingest failed after bounded retries");
+    this.name = "MonitorIngestFailedError";
+  }
+}
+
+export function buildMonitorTerminalIngest(
+  identity: MonitorTerminalIdentity,
+  report: Record<string, unknown>,
+  artifacts: readonly MonitorTerminalArtifact[],
+  locatorOwner: Pick<MonitorArtifactLocator, "ownerNode" | "ownerLane" | "ownerPvc" | "stateDir">,
+): MonitorCentralIngest {
+  const payload = immutablePayload(report, artifacts);
+  return {
+    ...identity,
+    updatedAt: stringValue(report.updatedAt) ?? stringValue(report.analyzedAt) ?? new Date().toISOString(),
+    status: stringValue(report.status) ?? (report.ok === true ? "ok" : "blocked"),
+    payload,
+    payloadHash: `sha256:${sha256(stableJson(payload))}`,
+    findings: recordArray(report.findings),
+    timeline: recordArray(report.timeline),
+    artifactLocators: artifacts.map((artifact) => ({
+      ...locatorOwner,
+      relativePath: artifact.relativePath,
+      kind: artifact.kind,
+      sha256: artifact.sha256,
+      sizeBytes: artifact.sizeBytes,
+      reachable: true,
+    })),
+  };
+}
+
+export async function submitMonitorTerminalIngest(
+  config: MonitorTerminalIngestConfig,
+  payload: MonitorCentralIngest,
+  fetchImpl: typeof fetch = fetch,
+): Promise<Record<string, unknown>> {
+  validateConfig(config);
+  const body = stableJson(payload);
+  let lastError: Record<string, unknown> | null = null;
+  for (let attempt = 1; attempt <= config.retry.maxAttempts; attempt += 1) {
+    try {
+      const response = await fetchImpl(config.endpoint, {
+        method: "POST",
+        headers: { "content-type": "application/json" },
+        body,
+        signal: AbortSignal.timeout(config.timeoutMs),
+      });
+      const responseBody = await response.json().catch(() => null);
+      if (response.ok && isRecord(responseBody) && responseBody.ok === true) {
+        return { ...responseBody, attempts: attempt, valuesRedacted: true };
+      }
+      lastError = { attempt, httpStatus: response.status, response: responseBody, valuesRedacted: true };
+    } catch (error) {
+      lastError = { attempt, reason: error instanceof Error ? error.message : String(error), valuesRedacted: true };
+    }
+    if (attempt < config.retry.maxAttempts && config.retry.delayMs > 0) await sleep(config.retry.delayMs);
+  }
+  throw new MonitorIngestFailedError({ endpoint: config.endpoint, attempts: config.retry.maxAttempts, lastError, valuesRedacted: true });
+}
+
+export function stableJson(value: unknown): string {
+  if (Array.isArray(value)) return `[${value.map(stableJson).join(",")}]`;
+  if (isRecord(value)) return `{${Object.entries(value).sort(([left], [right]) => left.localeCompare(right)).map(([key, item]) => `${JSON.stringify(key)}:${stableJson(item)}`).join(",")}}`;
+  return JSON.stringify(value);
+}
+
+function immutablePayload(report: Record<string, unknown>, artifacts: readonly MonitorTerminalArtifact[]): Record<string, unknown> {
+  return JSON.parse(stableJson({ report, artifacts })) as Record<string, unknown>;
+}
+
+function validateConfig(config: MonitorTerminalIngestConfig): void {
+  if (!/^https?:\/\//u.test(config.endpoint)) throw new Error("monitor ingest endpoint must be an absolute HTTP(S) URL");
+  if (!Number.isSafeInteger(config.timeoutMs) || config.timeoutMs < 1) throw new Error("monitor ingest timeoutMs must be a positive integer");
+  if (!Number.isSafeInteger(config.retry.maxAttempts) || config.retry.maxAttempts < 1) throw new Error("monitor ingest retry.maxAttempts must be a positive integer");
+  if (!Number.isSafeInteger(config.retry.delayMs) || config.retry.delayMs < 0) throw new Error("monitor ingest retry.delayMs must be a non-negative integer");
+}
+
+function recordArray(value: unknown): readonly Record<string, unknown>[] {
+  return Array.isArray(value) ? value.filter(isRecord) : [];
+}
+
+function stringValue(value: unknown): string | undefined {
+  return typeof value === "string" && value.length > 0 ? value : undefined;
+}
+
+function sha256(value: string): string {
+  return createHash("sha256").update(value).digest("hex");
+}
+
+function isRecord(value: unknown): value is Record<string, unknown> {
+  return typeof value === "object" && value !== null && !Array.isArray(value);
+}
+
+function sleep(delayMs: number): Promise<void> {
+  return new Promise((resolve) => setTimeout(resolve, delayMs));
+}