From 35a59b27c619172d02f90d43afe24faa754d8e27 Mon Sep 17 00:00:00 2001 From: AgentRun Codex Date: Sun, 12 Jul 2026 21:06:08 +0000 Subject: [PATCH] fix: complete monitor ingest migration contracts --- scripts/monitor-sqlite-migration-import.ts | 30 +++ .../hwlab-node-web-sentinel-cicd-shared.ts | 1 - scripts/src/hwlab-node-web-sentinel-cicd.ts | 32 +++- .../src/hwlab-node-web-sentinel-p5-observe.ts | 28 ++- scripts/src/hwlab-node/web-probe-observe.ts | 5 +- scripts/src/monitor-central-persistence.ts | 18 +- scripts/src/monitor-sqlite-migration.ts | 177 ++++++------------ .../monitor-terminal-ingest-migration.test.ts | 21 ++- scripts/src/monitor-terminal-ingest.ts | 144 +++++--------- 9 files changed, 209 insertions(+), 247 deletions(-) create mode 100644 scripts/monitor-sqlite-migration-import.ts diff --git a/scripts/monitor-sqlite-migration-import.ts b/scripts/monitor-sqlite-migration-import.ts new file mode 100644 index 00000000..a751c73f --- /dev/null +++ b/scripts/monitor-sqlite-migration-import.ts @@ -0,0 +1,30 @@ +// SPEC: PJ2026-01060508 Web哨兵 draft-2026-07-12-p17-monitor-central-persistence. +// Responsibility: Explicit one-shot SQLite snapshot import through the central Monitor PostgreSQL store. +import { readFileSync } from "node:fs"; + +import { createPostgresMonitorCentralStore, structuredMonitorCentralError } from "./src/monitor-central-persistence"; +import { importMonitorSqliteSnapshot, type MonitorSqliteSnapshot } from "./src/monitor-sqlite-migration"; + +function required(name: string): string { + const value = process.env[name]; + if (!value) throw Object.assign(new Error(`${name} is required`), { code: "monitor_migration_runtime_config_missing" }); + return value; +} + +function positive(name: string): number { + const value = Number(required(name)); + if (!Number.isSafeInteger(value) || value < 1) throw Object.assign(new Error(`${name} must be a positive integer`), { code: "monitor_migration_runtime_config_invalid" }); + return value; +} + +try { + const snapshotPath = process.argv[2]; + if (!snapshotPath) throw Object.assign(new Error("snapshot path is required"), { code: "monitor_migration_snapshot_missing" }); + const snapshot = JSON.parse(readFileSync(snapshotPath, "utf8")) as MonitorSqliteSnapshot; + const store = createPostgresMonitorCentralStore(required("DATABASE_URL"), { poolMax: positive("MONITOR_CENTRAL_PG_POOL_MAX"), idleTimeoutSeconds: positive("MONITOR_CENTRAL_PG_IDLE_TIMEOUT_SECONDS") }); + await store.migrate(); + console.log(JSON.stringify(await importMonitorSqliteSnapshot(snapshot, store))); +} catch (error) { + console.log(JSON.stringify({ ok: false, error: structuredMonitorCentralError(error), valuesRedacted: true })); + process.exitCode = 2; +} diff --git a/scripts/src/hwlab-node-web-sentinel-cicd-shared.ts b/scripts/src/hwlab-node-web-sentinel-cicd-shared.ts index c53fdaa8..f2464c2f 100644 --- a/scripts/src/hwlab-node-web-sentinel-cicd-shared.ts +++ b/scripts/src/hwlab-node-web-sentinel-cicd-shared.ts @@ -33,7 +33,6 @@ export type WebProbeSentinelOptions = readonly node: string; readonly lane: string; readonly sentinelId: string | null; - readonly sourcePath: string | null; readonly snapshotPath: string | null; readonly confirm: boolean; readonly json: boolean; diff --git a/scripts/src/hwlab-node-web-sentinel-cicd.ts b/scripts/src/hwlab-node-web-sentinel-cicd.ts index 0c36750a..e96178b2 100644 --- a/scripts/src/hwlab-node-web-sentinel-cicd.ts +++ b/scripts/src/hwlab-node-web-sentinel-cicd.ts @@ -20,7 +20,7 @@ import type { CicdDeliveryAuthority } from "./cicd-delivery-authority"; import { startJob } from "./jobs"; import { webProbeSentinelConfigPlan, withWebProbeSentinelConfigRendered } from "./hwlab-node-web-sentinel-config"; import { readWebProbeSentinelConfigRefTarget } from "./hwlab-node-web-sentinel-config-ref"; -import { effectiveWebProbeSentinelPublicExposure, requireSentinelIdForRegistry, resolveWebProbeSentinel } from "./hwlab-node-web-sentinel-resolver"; +import { effectiveWebProbeSentinelPublicExposure, requireSentinelIdForRegistry, resolveWebProbeSentinel, webProbeSentinelRegistryRows } from "./hwlab-node-web-sentinel-resolver"; import type { HwlabRuntimeLaneSpec } from "./hwlab-node-lanes"; import { guardedSentinelDeliveryAction, @@ -192,17 +192,16 @@ export function runWebProbeSentinelCommand(spec: HwlabRuntimeLaneSpec, options: } function runSentinelMigration(spec: HwlabRuntimeLaneSpec, options: Extract): 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 sources = resolvedMigrationSources(spec, options.sentinelId); const command = `web-probe sentinel migration ${options.action}`; if (options.action === "plan") { - const projection = planMonitorSqliteSources([source]); + const projection = planMonitorSqliteSources(sources); return rendered(true, command, renderMigrationCompact(projection), projection); } if (options.action === "export") { + if (sources.length !== 1) throw new Error("web-probe sentinel migration export requires --sentinel when more than one resolved registry source exists"); if (options.snapshotPath === null) throw new Error("web-probe sentinel migration export requires --snapshot"); - const snapshot = exportMonitorSqliteSnapshot(source, options.snapshotPath); + const snapshot = exportMonitorSqliteSnapshot(sources[0]!, 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); @@ -211,12 +210,29 @@ function runSentinelMigration(spec: HwlabRuntimeLaneSpec, options: Extract[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); + const source = sources.find((item) => item.sentinelId === snapshot.source.sentinelId && item.runtimeConfigRef === snapshot.source.runtimeConfigRef); + if (source === undefined) throw new Error("migration snapshot source is not in the current resolved registry"); + const runtime = recordTarget(readWebProbeSentinelConfigRefTarget(spec, source.runtimeConfigRef), source.runtimeConfigRef); + const databaseUrl = stringAtNullable(runtime, "monitor.migration.databaseUrl"); + const poolMax = numberAtNullable(runtime, "monitor.migration.poolMax"); + const idleTimeoutSeconds = numberAtNullable(runtime, "monitor.migration.idleTimeoutSeconds"); + if (databaseUrl === null || poolMax === null || idleTimeoutSeconds === null) throw new Error("monitor.migration databaseUrl, poolMax, and idleTimeoutSeconds must be declared by owning runtime YAML before import"); + const result = runCommand(["bun", "scripts/monitor-sqlite-migration-import.ts", options.snapshotPath], repoRoot, { timeoutMs: 120_000, env: { ...process.env, DATABASE_URL: databaseUrl, MONITOR_CENTRAL_PG_POOL_MAX: String(poolMax), MONITOR_CENTRAL_PG_IDLE_TIMEOUT_SECONDS: String(idleTimeoutSeconds) } }); + const projection = parseJsonObject(result.stdout) ?? { ok: false, command, error: "monitor-migration-import-unparseable", stderr: short(result.stderr), valuesRedacted: true }; + return rendered(projection.ok === true, command, renderMigrationCompact(projection), projection); } return rendered(verification.ok === true, command, renderMigrationCompact(verification), verification); } +function resolvedMigrationSources(spec: HwlabRuntimeLaneSpec, sentinelId: string | null): readonly MonitorSqliteSource[] { + const ids = sentinelId === null ? webProbeSentinelRegistryRows(spec).filter((item) => item.enabled).map((item) => item.id) : [sentinelId]; + return ids.map((id) => { + const sentinel = resolveWebProbeSentinel(spec, id); + const runtime = recordTarget(readWebProbeSentinelConfigRefTarget(spec, sentinel.configRefs.runtime), sentinel.configRefs.runtime); + return { sentinelId: sentinel.id, node: spec.nodeId, lane: spec.lane, path: stringAt(runtime, "sqlite.path"), runtimeConfigRef: sentinel.configRefs.runtime }; + }); +} + function renderMigrationCompact(value: Record): 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(" "); diff --git a/scripts/src/hwlab-node-web-sentinel-p5-observe.ts b/scripts/src/hwlab-node-web-sentinel-p5-observe.ts index 4c8648c2..ac2af1ee 100644 --- a/scripts/src/hwlab-node-web-sentinel-p5-observe.ts +++ b/scripts/src/hwlab-node-web-sentinel-p5-observe.ts @@ -34,6 +34,7 @@ import { withWarnings, } from "./hwlab-node-web-sentinel-cicd"; import { emitWebProbeSentinelSpan } from "./hwlab-node-web-sentinel-otel"; +import { buildMonitorTerminalIngest, resolveMonitorTerminalWriter } from "./monitor-terminal-ingest"; const QUICK_VERIFY_ANALYSIS_SUMMARY_TIMEOUT_SECONDS = 55; @@ -974,6 +975,7 @@ function clipTail(value: string, maxLength: number): string { } function recordQuickVerify(state: SentinelCicdState, payload: Record): Record { + const writer = resolveMonitorTerminalWriter(state.runtime); const views = compactQuickVerifyRecordViews(record(payload.views)); const summary = { reason: payload.reason, @@ -991,7 +993,7 @@ function recordQuickVerify(state: SentinelCicdState, payload: Record, config: NonNullable["ingest"]>, views: Record, summary: Record): Record { + const analysis = record(payload.analysis); + const stateDir = stringAtNullable(payload, "stateDir") ?? stringAtNullable(analysis, "stateDir") ?? ""; + const report = { ...analysis, findings: Array.isArray(payload.findings) ? payload.findings.map(record) : [], timeline: [{ phase: "analyze", status: payload.status, at: new Date().toISOString() }], summary, views, valuesRedacted: true }; + const artifacts = [{ relativePath: "analysis/report.json", kind: "report-json", sha256: String(payload.reportJsonSha256 ?? analysis.reportJsonSha256 ?? "sha256:missing"), sizeBytes: numberAtNullable(analysis, "reportJsonBytes") ?? 0 }]; + const ingest = buildMonitorTerminalIngest({ sentinelId: state.sentinelId, node: state.spec.nodeId, lane: state.spec.lane, runId: String(payload.runId) }, report, artifacts, { ownerNode: state.spec.nodeId, ownerLane: state.spec.lane, ownerPvc: stringAt(state.runtime, "pvcName"), stateDir }); + let lastError: Record | null = null; + for (let attempt = 1; attempt <= config.retry.maxAttempts; attempt += 1) { + const result = runCommand(["curl", "--fail-with-body", "--silent", "--show-error", "--max-time", String(Math.max(1, Math.ceil(config.timeoutMs / 1000))), "-X", "POST", "-H", "content-type: application/json", "--data-binary", JSON.stringify(ingest), config.endpoint], repoRoot, { timeoutMs: config.timeoutMs }); + if (result.exitCode === 0) return { ok: true, writer: "central", attempts: attempt, payloadHash: ingest.payloadHash, valuesRedacted: true }; + lastError = { attempt, exitCode: result.exitCode, stderr: short(result.stderr), valuesRedacted: true }; + } + return { ok: false, writer: "central", error: "monitor-ingest-failed", attempts: config.retry.maxAttempts, payloadHash: ingest.payloadHash, lastError, valuesRedacted: true }; } function sentinelOtelContext(state: SentinelCicdState): { readonly node: string; readonly lane: string; readonly sentinelId: string; readonly namespace: string | null; readonly runtime: Record; readonly cicd: Record } { diff --git a/scripts/src/hwlab-node/web-probe-observe.ts b/scripts/src/hwlab-node/web-probe-observe.ts index 0cc86ce5..766474b6 100644 --- a/scripts/src/hwlab-node/web-probe-observe.ts +++ b/scripts/src/hwlab-node/web-probe-observe.ts @@ -88,7 +88,6 @@ 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; @@ -222,10 +221,10 @@ export function parseNodeWebProbeSentinelOptions(args: string[]): NodeWebProbeSe } 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]"); + throw new Error("web-probe sentinel migration usage: migration plan|export|import|verify --node NODE --lane vNN [--sentinel ID] [--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") }; + sentinel = { kind: "migration", action: migrationAction, node, lane, sentinelId, 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"); diff --git a/scripts/src/monitor-central-persistence.ts b/scripts/src/monitor-central-persistence.ts index 24524307..a85a20a8 100644 --- a/scripts/src/monitor-central-persistence.ts +++ b/scripts/src/monitor-central-persistence.ts @@ -50,7 +50,7 @@ export interface MonitorCentralStore { ping(): Promise; schemaVersion(): Promise; capability(): Promise<{ readonly ingest: boolean; readonly query: boolean }>; - ingest(input: NormalizedIngest): Promise<{ readonly disposition: "inserted" | "idempotent" }>; + ingest(input: NormalizedMonitorCentralIngest): Promise<{ readonly disposition: "inserted" | "idempotent" }>; updateArtifactAvailability(scope: Required, runId: string, locatorIndex: number, reachable: boolean): Promise; overview(scope: MonitorScope): Promise>; runs(scope: MonitorScope, options: RunQueryOptions): Promise; @@ -59,7 +59,7 @@ export interface MonitorCentralStore { findings(scope: MonitorScope, options: FindingQueryOptions): Promise[]>; } -interface NormalizedIngest extends Required> { +export interface NormalizedMonitorCentralIngest extends Required> { readonly payloadHash: string; } @@ -117,7 +117,7 @@ export function createMonitorCentralService(options: MonitorCentralServiceOption const pageSize = requirePositiveInteger(options.pageSize, "monitor_central_page_size_missing"); const store = options.store ?? createPostgresMonitorCentralStore(requireDatabaseUrl(options.databaseUrl), requirePostgresOptions(options.postgres)); const ingest = async (input: MonitorCentralIngest) => { - const normalized = normalizeIngest(input); + const normalized = normalizeMonitorCentralIngest(input); const result = await store.ingest(normalized); return { ok: true, ...result, identity: identityOf(normalized), payloadHash: normalized.payloadHash, valuesRedacted: true }; }; @@ -280,7 +280,7 @@ export const MONITOR_CENTRAL_MIGRATIONS = [ ]; export function createInMemoryMonitorCentralStore(): MonitorCentralStore { - const records = new Map(); + const records = new Map(); return { async migrate() {}, async ping() {}, @@ -331,10 +331,10 @@ export function createInMemoryMonitorCentralStore(): MonitorCentralStore { function requireDatabaseUrl(databaseUrl: string | null | undefined): string { const value = databaseUrl ?? process.env.DATABASE_URL; if (!value) throw new MonitorCentralConfigurationError(); return value; } function requirePostgresOptions(value: PostgresMonitorCentralStoreOptions | undefined): PostgresMonitorCentralStoreOptions { if (!value) throw Object.assign(new Error("Monitor central PostgreSQL pool configuration must be injected"), { code: "monitor_central_pg_options_missing" }); return value; } function requirePositiveInteger(value: number | undefined, code: string): number { if (!Number.isSafeInteger(value) || value < 1) throw Object.assign(new Error(`${code} must be a positive integer`), { code }); return value; } -function normalizeIngest(input: MonitorCentralIngest): NormalizedIngest { +export function normalizeMonitorCentralIngest(input: MonitorCentralIngest): NormalizedMonitorCentralIngest { for (const key of ["sentinelId", "node", "lane", "runId"] as const) if (!input[key]) throw invalid(`missing required ingest field: ${key}`); const normalized = { ...input, updatedAt: input.updatedAt ?? new Date().toISOString(), status: input.status ?? "unknown", findings: input.findings ?? [], artifactLocators: (input.artifactLocators ?? []).map(normalizeLocator), timeline: input.timeline ?? [] }; - return { ...normalized, payloadHash: `sha256:${createHash("sha256").update(stableJson({ status: normalized.status, payload: normalized.payload, findings: normalized.findings, artifactLocators: normalized.artifactLocators.map(immutableLocator), timeline: normalized.timeline })).digest("hex")}` }; + return { ...normalized, payloadHash: `sha256:${createHash("sha256").update(monitorCentralStableJson({ status: normalized.status, payload: normalized.payload, findings: normalized.findings, artifactLocators: normalized.artifactLocators.map(immutableLocator), timeline: normalized.timeline })).digest("hex")}` }; } function immutableLocator(locator: MonitorArtifactLocator): Omit { const { reachable: _reachable, ...immutable } = locator; return immutable; } function normalizeLocator(value: MonitorArtifactLocator): MonitorArtifactLocator { @@ -367,17 +367,17 @@ function identityKey(value: MonitorScope & { readonly runId: string }): string { function conflict(value: MonitorCentralIngest): Error & { code: string } { return Object.assign(new Error("ingest identity already has a different terminal envelope"), { code: "payload_hash_conflict", identity: identityOf(value) }); } function publicScope(scope: MonitorScope): Record { return { sentinelId: scope.sentinelId ?? null, node: scope.node ?? null, lane: scope.lane ?? null }; } function matches(scope: MonitorScope, value: MonitorScope): boolean { return (!scope.sentinelId || scope.sentinelId === value.sentinelId) && (!scope.node || scope.node === value.node) && (!scope.lane || scope.lane === value.lane); } -function runFrom(value: NormalizedIngest): MonitorRun { return { sentinelId: value.sentinelId, node: value.node, lane: value.lane, runId: value.runId, updatedAt: value.updatedAt, status: value.status, payloadHash: value.payloadHash }; } +function runFrom(value: NormalizedMonitorCentralIngest): MonitorRun { return { sentinelId: value.sentinelId, node: value.node, lane: value.lane, runId: value.runId, updatedAt: value.updatedAt, status: value.status, payloadHash: value.payloadHash }; } function sortRuns(runs: readonly MonitorRun[]): MonitorRun[] { return [...runs].sort((left, right) => right.updatedAt.localeCompare(left.updatedAt) || right.runId.localeCompare(left.runId)); } function afterRunCursor(run: MonitorRun, cursorValue: RunCursor | null): boolean { return !cursorValue || run.updatedAt < cursorValue.updatedAt || (run.updatedAt === cursorValue.updatedAt && run.runId < cursorValue.runId); } function compareFindings(left: Record, right: Record): number { return String(right.updatedAt).localeCompare(String(left.updatedAt)) || String(right.runId).localeCompare(String(left.runId)) || Number(right.findingIndex) - Number(left.findingIndex); } function afterFindingCursor(finding: Record, cursorValue: FindingCursor | null): boolean { if (!cursorValue) return true; return String(finding.updatedAt) < cursorValue.updatedAt || (finding.updatedAt === cursorValue.updatedAt && (String(finding.runId) < cursorValue.runId || (finding.runId === cursorValue.runId && Number(finding.findingIndex) < cursorValue.findingIndex))); } -function stableJson(value: unknown): string { if (Array.isArray(value)) return `[${value.map(stableJson).join(",")}]`; if (value && typeof value === "object") return `{${Object.entries(value as Record).sort(([left], [right]) => left.localeCompare(right)).map(([key, item]) => `${JSON.stringify(key)}:${stableJson(item)}`).join(",")}}`; return JSON.stringify(value); } +export function monitorCentralStableJson(value: unknown): string { if (Array.isArray(value)) return `[${value.map(monitorCentralStableJson).join(",")}]`; if (value && typeof value === "object") return `{${Object.entries(value as Record).sort(([left], [right]) => left.localeCompare(right)).map(([key, item]) => `${JSON.stringify(key)}:${monitorCentralStableJson(item)}`).join(",")}}`; return JSON.stringify(value); } function scopeWhere(scope: MonitorScope): { readonly text: string; readonly values: readonly string[] } { const entries = Object.entries({ sentinel_id: scope.sentinelId, node: scope.node, lane: scope.lane }).filter(([, value]) => value); return { text: entries.length ? entries.map(([key], index) => `r.${key} = $${index + 1}`).join(" and ") : "true", values: entries.map(([, value]) => String(value)) }; } function identityWhere(scope: Required, runId: string): { readonly text: string; readonly values: readonly string[] } { return { text: "r.sentinel_id = $1 and r.node = $2 and r.lane = $3 and r.run_id = $4", values: [scope.sentinelId, scope.node, scope.lane, runId] }; } async function queryRuns(sql: postgres.Sql, scope: MonitorScope, options: RunQueryOptions): Promise { const where = scopeWhere(scope); const values: unknown[] = [...where.values]; let cursorClause = ""; if (options.cursor) { values.push(options.cursor.updatedAt, options.cursor.runId); cursorClause = `and (r.updated_at, r.run_id) < ($${values.length - 1}, $${values.length})`; } values.push(options.limit); const rows = await sql.unsafe(`select r.sentinel_id, r.node, r.lane, r.run_id, r.updated_at, r.status, r.payload_hash from monitor.runs r where ${where.text} ${cursorClause} order by r.updated_at desc, r.run_id desc limit $${values.length}`, values); return rows.map(runRow); } async function countRuns(sql: postgres.Sql, scope: MonitorScope): Promise { const where = scopeWhere(scope); const rows = await sql.unsafe(`select count(*)::int as count from monitor.runs r where ${where.text}`, where.values); return Number(rows[0]?.count ?? 0); } -async function updatePostgresArtifactAvailability(transaction: postgres.TransactionSql, input: NormalizedIngest): Promise { +async function updatePostgresArtifactAvailability(transaction: postgres.TransactionSql, input: NormalizedMonitorCentralIngest): Promise { for (const [index, locator] of input.artifactLocators.entries()) await transaction` update monitor.artifact_locators set reachable = ${locator.reachable} diff --git a/scripts/src/monitor-sqlite-migration.ts b/scripts/src/monitor-sqlite-migration.ts index 0fdbf5f3..be8347eb 100644 --- a/scripts/src/monitor-sqlite-migration.ts +++ b/scripts/src/monitor-sqlite-migration.ts @@ -1,146 +1,89 @@ // 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. +// Responsibility: Registry-resolved read-only legacy SQLite snapshots and one-shot canonical central-store import. 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; -} +import type { MonitorArtifactLocator, MonitorCentralIngest, MonitorCentralStore } from "./monitor-central-persistence"; +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; } +export interface MonitorSqliteTable { readonly name: string; readonly sql: string; readonly rowCount: number; readonly rows: readonly Record[]; } 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[] }[]; - 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; + readonly schema: string; readonly createdAt: string; readonly source: MonitorSqliteSource; readonly fingerprint: string; readonly integrity: string; readonly byteLength: number; + readonly tables: readonly MonitorSqliteTable[]; + readonly artifactLocator: { readonly path: string; readonly sha256: string; readonly sizeBytes: number; readonly kind: "legacy-sqlite" }; + readonly manifestContentHash: string; } +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 { - 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 }; + return { ok: true, command: "web-probe sentinel migration plan", sources: sources.map(sourcePlan), sourceCount: sources.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 sourceBytes = readFileSync(path); 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(); - } + 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 snapshot: MonitorSqliteSnapshot = { ...unsigned, manifestContentHash: `sha256:${sha256(Buffer.from(monitorCentralStableJson(unsigned)))}` }; + writeSnapshot(outputPath, `${monitorCentralStableJson(snapshot)}\n`); + return snapshot; + } finally { database.close(); } } export function verifyMonitorSqliteSnapshot(snapshot: MonitorSqliteSnapshot): Record { - const expected = `sha256:${sha256(Buffer.from(stableJson({ ...snapshot, artifactLocator: { ...snapshot.artifactLocator, sha256: "", byteLength: 0 } })))} ` .trim(); + const sourceBytes = readFileSync(assertReadOnlySqliteSource(snapshot.artifactLocator.path)); + const locatorValid = snapshot.artifactLocator.sha256 === `sha256:${sha256(sourceBytes)}` && snapshot.artifactLocator.sizeBytes === sourceBytes.byteLength; + const fingerprintValid = snapshot.fingerprint === snapshot.artifactLocator.sha256 && snapshot.byteLength === snapshot.artifactLocator.sizeBytes; 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, - }; + const manifestContentHashValid = snapshot.manifestContentHash === `sha256:${sha256(Buffer.from(monitorCentralStableJson({ schema: snapshot.schema, createdAt: snapshot.createdAt, source: snapshot.source, fingerprint: snapshot.fingerprint, integrity: snapshot.integrity, byteLength: snapshot.byteLength, tables: snapshot.tables, artifactLocator: snapshot.artifactLocator })))}`; + const rowsValid = snapshot.tables.every((table) => table.rowCount === table.rows.length); + return { ok: snapshot.schema === MONITOR_SQLITE_MIGRATION_SCHEMA && snapshot.integrity === "ok" && locatorValid && fingerprintValid && manifestContentHashValid && rowsValid, command: "web-probe sentinel migration verify", fingerprint: snapshot.fingerprint, integrity: snapshot.integrity, reconciliation, artifactLocator: { ...snapshot.artifactLocator, valid: locatorValid }, schemaValid: snapshot.schema === MONITOR_SQLITE_MIGRATION_SCHEMA, manifestContentHashValid, rowsValid, valuesRedacted: true }; } -export function reconcileTables(tables: readonly MonitorSqliteSnapshot["tables"][number][]): MonitorMigrationReconciliation { +export async function importMonitorSqliteSnapshot(snapshot: MonitorSqliteSnapshot, store: MonitorCentralStore): Promise> { + 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 dispositions = await Promise.all(ingests.map((input) => store.ingest(input))); + const reconciliation = reconcileTables(snapshot.tables); + return { ok: true, command: "web-probe sentinel migration import", source: snapshot.source, 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 }; +} + +export function reconcileTables(tables: readonly MonitorSqliteTable[]): 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"]), - }; + 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 { - const row = value as Record; - 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"); +function canonicalIngests(snapshot: MonitorSqliteSnapshot): readonly ReturnType[] { + const rows = tableRows(snapshot.tables, ["runs", "run"]); + 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((item) => jsonField(item, ["finding", "value_json"]) ?? item); + const rowLocators = locators.filter((item) => stringField(item, ["run_id"]) === runId).map((item) => locatorFromRow(item, snapshot)); + const payload = jsonField(payloads.find((item) => stringField(item, ["run_id"]) === runId), ["payload", "value_json"]) ?? { legacy: { run: row, metadata: metadata.filter((item) => String(item.key ?? "").includes(runId)), sourceFingerprint: snapshot.fingerprint } }; + 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: rowFindings, artifactLocators: rowLocators, timeline: [] }); + }); } +function locatorFromRow(row: Record, source: MonitorSqliteSnapshot): MonitorArtifactLocator { return { ownerNode: stringField(row, ["owner_node", "ownerNode"]) || source.source.node, ownerLane: stringField(row, ["owner_lane", "ownerLane"]) || source.source.lane, ownerPvc: stringField(row, ["owner_pvc", "ownerPvc"]) || "legacy-sqlite", stateDir: stringField(row, ["state_dir", "stateDir"]) || dirname(source.source.path), relativePath: stringField(row, ["relative_path", "relativePath"]) || source.source.path, sha256: stringField(row, ["sha256"]) || source.fingerprint, sizeBytes: numberField(row, ["size_bytes", "sizeBytes"]) ?? source.byteLength, kind: stringField(row, ["kind"]) || "legacy-sqlite", reachable: row.reachable !== false }; } +function tableRows(tables: readonly MonitorSqliteTable[], names: readonly string[]): readonly Record[] { return tables.filter((table) => names.includes(table.name)).flatMap((table) => table.rows); } +function sourcePlan(source: MonitorSqliteSource): Record { 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 quoteIdentifier(value: string): string { return `"${value.replaceAll("\"", "\"\"")}"`; } +function normalizeRow(value: unknown): Record { const row = value as Record; return Object.fromEntries(Object.entries(row).map(([key, item]) => [key, item instanceof Uint8Array ? Buffer.from(item).toString("base64") : item])); } +function jsonField(row: Record | undefined, fields: readonly string[]): Record | 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 : null; } catch { return null; } return value && typeof value === "object" && !Array.isArray(value) ? value as Record : null; } +function stringField(row: Record | 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, 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"); } diff --git a/scripts/src/monitor-terminal-ingest-migration.test.ts b/scripts/src/monitor-terminal-ingest-migration.test.ts index addf58c4..c3bd771a 100644 --- a/scripts/src/monitor-terminal-ingest-migration.test.ts +++ b/scripts/src/monitor-terminal-ingest-migration.test.ts @@ -1,13 +1,14 @@ // 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 { mkdtempSync, readFileSync, rmSync } 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"; +import { createInMemoryMonitorCentralStore } from "./monitor-central-persistence"; +import { exportMonitorSqliteSnapshot, importMonitorSqliteSnapshot, planMonitorSqliteSources, verifyMonitorSqliteSnapshot } from "./monitor-sqlite-migration"; +import { buildMonitorTerminalIngest, MonitorIngestFailedError, resolveMonitorTerminalWriter, 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" }; @@ -15,13 +16,21 @@ test("terminal ingest payload is immutable and retries bounded failures", async report.findings[0].id = "changed"; assert.equal((ingest.payload.report as { findings: { id: string }[] }).findings[0].id, "blocked"); assert.match(ingest.payloadHash ?? "", /^sha256:/u); + assert.equal(ingest.payloadHash, buildMonitorTerminalIngest({ sentinelId: "fixture", node: "NC01", lane: "v03", runId: "run-1" }, { ok: false, findings: [{ id: "blocked" }], updatedAt: "2026-07-12T00:00:00.000Z" }, [{ relativePath: "analysis/report.json", kind: "report", sha256: "sha256:report", sizeBytes: 42 }], { ownerNode: "NC01", ownerLane: "v03", ownerPvc: "fixture-pvc", stateDir: "/state" }).payloadHash); 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", () => { +test("terminal writer defaults to legacy authority and validates central selection", () => { + assert.deepEqual(resolveMonitorTerminalWriter({}), { mode: "legacy-sqlite" }); + const central = resolveMonitorTerminalWriter({ monitor: { terminalWriter: { mode: "central", ingest: { endpoint: "https://monitor.example.test/api/ingest", timeoutMs: 1_000, retry: { maxAttempts: 2, delayMs: 0 } } } } }); + assert.equal(central.mode, "central"); + assert.throws(() => resolveMonitorTerminalWriter({ monitor: { terminalWriter: { mode: "central" } } })); +}); + +test("SQLite snapshot reconciles the 228/242/234 fixture shape without production assumptions", async () => { const directory = mkdtempSync(join(tmpdir(), "monitor-migration-")); try { const sqlitePath = join(directory, "legacy.sqlite"); @@ -43,6 +52,10 @@ test("SQLite snapshot reconciles the 228/242/234 fixture shape without productio 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); + const store = createInMemoryMonitorCentralStore(); + assert.deepEqual((await importMonitorSqliteSnapshot(snapshot, store)).imported, { inserted: 228, idempotent: 0 }); + assert.deepEqual((await importMonitorSqliteSnapshot(snapshot, store)).imported, { inserted: 0, idempotent: 228 }); + assert.equal(verifyMonitorSqliteSnapshot({ ...snapshot, manifestContentHash: "sha256:tampered" }).ok, false); } finally { rmSync(directory, { recursive: true, force: true }); } diff --git a/scripts/src/monitor-terminal-ingest.ts b/scripts/src/monitor-terminal-ingest.ts index 2b1ecefd..a11f7544 100644 --- a/scripts/src/monitor-terminal-ingest.ts +++ b/scripts/src/monitor-terminal-ingest.ts @@ -1,106 +1,46 @@ // 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"; - +// Responsibility: Build canonical immutable terminal envelopes and submit them with YAML-declared bounded retries. import type { MonitorArtifactLocator, MonitorCentralIngest } from "./monitor-central-persistence"; +import { monitorCentralStableJson, normalizeMonitorCentralIngest } 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 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) { - super("Monitor terminal ingest failed after bounded retries"); - this.name = "MonitorIngestFailedError"; - } + constructor(readonly details: Record) { super("Monitor terminal ingest failed after bounded retries"); this.name = "MonitorIngestFailedError"; } } -export function buildMonitorTerminalIngest( - identity: MonitorTerminalIdentity, - report: Record, - artifacts: readonly MonitorTerminalArtifact[], - locatorOwner: Pick, -): MonitorCentralIngest { - const payload = immutablePayload(report, artifacts); - return { +export function buildMonitorTerminalIngest(identity: MonitorTerminalIdentity, report: Record, artifacts: readonly MonitorTerminalArtifact[], locatorOwner: Pick): MonitorCentralIngest { + const immutableReport = JSON.parse(monitorCentralStableJson(report)) as Record; + return normalizeMonitorCentralIngest({ ...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, - })), - }; + updatedAt: stringValue(immutableReport.updatedAt) ?? stringValue(immutableReport.analyzedAt) ?? new Date().toISOString(), + status: stringValue(immutableReport.status) ?? (immutableReport.ok === true ? "ok" : "blocked"), + payload: { report: immutableReport, artifacts: artifacts.map((artifact) => ({ ...artifact })) }, + findings: recordArray(immutableReport.findings), + timeline: recordArray(immutableReport.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> { +export async function submitMonitorTerminalIngest(config: MonitorTerminalIngestConfig, payload: MonitorCentralIngest, fetchImpl: typeof fetch = fetch): Promise> { validateConfig(config); - const body = stableJson(payload); + const canonical = normalizeMonitorCentralIngest(payload); + const body = monitorCentralStableJson(canonical); let lastError: Record | 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 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 }; - } + if (response.ok && isRecord(responseBody) && responseBody.ok === true) return { ...responseBody, attempts: attempt, payloadHash: canonical.payloadHash, 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 }; - } + } 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, artifacts: readonly MonitorTerminalArtifact[]): Record { - return JSON.parse(stableJson({ report, artifacts })) as Record; + throw new MonitorIngestFailedError({ endpoint: config.endpoint, attempts: config.retry.maxAttempts, payloadHash: canonical.payloadHash, lastError, valuesRedacted: true }); } function validateConfig(config: MonitorTerminalIngestConfig): void { @@ -109,23 +49,25 @@ function validateConfig(config: MonitorTerminalIngestConfig): void { 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[] { return Array.isArray(value) ? value.filter(isRecord) : []; } +function stringValue(value: unknown): string | undefined { return typeof value === "string" && value.length > 0 ? value : undefined; } +function isRecord(value: unknown): value is Record { return typeof value === "object" && value !== null && !Array.isArray(value); } +function sleep(delayMs: number): Promise { return new Promise((resolve) => setTimeout(resolve, delayMs)); } -function recordArray(value: unknown): readonly Record[] { - return Array.isArray(value) ? value.filter(isRecord) : []; -} +export interface MonitorTerminalWriterConfig { readonly mode: "legacy-sqlite" | "central"; readonly ingest?: MonitorTerminalIngestConfig; } -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 { - return typeof value === "object" && value !== null && !Array.isArray(value); -} - -function sleep(delayMs: number): Promise { - return new Promise((resolve) => setTimeout(resolve, delayMs)); +export function resolveMonitorTerminalWriter(runtime: Record): MonitorTerminalWriterConfig { + const monitor = runtime.monitor; + const writer = monitor && typeof monitor === "object" && !Array.isArray(monitor) ? (monitor as Record).terminalWriter : null; + if (!writer || typeof writer !== "object" || Array.isArray(writer)) return { mode: "legacy-sqlite" }; + const value = writer as Record; + const mode = value.mode; + if (mode !== "legacy-sqlite" && mode !== "central") throw new Error("monitor.terminalWriter.mode must be legacy-sqlite or central"); + if (mode === "legacy-sqlite") return { mode }; + const ingest = value.ingest; + if (!ingest || typeof ingest !== "object" || Array.isArray(ingest)) throw new Error("monitor.terminalWriter.ingest is required when mode=central"); + const config = ingest as Record; + const retry = config.retry; + if (!retry || typeof retry !== "object" || Array.isArray(retry)) throw new Error("monitor.terminalWriter.ingest.retry is required when mode=central"); + return { mode, ingest: { endpoint: String(config.endpoint ?? ""), timeoutMs: Number(config.timeoutMs), retry: { maxAttempts: Number((retry as Record).maxAttempts), delayMs: Number((retry as Record).delayMs) } } }; }