fix: address monitor ingest review blockers
This commit is contained in:
AgentRun Codex
@@ -2,7 +2,7 @@
|
||||
// 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 { createPostgresMonitorCentralStore, MONITOR_CENTRAL_SCHEMA_VERSION, structuredMonitorCentralError } from "./src/monitor-central-persistence";
|
||||
import { importMonitorSqliteSnapshot, type MonitorSqliteSnapshot } from "./src/monitor-sqlite-migration";
|
||||
|
||||
function required(name: string): string {
|
||||
@@ -22,7 +22,9 @@ try {
|
||||
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();
|
||||
await store.ping();
|
||||
if (await store.schemaVersion() !== MONITOR_CENTRAL_SCHEMA_VERSION) throw Object.assign(new Error("central monitor schema is not ready"), { code: "monitor_migration_schema_unready" });
|
||||
if (!(await store.capability()).ingest) throw Object.assign(new Error("central monitor ingest capability is unavailable"), { code: "monitor_migration_capability_unavailable" });
|
||||
console.log(JSON.stringify(await importMonitorSqliteSnapshot(snapshot, store)));
|
||||
} catch (error) {
|
||||
console.log(JSON.stringify({ ok: false, error: structuredMonitorCentralError(error), valuesRedacted: true }));
|
||||
|
||||
@@ -0,0 +1,13 @@
|
||||
// SPEC: PJ2026-01060508 Web哨兵 draft-2026-07-12-p17-monitor-central-persistence.
|
||||
// Responsibility: Synchronous CLI adapter for the canonical bounded terminal ingest client.
|
||||
import { structuredMonitorCentralError } from "./src/monitor-central-persistence";
|
||||
import { submitMonitorTerminalIngest, type MonitorTerminalIngestConfig } from "./src/monitor-terminal-ingest";
|
||||
|
||||
try {
|
||||
const input = JSON.parse(await Bun.stdin.text()) as { config: MonitorTerminalIngestConfig; ingest: Parameters<typeof submitMonitorTerminalIngest>[1] };
|
||||
const result = await submitMonitorTerminalIngest(input.config, input.ingest);
|
||||
console.log(JSON.stringify({ ok: true, ...result, valuesRedacted: true }));
|
||||
} catch (error) {
|
||||
console.log(JSON.stringify({ ok: false, error: structuredMonitorCentralError(error), valuesRedacted: true }));
|
||||
process.exitCode = 2;
|
||||
}
|
||||
@@ -213,11 +213,11 @@ function runSentinelMigration(spec: HwlabRuntimeLaneSpec, options: Extract<WebPr
|
||||
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) } });
|
||||
if (stringAtNullable(runtime, "monitor.migration.databaseUrlSourceRef") === null || stringAtNullable(runtime, "monitor.migration.databaseUrlTargetKey") !== "DATABASE_URL" || poolMax === null || idleTimeoutSeconds === null) throw new Error("monitor.migration must declare databaseUrlSourceRef, databaseUrlTargetKey=DATABASE_URL, poolMax, and idleTimeoutSeconds");
|
||||
if (!process.env.DATABASE_URL) throw new Error("monitor migration DATABASE_URL must be injected by the controlled Secret consumer");
|
||||
const result = runCommand(["bun", "scripts/monitor-sqlite-migration-import.ts", options.snapshotPath], repoRoot, { timeoutMs: 120_000, env: { ...process.env, 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);
|
||||
}
|
||||
@@ -229,7 +229,7 @@ function resolvedMigrationSources(spec: HwlabRuntimeLaneSpec, sentinelId: string
|
||||
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 };
|
||||
return { sentinelId: sentinel.id, node: spec.nodeId, lane: spec.lane, path: stringAt(runtime, "sqlite.path"), runtimeConfigRef: sentinel.configRefs.runtime, stateRoot: stringAt(runtime, "stateRoot"), ownerPvc: stringAt(runtime, "pvcName") };
|
||||
});
|
||||
}
|
||||
|
||||
|
||||
@@ -34,7 +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";
|
||||
import { buildMonitorTerminalIngest, MonitorIngestFailedError, resolveMonitorTerminalWriter } from "./monitor-terminal-ingest";
|
||||
|
||||
const QUICK_VERIFY_ANALYSIS_SUMMARY_TIMEOUT_SECONDS = 55;
|
||||
|
||||
@@ -1028,17 +1028,18 @@ function recordQuickVerify(state: SentinelCicdState, payload: Record<string, unk
|
||||
|
||||
function submitCentralTerminalIngest(state: SentinelCicdState, payload: Record<string, unknown>, config: NonNullable<ReturnType<typeof resolveMonitorTerminalWriter>["ingest"]>, views: Record<string, unknown>, summary: Record<string, unknown>): Record<string, unknown> {
|
||||
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<string, unknown> | 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 };
|
||||
try {
|
||||
const stateDir = stringAtNullable(payload, "stateDir") ?? stringAtNullable(analysis, "stateDir");
|
||||
const report = { ...analysis, updatedAt: stringAtNullable(analysis, "reportUpdatedAt") ?? stringAtNullable(payload, "updatedAt"), findings: Array.isArray(payload.findings) ? payload.findings.map(record) : [], summary, views, valuesRedacted: true };
|
||||
const artifacts = [{ relativePath: "analysis/report.json", kind: "report-json", sha256: stringAtNullable(payload, "reportJsonSha256") ?? stringAtNullable(analysis, "reportJsonSha256") ?? "", sizeBytes: numberAtNullable(analysis, "reportJsonBytes") ?? -1 }];
|
||||
const ingest = buildMonitorTerminalIngest({ sentinelId: state.sentinelId, node: state.spec.nodeId, lane: state.spec.lane, runId: stringAt(payload, "runId") }, report, artifacts, { ownerNode: state.spec.nodeId, ownerLane: state.spec.lane, ownerPvc: stringAt(state.runtime, "pvcName"), stateDir: stateDir ?? "" });
|
||||
const result = runCommand(["bun", "scripts/monitor-terminal-ingest-submit.ts"], repoRoot, { timeoutMs: config.timeoutMs + config.retry.maxAttempts * config.retry.delayMs + 1_000, input: JSON.stringify({ config, ingest }) });
|
||||
const response = parseJsonObject(result.stdout);
|
||||
if (result.exitCode === 0 && response?.ok === true) return { ...response, writer: "central", valuesRedacted: true };
|
||||
return { ok: false, writer: "central", error: "monitor-ingest-failed", response, stderr: short(result.stderr), payloadHash: ingest.payloadHash, valuesRedacted: true };
|
||||
} catch (error) {
|
||||
return { ok: false, writer: "central", error: "monitor-ingest-failed", details: error instanceof MonitorIngestFailedError ? error.details : { reason: error instanceof Error ? error.message : String(error) }, 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<string, unknown>; readonly cicd: Record<string, unknown> } {
|
||||
@@ -1430,6 +1431,8 @@ function analysisSummaryFromAnalyzeResult(analysis: ChildCliResult, fallbackStat
|
||||
stateDir: stringAtNullable(source, "stateDir") ?? fallbackStateDir,
|
||||
reportJsonPath: stringAtNullable(source, "reportJsonPath"),
|
||||
reportJsonSha256,
|
||||
reportJsonBytes: numberAtNullable(source, "reportJsonBytes"),
|
||||
reportUpdatedAt: stringAtNullable(source, "reportUpdatedAt"),
|
||||
reportMdPath: stringAtNullable(source, "reportMdPath"),
|
||||
reportMdSha256: stringAtNullable(source, "reportMdSha256"),
|
||||
findingCount,
|
||||
@@ -1513,7 +1516,7 @@ function readAnalysisSummaryFromWorkspace(state: SentinelCicdState, stateDir: st
|
||||
"function pageMemory(page){const eff=rec(page.effectiveMemory); const heap=rec(page.heapUsage); const perf=rec(page.performance); const metrics=rec(perf.metrics); const heapUsed=num(eff.heapUsedMb)??mb(heap.usedSize); const jsHeap=num(eff.jsHeapUsedMb)??mb(metrics.JSHeapUsedSize); const memory=heapUsed??jsHeap; if(memory==null)return null; return {memoryMb:memory,heapUsedMb:heapUsed,jsHeapUsedMb:jsHeap,effectiveHeapUsedMb:num(eff.effectiveHeapUsedMb),effectiveJsHeapUsedMb:num(eff.effectiveJsHeapUsedMb),domNodes:num(eff.domNodes)??num(rec(page.domCounters).nodes)??num(metrics.Nodes)}; }",
|
||||
"function browserProcessPageSeries(){const p=path.join(stateDir,'browser-process.jsonl'); let lines=[]; try{lines=String(read(p)||'').split(/\\r?\\n/).filter(Boolean)}catch{} const map=new Map(); let firstTs=null; for(const line of lines){let row=null; try{row=JSON.parse(line)}catch{} if(!row||row.type!=='browser-process-sample')continue; const ts=row.ts||null; const tsMs=Date.parse(String(ts||'')); if(!ts||!Number.isFinite(tsMs))continue; for(const page of arr(row.pages)){const memory=pageMemory(rec(page)); if(!memory)continue; if(firstTs==null)firstTs=tsMs; const role=clip(page.pageRole||'page',32); const id=clip(page.pageId||`${role}-${page.pageEpoch??map.size}`,80); const key=`${role}:${id}`; const current=map.get(key)||{key,pageRole:role,pageId:id,label:`${role} ${String(id).slice(0,10)}`,url:clip(page.url,180),points:[],valuesRedacted:true}; current.points.push({ts,elapsedSeconds:Math.max(0,Math.round((tsMs-firstTs)/1000)),elapsedMinutes:Math.round(Math.max(0,(tsMs-firstTs)/60000)*100)/100,...memory,valuesRedacted:true}); map.set(key,current);} } return Array.from(map.values()).filter((item)=>item.points.length>0); }",
|
||||
"const browserPageSeries=browserProcessPageSeries();",
|
||||
"console.log(JSON.stringify({ok:!!report,reason:report?null:(jsonBuf?'report-json-parse-failed':'report-json-missing'),reportReadWaitMs:waitMs,reportParseError:clip(reportParseError,220),reportOk:!!report&&report.ok!==false,stateDir,reportJsonPath:reportPath,reportJsonSha256:sha(jsonBuf),reportMdPath,reportMdSha256:sha(read(reportMdPath)),findingCount:Number(report?.findingCount??findings.length),artifactCount,screenshot,findings,counts:rec(report?.counts),analysisWindow:rec(report?.analysisWindow??report?.windows?.recent?.summary),pagePerformanceSlowApi:slow,browserProcess:{source:'analysis-browser-process-jsonl',unit:'MB',metric:'page-heap-used',pageCount:browserPageSeries.length,sampleCount:browserPageSeries.reduce((sum,item)=>sum+item.points.length,0),pageSeries:browserPageSeries,valuesRedacted:true},requestRate:compactRequestRate(report?.requestRate),valuesRedacted:true}));",
|
||||
"console.log(JSON.stringify({ok:!!report,reason:report?null:(jsonBuf?'report-json-parse-failed':'report-json-missing'),reportReadWaitMs:waitMs,reportParseError:clip(reportParseError,220),reportOk:!!report&&report.ok!==false,stateDir,reportJsonPath:reportPath,reportJsonSha256:sha(jsonBuf),reportJsonBytes:jsonBuf.length,reportUpdatedAt:typeof report?.updatedAt==='string'?report.updatedAt:(typeof report?.analyzedAt==='string'?report.analyzedAt:null),reportMdPath,reportMdSha256:sha(read(reportMdPath)),findingCount:Number(report?.findingCount??findings.length),artifactCount,screenshot,findings,counts:rec(report?.counts),analysisWindow:rec(report?.analysisWindow??report?.windows?.recent?.summary),pagePerformanceSlowApi:slow,browserProcess:{source:'analysis-browser-process-jsonl',unit:'MB',metric:'page-heap-used',pageCount:browserPageSeries.length,sampleCount:browserPageSeries.reduce((sum,item)=>sum+item.points.length,0),pageSeries:browserPageSeries,valuesRedacted:true},requestRate:compactRequestRate(report?.requestRate),valuesRedacted:true}));",
|
||||
"NODE",
|
||||
].join("\n");
|
||||
const result = runCommand(["trans", `${state.spec.nodeId}:${state.spec.workspace}`, "sh"], repoRoot, { input: script, timeoutMs: Math.min(timeoutSeconds, 60) * 1000 });
|
||||
|
||||
@@ -51,6 +51,7 @@ export interface MonitorCentralStore {
|
||||
schemaVersion(): Promise<string | null>;
|
||||
capability(): Promise<{ readonly ingest: boolean; readonly query: boolean }>;
|
||||
ingest(input: NormalizedMonitorCentralIngest): Promise<{ readonly disposition: "inserted" | "idempotent" }>;
|
||||
importSource(manifest: MonitorImportManifest, inputs: readonly NormalizedMonitorCentralIngest[]): Promise<{ readonly disposition: "inserted" | "idempotent"; readonly runs: readonly { readonly disposition: "inserted" | "idempotent" }[] }>;
|
||||
updateArtifactAvailability(scope: Required<MonitorScope>, runId: string, locatorIndex: number, reachable: boolean): Promise<void>;
|
||||
overview(scope: MonitorScope): Promise<Record<string, unknown>>;
|
||||
runs(scope: MonitorScope, options: RunQueryOptions): Promise<readonly MonitorRun[]>;
|
||||
@@ -63,6 +64,12 @@ export interface NormalizedMonitorCentralIngest extends Required<Omit<MonitorCen
|
||||
readonly payloadHash: string;
|
||||
}
|
||||
|
||||
export interface MonitorImportManifest {
|
||||
readonly manifestId: string;
|
||||
readonly sourceFingerprint: string;
|
||||
readonly rowCount: number;
|
||||
}
|
||||
|
||||
interface RunCursor {
|
||||
readonly updatedAt: string;
|
||||
readonly runId: string;
|
||||
@@ -186,7 +193,69 @@ export function createPostgresMonitorCentralStore(databaseUrl: string, options:
|
||||
return { ingest: rows[0]?.can_ingest === true, query: rows[0]?.can_query === true };
|
||||
},
|
||||
async ingest(input) {
|
||||
return sql.begin((transaction) => writePostgresIngest(transaction, input));
|
||||
},
|
||||
async importSource(manifest, inputs) {
|
||||
return sql.begin(async (transaction) => {
|
||||
const existing = await transaction<{ source_fingerprint: string; row_count: string | number }[]>`
|
||||
select source_fingerprint, row_count from monitor.import_manifests where manifest_id = ${manifest.manifestId} for update`;
|
||||
if (existing[0]) {
|
||||
if (existing[0].source_fingerprint !== manifest.sourceFingerprint || Number(existing[0].row_count) !== manifest.rowCount) {
|
||||
throw Object.assign(new Error("monitor import manifest conflict"), { code: "monitor_import_manifest_conflict" });
|
||||
}
|
||||
return { disposition: "idempotent" as const, runs: inputs.map(() => ({ disposition: "idempotent" as const })) };
|
||||
}
|
||||
const runs = [] as { disposition: "inserted" | "idempotent" }[];
|
||||
for (const input of inputs) runs.push(await writePostgresIngest(transaction, input));
|
||||
await transaction`
|
||||
insert into monitor.import_manifests (manifest_id, source_fingerprint, row_count, imported_at)
|
||||
values (${manifest.manifestId}, ${manifest.sourceFingerprint}, ${manifest.rowCount}, now())`;
|
||||
return { disposition: "inserted" as const, runs };
|
||||
});
|
||||
},
|
||||
async updateArtifactAvailability(scope, runId, locatorIndex, reachable) {
|
||||
await sql`
|
||||
update monitor.artifact_locators
|
||||
set reachable = ${reachable}
|
||||
where sentinel_id = ${scope.sentinelId} and node = ${scope.node} and lane = ${scope.lane} and run_id = ${runId} and locator_index = ${locatorIndex}`;
|
||||
},
|
||||
async overview(scope) {
|
||||
const rows = await queryRuns(sql, scope, { limit: 1, cursor: null });
|
||||
return { scope: publicScope(scope), runCount: await countRuns(sql, scope), latest: rows[0] ?? null };
|
||||
},
|
||||
async runs(scope, options) { return queryRuns(sql, scope, options); },
|
||||
async run(scope, runId) {
|
||||
const where = identityWhere(scope, runId);
|
||||
const rows = await sql.unsafe(`select r.sentinel_id, r.node, r.lane, r.run_id, r.updated_at, r.status, r.payload_hash, p.payload from monitor.runs r join monitor.run_payloads p using (sentinel_id, node, lane, run_id) where ${where.text} limit 1`, where.values);
|
||||
const row = rows[0] as Record<string, unknown> | undefined;
|
||||
if (!row) return null;
|
||||
const identity = rowIdentity(row);
|
||||
const [findings, locators, timeline] = await Promise.all([
|
||||
sql.unsafe(`select finding from monitor.findings where ${identity.text} order by finding_index`, identity.values),
|
||||
sql.unsafe(`select owner_node, owner_lane, owner_pvc, state_dir, relative_path, sha256, size_bytes, kind, reachable from monitor.artifact_locators where ${identity.text} order by locator_index`, identity.values),
|
||||
sql.unsafe(`select event from monitor.timeline_events where ${identity.text} order by event_index`, identity.values),
|
||||
]);
|
||||
return { ...runRow(row), payload: row.payload, findings: findings.map((item) => item.finding), artifactLocators: locators.map(locatorRow), timeline: timeline.map((item) => item.event) };
|
||||
},
|
||||
async view(scope, runId, view) {
|
||||
const run = await this.run(scope, runId);
|
||||
if (!run) return null;
|
||||
const views = record(run.payload.views);
|
||||
const value = views[view];
|
||||
return value === undefined ? null : { run: runFrom(run), view, value };
|
||||
},
|
||||
async findings(scope, options) {
|
||||
const where = scopeWhere(scope);
|
||||
const values: unknown[] = [...where.values];
|
||||
const cursor = options.cursor ? (() => { values.push(options.cursor.updatedAt, options.cursor.runId, options.cursor.findingIndex); return `and (r.updated_at, r.run_id, f.finding_index) < ($${values.length - 2}, $${values.length - 1}, $${values.length})`; })() : "";
|
||||
values.push(options.limit);
|
||||
const rows = await sql.unsafe(`select f.finding, f.finding_index, r.sentinel_id, r.node, r.lane, r.run_id, r.updated_at, r.status, r.payload_hash from monitor.findings f join monitor.runs r using (sentinel_id, node, lane, run_id) where ${where.text} ${cursor} order by r.updated_at desc, r.run_id desc, f.finding_index desc limit $${values.length}`, values);
|
||||
return rows.map(mapPostgresFindingRow);
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
async function writePostgresIngest(transaction: postgres.TransactionSql, input: NormalizedMonitorCentralIngest): Promise<{ disposition: "inserted" | "idempotent" }> {
|
||||
await transaction`
|
||||
insert into monitor.runners (sentinel_id, node, lane, updated_at)
|
||||
values (${input.sentinelId}, ${input.node}, ${input.lane}, ${input.updatedAt})
|
||||
@@ -218,52 +287,7 @@ export function createPostgresMonitorCentralStore(databaseUrl: string, options:
|
||||
for (const [index, event] of input.timeline.entries()) await transaction`
|
||||
insert into monitor.timeline_events (sentinel_id, node, lane, run_id, event_index, event)
|
||||
values (${input.sentinelId}, ${input.node}, ${input.lane}, ${input.runId}, ${index}, ${JSON.stringify(event)}::jsonb)`;
|
||||
return { disposition: "inserted" as const };
|
||||
});
|
||||
},
|
||||
async updateArtifactAvailability(scope, runId, locatorIndex, reachable) {
|
||||
await sql`
|
||||
update monitor.artifact_locators
|
||||
set reachable = ${reachable}
|
||||
where sentinel_id = ${scope.sentinelId} and node = ${scope.node} and lane = ${scope.lane} and run_id = ${runId} and locator_index = ${locatorIndex}`;
|
||||
},
|
||||
async overview(scope) {
|
||||
const rows = await queryRuns(sql, scope, { limit: 1, cursor: null });
|
||||
return { scope: publicScope(scope), runCount: await countRuns(sql, scope), latest: rows[0] ?? null };
|
||||
},
|
||||
async runs(scope, options) { return queryRuns(sql, scope, options); },
|
||||
async run(scope, runId) {
|
||||
const where = identityWhere(scope, runId);
|
||||
const rows = await sql.unsafe(`select r.sentinel_id, r.node, r.lane, r.run_id, r.updated_at, r.status, r.payload_hash, p.payload from monitor.runs r join monitor.run_payloads p using (sentinel_id, node, lane, run_id) where ${where.text} limit 1`, where.values);
|
||||
const row = rows[0] as Record<string, unknown> | undefined;
|
||||
if (!row) return null;
|
||||
const identity = rowIdentity(row);
|
||||
const [findings, locators, timeline] = await Promise.all([
|
||||
sql.unsafe(`select finding from monitor.findings where ${identity.text} order by finding_index`, identity.values),
|
||||
sql.unsafe(`select owner_node, owner_lane, owner_pvc, state_dir, relative_path, sha256, size_bytes, kind, reachable from monitor.artifact_locators where ${identity.text} order by locator_index`, identity.values),
|
||||
sql.unsafe(`select event from monitor.timeline_events where ${identity.text} order by event_index`, identity.values),
|
||||
]);
|
||||
return { ...runRow(row), payload: row.payload, findings: findings.map((item) => item.finding), artifactLocators: locators.map(locatorRow), timeline: timeline.map((item) => item.event) };
|
||||
},
|
||||
async view(scope, runId, view) {
|
||||
const detail = await this.run(scope, runId);
|
||||
const views = detail?.payload && typeof detail.payload === "object" ? (detail.payload as Record<string, unknown>).views : null;
|
||||
const value = views && typeof views === "object" ? (views as Record<string, unknown>)[view] : undefined;
|
||||
return value === undefined ? null : { runId, name: view, value };
|
||||
},
|
||||
async findings(scope, options) {
|
||||
const where = scopeWhere(scope);
|
||||
const values: unknown[] = [...where.values];
|
||||
let cursor = "";
|
||||
if (options.cursor) {
|
||||
values.push(options.cursor.updatedAt, options.cursor.runId, options.cursor.findingIndex);
|
||||
cursor = `and (r.updated_at, r.run_id, f.finding_index) < ($${values.length - 2}, $${values.length - 1}, $${values.length})`;
|
||||
}
|
||||
values.push(options.limit);
|
||||
const rows = await sql.unsafe(`select f.finding, f.finding_index, r.sentinel_id, r.node, r.lane, r.run_id, r.updated_at, r.status, r.payload_hash from monitor.findings f join monitor.runs r using (sentinel_id, node, lane, run_id) where ${where.text} ${cursor} order by r.updated_at desc, r.run_id desc, f.finding_index desc limit $${values.length}`, values);
|
||||
return rows.map(mapPostgresFindingRow);
|
||||
},
|
||||
};
|
||||
return { disposition: "inserted" as const };
|
||||
}
|
||||
|
||||
export const MONITOR_CENTRAL_MIGRATIONS = [
|
||||
@@ -281,6 +305,7 @@ export const MONITOR_CENTRAL_MIGRATIONS = [
|
||||
|
||||
export function createInMemoryMonitorCentralStore(): MonitorCentralStore {
|
||||
const records = new Map<string, NormalizedMonitorCentralIngest>();
|
||||
const manifests = new Map<string, MonitorImportManifest>();
|
||||
return {
|
||||
async migrate() {},
|
||||
async ping() {},
|
||||
@@ -297,6 +322,25 @@ export function createInMemoryMonitorCentralStore(): MonitorCentralStore {
|
||||
records.set(key, structuredClone(input));
|
||||
return { disposition: "inserted" };
|
||||
},
|
||||
async importSource(manifest, inputs) {
|
||||
const previous = manifests.get(manifest.manifestId);
|
||||
if (previous) {
|
||||
if (previous.sourceFingerprint !== manifest.sourceFingerprint || previous.rowCount !== manifest.rowCount) throw Object.assign(new Error("monitor import manifest conflict"), { code: "monitor_import_manifest_conflict" });
|
||||
return { disposition: "idempotent", runs: inputs.map(() => ({ disposition: "idempotent" as const })) };
|
||||
}
|
||||
const staged = new Map(records);
|
||||
const runs: { disposition: "inserted" | "idempotent" }[] = [];
|
||||
for (const input of inputs) {
|
||||
const existing = staged.get(identityKey(input));
|
||||
if (existing && existing.payloadHash !== input.payloadHash) throw conflict(input);
|
||||
if (existing) runs.push({ disposition: "idempotent" });
|
||||
else { staged.set(identityKey(input), structuredClone(input)); runs.push({ disposition: "inserted" }); }
|
||||
}
|
||||
records.clear();
|
||||
for (const [key, value] of staged) records.set(key, value);
|
||||
manifests.set(manifest.manifestId, { ...manifest });
|
||||
return { disposition: "inserted", runs };
|
||||
},
|
||||
async updateArtifactAvailability(scope, runId, locatorIndex, reachable) {
|
||||
const key = identityKey({ ...scope, runId });
|
||||
const existing = records.get(key);
|
||||
|
||||
@@ -2,14 +2,14 @@
|
||||
// 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 { dirname, join, resolve } from "node:path";
|
||||
import { Database } from "bun:sqlite";
|
||||
|
||||
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 MonitorSqliteSource { readonly sentinelId: string; readonly node: string; readonly lane: string; readonly path: string; readonly runtimeConfigRef: string; readonly stateRoot: string; readonly ownerPvc: string; }
|
||||
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;
|
||||
@@ -32,7 +32,7 @@ export function exportMonitorSqliteSnapshot(source: MonitorSqliteSource, outputP
|
||||
if (integrity !== "ok") throw new Error(`legacy SQLite integrity_check failed: ${integrity}`);
|
||||
const tables = (database.query("SELECT name, sql FROM sqlite_master WHERE type = 'table' AND name NOT LIKE 'sqlite_%' ORDER BY name").all() as { name: string; sql: string }[]).map((table) => ({ name: table.name, sql: table.sql, rowCount: Number(database.query(`SELECT COUNT(*) AS count FROM ${quoteIdentifier(table.name)}`).get()?.count ?? 0), rows: database.query(`SELECT * FROM ${quoteIdentifier(table.name)}`).all().map(normalizeRow) }));
|
||||
const unsigned = { schema: MONITOR_SQLITE_MIGRATION_SCHEMA, createdAt: new Date().toISOString(), source: { ...source, path }, fingerprint: `sha256:${sha256(sourceBytes)}`, integrity, byteLength: sourceBytes.byteLength, tables, artifactLocator: { path, sha256: `sha256:${sha256(sourceBytes)}`, sizeBytes: sourceBytes.byteLength, kind: "legacy-sqlite" } };
|
||||
const snapshot: MonitorSqliteSnapshot = { ...unsigned, manifestContentHash: `sha256:${sha256(Buffer.from(monitorCentralStableJson(unsigned)))}` };
|
||||
const snapshot: MonitorSqliteSnapshot = { ...unsigned, manifestContentHash: monitorSnapshotContentHash(unsigned) };
|
||||
writeSnapshot(outputPath, `${monitorCentralStableJson(snapshot)}\n`);
|
||||
return snapshot;
|
||||
} finally { database.close(); }
|
||||
@@ -43,7 +43,7 @@ export function verifyMonitorSqliteSnapshot(snapshot: MonitorSqliteSnapshot): Re
|
||||
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);
|
||||
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 manifestContentHashValid = snapshot.manifestContentHash === monitorSnapshotContentHash(snapshot);
|
||||
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 };
|
||||
}
|
||||
@@ -52,9 +52,11 @@ export async function importMonitorSqliteSnapshot(snapshot: MonitorSqliteSnapsho
|
||||
const verified = verifyMonitorSqliteSnapshot(snapshot);
|
||||
if (verified.ok !== true) throw Object.assign(new Error("snapshot verification failed"), { code: "monitor-migration-snapshot-invalid", verification: verified });
|
||||
const ingests = canonicalIngests(snapshot);
|
||||
const dispositions = await Promise.all(ingests.map((input) => store.ingest(input)));
|
||||
const manifest = { manifestId: snapshot.manifestContentHash, sourceFingerprint: snapshot.fingerprint, rowCount: snapshot.tables.reduce((total, table) => total + table.rowCount, 0) };
|
||||
const batch = await store.importSource(manifest, ingests);
|
||||
const dispositions = batch.runs;
|
||||
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 };
|
||||
return { ok: true, command: "web-probe sentinel migration import", source: snapshot.source, manifest, disposition: batch.disposition, imported: { inserted: dispositions.filter((item) => item.disposition === "inserted").length, idempotent: dispositions.filter((item) => item.disposition === "idempotent").length }, reconciliation, payloadHashCount: new Set(ingests.map((item) => item.payloadHash)).size, valuesRedacted: true };
|
||||
}
|
||||
|
||||
export function reconcileTables(tables: readonly MonitorSqliteTable[]): MonitorMigrationReconciliation {
|
||||
@@ -71,12 +73,17 @@ function canonicalIngests(snapshot: MonitorSqliteSnapshot): readonly ReturnType<
|
||||
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 report = metadataValue(metadata, `run.report.${runId}`);
|
||||
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: [] });
|
||||
const reportLocator = reportLocatorFromRun(row, snapshot);
|
||||
const payload = jsonField(payloads.find((item) => stringField(item, ["run_id"]) === runId), ["payload", "value_json"]) ?? { ...(report ?? {}), legacy: { run: row, sourceFingerprint: snapshot.fingerprint } };
|
||||
const reportFindings = Array.isArray(report?.findings) ? report.findings.filter(isRecord) : [];
|
||||
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.length ? rowFindings : reportFindings, artifactLocators: reportLocator ? [...rowLocators, reportLocator] : rowLocators, timeline: [] });
|
||||
});
|
||||
}
|
||||
function locatorFromRow(row: Record<string, unknown>, 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 locatorFromRow(row: Record<string, unknown>, source: MonitorSqliteSnapshot): MonitorArtifactLocator { const ownerPvc = stringField(row, ["owner_pvc", "ownerPvc"]); const relativePath = stringField(row, ["relative_path", "relativePath"]); const hash = stringField(row, ["sha256"]); const sizeBytes = numberField(row, ["size_bytes", "sizeBytes"]); if (!ownerPvc || !relativePath || !hash || sizeBytes === null) throw Object.assign(new Error("legacy artifact locator is incomplete"), { code: "monitor-migration-locator-invalid" }); return { ownerNode: stringField(row, ["owner_node", "ownerNode"]) || source.source.node, ownerLane: stringField(row, ["owner_lane", "ownerLane"]) || source.source.lane, ownerPvc, stateDir: stringField(row, ["state_dir", "stateDir"]), relativePath, sha256: hash, sizeBytes, kind: stringField(row, ["kind"]) || "artifact", reachable: row.reachable !== false }; }
|
||||
function reportLocatorFromRun(row: Record<string, unknown>, snapshot: MonitorSqliteSnapshot): MonitorArtifactLocator | null { const stateDir = stringField(row, ["state_dir", "stateDir"]); const expectedHash = stringField(row, ["report_json_sha256", "reportJsonSha256"]); if (!stateDir && !expectedHash) return null; if (!stateDir || !expectedHash) throw Object.assign(new Error("legacy report locator is incomplete"), { code: "monitor-migration-report-locator-invalid" }); const path = join(snapshot.source.stateRoot, stateDir, "analysis", "report.json"); const bytes = readFileSync(path); const actualHash = `sha256:${sha256(bytes)}`; if (actualHash !== expectedHash) throw Object.assign(new Error("legacy report hash does not match artifact"), { code: "monitor-migration-report-locator-invalid" }); return { ownerNode: snapshot.source.node, ownerLane: snapshot.source.lane, ownerPvc: snapshot.source.ownerPvc, stateDir, relativePath: "analysis/report.json", sha256: actualHash, sizeBytes: bytes.byteLength, kind: "report-json", reachable: true }; }
|
||||
function metadataValue(rows: readonly Record<string, unknown>[], key: string): Record<string, unknown> | null { return jsonField(rows.find((row) => stringField(row, ["key"]) === key), ["value_json", "value"]); }
|
||||
function tableRows(tables: readonly MonitorSqliteTable[], names: readonly string[]): readonly Record<string, unknown>[] { return tables.filter((table) => names.includes(table.name)).flatMap((table) => table.rows); }
|
||||
function sourcePlan(source: MonitorSqliteSource): Record<string, unknown> { const path = resolve(source.path); const state = existsSync(path) ? statSync(path) : null; return { sentinelId: source.sentinelId, node: source.node, lane: source.lane, path, runtimeConfigRef: source.runtimeConfigRef, exists: state !== null, isFile: state?.isFile() ?? false, byteLength: state?.size ?? null, mutation: false, valuesRedacted: true }; }
|
||||
function assertReadOnlySqliteSource(value: string): string { const path = resolve(value); const state = statSync(path); if (!state.isFile()) throw new Error(`legacy SQLite source is not a file: ${path}`); const descriptor = openSync(path, "r"); closeSync(descriptor); return path; }
|
||||
@@ -87,3 +94,5 @@ function jsonField(row: Record<string, unknown> | undefined, fields: readonly st
|
||||
function stringField(row: Record<string, unknown> | undefined, fields: readonly string[]): string { const value = row === undefined ? null : fields.map((field) => row[field]).find((item) => typeof item === "string" && item.length > 0); return typeof value === "string" ? value : ""; }
|
||||
function numberField(row: Record<string, unknown>, fields: readonly string[]): number | null { const value = fields.map((field) => row[field]).find((item) => typeof item === "number"); return typeof value === "number" && Number.isSafeInteger(value) && value >= 0 ? value : null; }
|
||||
function sha256(value: Uint8Array): string { return createHash("sha256").update(value).digest("hex"); }
|
||||
export function monitorSnapshotContentHash(snapshot: Omit<MonitorSqliteSnapshot, "manifestContentHash"> | MonitorSqliteSnapshot): string { const { manifestContentHash: _ignored, ...unsigned } = snapshot as MonitorSqliteSnapshot; return `sha256:${sha256(Buffer.from(monitorCentralStableJson(unsigned)))}`; }
|
||||
function isRecord(value: unknown): value is Record<string, unknown> { return typeof value === "object" && value !== null && !Array.isArray(value); }
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
// SPEC: PJ2026-01060508 Web哨兵 draft-2026-07-12-p17-monitor-central-persistence.
|
||||
import assert from "node:assert/strict";
|
||||
import { mkdtempSync, readFileSync, rmSync } from "node:fs";
|
||||
import { mkdirSync, mkdtempSync, readFileSync, rmSync, writeFileSync } from "node:fs";
|
||||
import { tmpdir } from "node:os";
|
||||
import { join } from "node:path";
|
||||
import test from "node:test";
|
||||
@@ -12,22 +12,27 @@ import { buildMonitorTerminalIngest, MonitorIngestFailedError, resolveMonitorTer
|
||||
|
||||
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" });
|
||||
const hash = `sha256:${"a".repeat(64)}`;
|
||||
const artifact = [{ relativePath: "analysis/report.json", kind: "report", sha256: hash, sizeBytes: 42 }];
|
||||
const ingest = buildMonitorTerminalIngest({ sentinelId: "fixture", node: "NC01", lane: "v03", runId: "run-1" }, report, artifact, { 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);
|
||||
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);
|
||||
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" }, artifact, { ownerNode: "NC01", ownerLane: "v03", ownerPvc: "fixture-pvc", stateDir: "/state" }).payloadHash);
|
||||
assert.throws(() => buildMonitorTerminalIngest({ sentinelId: "fixture", node: "NC01", lane: "v03", runId: "" }, report, artifact, { ownerNode: "NC01", ownerLane: "v03", ownerPvc: "fixture-pvc", stateDir: "/state" }), MonitorIngestFailedError);
|
||||
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("terminal writer defaults to legacy authority and validates central selection", () => {
|
||||
assert.deepEqual(resolveMonitorTerminalWriter({}), { mode: "legacy-sqlite" });
|
||||
test("terminal writer requires explicit mutually exclusive authority", () => {
|
||||
assert.throws(() => resolveMonitorTerminalWriter({}));
|
||||
assert.deepEqual(resolveMonitorTerminalWriter({ sqlite: { path: "/state/legacy.sqlite" } }), { 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" } } }));
|
||||
assert.throws(() => resolveMonitorTerminalWriter({ sqlite: { path: "/state/legacy.sqlite" }, monitor: { terminalWriter: { mode: "central", ingest: { endpoint: "https://monitor.example.test/api/ingest", timeoutMs: 1_000, retry: { maxAttempts: 2, delayMs: 0 } } } } }));
|
||||
});
|
||||
|
||||
test("SQLite snapshot reconciles the 228/242/234 fixture shape without production assumptions", async () => {
|
||||
@@ -36,14 +41,19 @@ test("SQLite snapshot reconciles the 228/242/234 fixture shape without productio
|
||||
const sqlitePath = join(directory, "legacy.sqlite");
|
||||
const snapshotPath = join(directory, "snapshot.json");
|
||||
const database = new Database(sqlitePath);
|
||||
database.run("CREATE TABLE runs (id TEXT)");
|
||||
const stateRoot = join(directory, "state");
|
||||
const reportPath = join(stateRoot, "run-0", "analysis", "report.json");
|
||||
mkdirSync(join(stateRoot, "run-0", "analysis"), { recursive: true });
|
||||
writeFileSync(reportPath, JSON.stringify({ updatedAt: "2026-07-12T00:00:00.000Z", views: { summary: { title: "legacy summary" } }, findings: [{ id: "report-finding" }] }));
|
||||
const reportHash = `sha256:${Bun.CryptoHasher.hash("sha256", readFileSync(reportPath), "hex")}`;
|
||||
database.run("CREATE TABLE runs (id TEXT, state_dir TEXT, report_json_sha256 TEXT, updated_at 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}`]);
|
||||
database.run("CREATE TABLE metadata (id TEXT, key TEXT, value_json TEXT)");
|
||||
for (let index = 0; index < 228; index += 1) database.run("INSERT INTO runs VALUES (?, ?, ?, ?)", [`run-${index}`, index === 0 ? "run-0" : null, index === 0 ? reportHash : null, "2026-07-12T00:00:00.000Z"]);
|
||||
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}`]);
|
||||
for (let index = 0; index < 234; index += 1) database.run("INSERT INTO metadata VALUES (?, ?, ?)", [`metadata-${index}`, index === 0 ? "run.report.run-0" : `metadata-${index}`, index === 0 ? JSON.stringify({ views: { summary: { title: "legacy summary" } }, findings: [{ id: "report-finding" }] }) : "{}"]);
|
||||
database.close();
|
||||
const source = { sentinelId: "fixture", node: "NC01", lane: "v03", path: sqlitePath, runtimeConfigRef: "fixture#runtime" };
|
||||
const source = { sentinelId: "fixture", node: "NC01", lane: "v03", path: sqlitePath, runtimeConfigRef: "fixture#runtime", stateRoot, ownerPvc: "fixture-pvc" };
|
||||
const plan = planMonitorSqliteSources([source]);
|
||||
assert.equal((plan.sources as { exists: boolean }[])[0].exists, true);
|
||||
const snapshot = exportMonitorSqliteSnapshot(source, snapshotPath);
|
||||
@@ -55,8 +65,26 @@ test("SQLite snapshot reconciles the 228/242/234 fixture shape without productio
|
||||
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 });
|
||||
const importedRun = await store.run({ sentinelId: "fixture", node: "NC01", lane: "v03" }, "run-0");
|
||||
assert.equal(importedRun?.artifactLocators[0]?.sha256, reportHash);
|
||||
assert.equal((await store.view({ sentinelId: "fixture", node: "NC01", lane: "v03" }, "run-0", "summary"))?.value.title, "legacy summary");
|
||||
assert.equal(verifyMonitorSqliteSnapshot({ ...snapshot, manifestContentHash: "sha256:tampered" }).ok, false);
|
||||
} finally {
|
||||
rmSync(directory, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
|
||||
test("single-source batch rolls back conflicts and preserves manifest idempotency", async () => {
|
||||
const store = createInMemoryMonitorCentralStore();
|
||||
const owner = { ownerNode: "NC01", ownerLane: "v03", ownerPvc: "fixture-pvc", stateDir: "run" };
|
||||
const artifact = [{ relativePath: "analysis/report.json", kind: "report-json", sha256: `sha256:${"b".repeat(64)}`, sizeBytes: 1 }];
|
||||
const first = buildMonitorTerminalIngest({ sentinelId: "fixture", node: "NC01", lane: "v03", runId: "first" }, { updatedAt: "2026-07-12T00:00:00.000Z" }, artifact, owner);
|
||||
const existing = buildMonitorTerminalIngest({ sentinelId: "fixture", node: "NC01", lane: "v03", runId: "existing" }, { updatedAt: "2026-07-12T00:00:00.000Z", status: "ok" }, artifact, owner);
|
||||
const conflicting = buildMonitorTerminalIngest({ sentinelId: "fixture", node: "NC01", lane: "v03", runId: "existing" }, { updatedAt: "2026-07-12T00:00:00.000Z", status: "blocked" }, artifact, owner);
|
||||
await store.ingest(existing);
|
||||
await assert.rejects(() => store.importSource({ manifestId: "sha256:batch-conflict", sourceFingerprint: "sha256:source", rowCount: 2 }, [first, conflicting]), (error: unknown) => (error as { code?: string }).code === "payload_hash_conflict");
|
||||
assert.equal(await store.run({ sentinelId: "fixture", node: "NC01", lane: "v03" }, "first"), null);
|
||||
const manifest = { manifestId: "sha256:batch-ok", sourceFingerprint: "sha256:source", rowCount: 1 };
|
||||
assert.equal((await store.importSource(manifest, [first])).disposition, "inserted");
|
||||
assert.equal((await store.importSource(manifest, [first])).disposition, "idempotent");
|
||||
});
|
||||
|
||||
@@ -15,9 +15,13 @@ export class MonitorIngestFailedError extends Error {
|
||||
|
||||
export function buildMonitorTerminalIngest(identity: MonitorTerminalIdentity, report: Record<string, unknown>, artifacts: readonly MonitorTerminalArtifact[], locatorOwner: Pick<MonitorArtifactLocator, "ownerNode" | "ownerLane" | "ownerPvc" | "stateDir">): MonitorCentralIngest {
|
||||
const immutableReport = JSON.parse(monitorCentralStableJson(report)) as Record<string, unknown>;
|
||||
if (!identity.runId) throw new MonitorIngestFailedError({ reason: "run_id_missing", valuesRedacted: true });
|
||||
const updatedAt = stringValue(immutableReport.updatedAt) ?? stringValue(immutableReport.analyzedAt);
|
||||
if (!updatedAt) throw new MonitorIngestFailedError({ reason: "terminal_timestamp_missing", runId: identity.runId, valuesRedacted: true });
|
||||
for (const artifact of artifacts) if (!/^sha256:[a-f0-9]{64}$/u.test(artifact.sha256) || !Number.isSafeInteger(artifact.sizeBytes) || artifact.sizeBytes < 1) throw new MonitorIngestFailedError({ reason: "terminal_artifact_unverified", runId: identity.runId, relativePath: artifact.relativePath, valuesRedacted: true });
|
||||
return normalizeMonitorCentralIngest({
|
||||
...identity,
|
||||
updatedAt: stringValue(immutableReport.updatedAt) ?? stringValue(immutableReport.analyzedAt) ?? new Date().toISOString(),
|
||||
updatedAt,
|
||||
status: stringValue(immutableReport.status) ?? (immutableReport.ok === true ? "ok" : "blocked"),
|
||||
payload: { report: immutableReport, artifacts: artifacts.map((artifact) => ({ ...artifact })) },
|
||||
findings: recordArray(immutableReport.findings),
|
||||
@@ -57,13 +61,22 @@ function sleep(delayMs: number): Promise<void> { return new Promise((resolve) =>
|
||||
export interface MonitorTerminalWriterConfig { readonly mode: "legacy-sqlite" | "central"; readonly ingest?: MonitorTerminalIngestConfig; }
|
||||
|
||||
export function resolveMonitorTerminalWriter(runtime: Record<string, unknown>): MonitorTerminalWriterConfig {
|
||||
const sqlite = runtime.sqlite;
|
||||
const sqlitePath = sqlite && typeof sqlite === "object" && !Array.isArray(sqlite) ? stringValue((sqlite as Record<string, unknown>).path) : undefined;
|
||||
const monitor = runtime.monitor;
|
||||
const writer = monitor && typeof monitor === "object" && !Array.isArray(monitor) ? (monitor as Record<string, unknown>).terminalWriter : null;
|
||||
if (!writer || typeof writer !== "object" || Array.isArray(writer)) return { mode: "legacy-sqlite" };
|
||||
if (!writer || typeof writer !== "object" || Array.isArray(writer)) {
|
||||
if (sqlitePath) return { mode: "legacy-sqlite" };
|
||||
throw new Error("terminal writer authority is undeclared: sqlite.path or monitor.terminalWriter is required");
|
||||
}
|
||||
const value = writer as Record<string, unknown>;
|
||||
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 };
|
||||
if (mode === "legacy-sqlite") {
|
||||
if (!sqlitePath) throw new Error("sqlite.path is required when mode=legacy-sqlite");
|
||||
return { mode };
|
||||
}
|
||||
if (sqlitePath) throw new Error("central terminal writer and sqlite.path are mutually exclusive");
|
||||
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<string, unknown>;
|
||||
|
||||
Reference in New Issue
Block a user