AgentRun Codex
215 lines
19 KiB
TypeScript
215 lines
19 KiB
TypeScript
// SPEC: PJ2026-01060508 Web哨兵 draft-2026-07-12-p17-monitor-central-persistence.
|
|
import assert from "node:assert/strict";
|
|
import { createHash } from "node:crypto";
|
|
import { mkdirSync, mkdtempSync, readFileSync, rmSync, writeFileSync } from "node:fs";
|
|
import { tmpdir } from "node:os";
|
|
import { join } from "node:path";
|
|
import test from "node:test";
|
|
import { Database } from "bun:sqlite";
|
|
|
|
import { createInMemoryMonitorCentralStore } from "./monitor-central-persistence";
|
|
import { exportMonitorSqliteSnapshot, importMonitorSqliteSnapshot, planMonitorSqliteSources, verifyMonitorSqliteSnapshot, verifyMonitorSqliteSnapshotAgainstStore } from "./monitor-sqlite-migration";
|
|
import { buildMonitorTerminalIngest, MonitorIngestFailedError, resolveMonitorTerminalWriter, submitMonitorTerminalIngest } from "./monitor-terminal-ingest";
|
|
import { compactWebObserveAnalyzeAnalysisForRaw } from "./hwlab-node/web-probe-observe";
|
|
|
|
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 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" }, 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 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("real compact analyze projects stable report time and artifact facts", () => {
|
|
const compact = compactWebObserveAnalyzeAnalysisForRaw({ generatedAt: "2026-07-12T00:00:00.000Z", analyzer: { reportJsonPath: "/state/run-1/analysis/report.json", reportJsonSha256: `sha256:${"c".repeat(64)}`, reportJsonBytes: 12 } });
|
|
assert.deepEqual({ reportUpdatedAt: compact.reportUpdatedAt, reportJsonSha256: compact.reportJsonSha256, reportJsonBytes: compact.reportJsonBytes }, { reportUpdatedAt: "2026-07-12T00:00:00.000Z", reportJsonSha256: `sha256:${"c".repeat(64)}`, reportJsonBytes: 12 });
|
|
const ingest = buildMonitorTerminalIngest({ sentinelId: "fixture", node: "NC01", lane: "v03", runId: "generated-at" }, compact, [{ relativePath: "analysis/report.json", kind: "report-json", sha256: `sha256:${"c".repeat(64)}`, sizeBytes: 12 }], { ownerNode: "NC01", ownerLane: "v03", ownerPvc: "fixture-pvc", stateDir: "/state/run-1" });
|
|
assert.equal(ingest.updatedAt, "2026-07-12T00:00:00.000Z");
|
|
});
|
|
|
|
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");
|
|
const snapshotPath = join(directory, "snapshot.json");
|
|
const database = new Database(sqlitePath);
|
|
const stateRoot = join(directory, "state");
|
|
const workspaceRoot = join(directory, "hwlab-v03");
|
|
const workspaceRun = join(workspaceRoot, "web-observe", "run-0");
|
|
const reportPath = join(workspaceRun, "analysis", "report.json");
|
|
mkdirSync(join(workspaceRun, "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, artifact_owner_pvc TEXT, artifact_count INTEGER, report_json_sha256 TEXT, updated_at TEXT)");
|
|
database.run("CREATE TABLE findings (run_id TEXT, sentinel_id TEXT, finding_id TEXT, severity TEXT, count INTEGER, summary TEXT, report_json_sha256 TEXT, created_at TEXT)");
|
|
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 ? workspaceRun : null, index === 0 ? "legacy-web-observe-workspace-nc01" : null, index === 0 ? 1 : 0, index === 0 ? reportHash : null, "2026-07-12T00:00:00.000Z"]);
|
|
for (let index = 0; index < 242; index += 1) database.run("INSERT INTO findings VALUES (?, ?, ?, ?, ?, ?, ?, ?)", [index === 0 ? "run-0" : `run-${index % 228}`, "fixture", index === 0 ? "report-finding" : `finding-${index}`, "warning", 1, "indexed", reportHash, "2026-07-12T00:00:00.000Z"]);
|
|
for (let index = 0; index < 234; index += 1) database.run("INSERT INTO metadata VALUES (?, ?, ?)", [`metadata-${index}`, index === 0 ? "run.report.run-0" : index === 1 ? "scheduler.heartbeat" : index === 2 ? "maintenance" : index === 3 ? "manual-trigger.latest" : `metadata-${index}`, index === 0 ? JSON.stringify({ views: { summary: { title: "legacy summary" } }, findings: [{ id: "report-finding", rootCause: "complete metadata detail" }] }) : index === 2 ? JSON.stringify(false) : JSON.stringify({ index })]);
|
|
database.close();
|
|
const source = { sentinelId: "fixture", node: "NC01", lane: "v03", path: sqlitePath, runtimeConfigRef: "fixture#runtime", stateRoot, ownerPvc: "fixture-pvc", artifactSources: [{ ownerPvc: "fixture-pvc", mountPath: stateRoot }, { ownerPvc: "legacy-web-observe-workspace-nc01", mountPath: workspaceRoot, hostPath: workspaceRoot }] };
|
|
const plan = planMonitorSqliteSources([source]);
|
|
assert.equal((plan.sources as { exists: boolean }[])[0].exists, true);
|
|
const snapshot = exportMonitorSqliteSnapshot(source, snapshotPath);
|
|
const verified = verifyMonitorSqliteSnapshot(snapshot);
|
|
assert.equal(verified.ok, true);
|
|
assert.deepEqual(verified.reconciliation, { runs: 228, findings: 242, metadata: 234, payloads: 0, locators: 0, latest: 0 });
|
|
const persisted = JSON.parse(readFileSync(snapshotPath, "utf8"));
|
|
assert.equal(persisted.schema, snapshot.schema);
|
|
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 importedMetadata = await store.runtimeMetadata({ sentinelId: "fixture", node: "NC01", lane: "v03" });
|
|
assert.equal(importedMetadata.length, 234);
|
|
assert.equal(importedMetadata.find((item) => item.key === "scheduler.heartbeat")?.updatedAt, snapshot.createdAt);
|
|
const importedRun = await store.run({ sentinelId: "fixture", node: "NC01", lane: "v03" }, "run-0");
|
|
assert.equal(importedRun?.artifactLocators[0]?.sha256, reportHash);
|
|
assert.equal(importedRun?.artifactLocators[0]?.ownerPvc, "legacy-web-observe-workspace-nc01");
|
|
const reportFindings = importedRun?.findings.filter((finding) => finding.id === "report-finding") ?? [];
|
|
assert.deepEqual(reportFindings, [{ id: "report-finding", rootCause: "complete metadata detail" }]);
|
|
assert.equal((await store.view({ sentinelId: "fixture", node: "NC01", lane: "v03" }, "run-0", "summary"))?.value.title, "legacy summary");
|
|
const pgVerification = await verifyMonitorSqliteSnapshotAgainstStore(snapshot, store);
|
|
assert.equal(pgVerification.ok, true);
|
|
assert.deepEqual([...new Set((pgVerification.checks as { name: string }[]).map((item) => item.name))].sort(), ["artifactLocator", "finding", "globalOverview", "run", "runtimeMetadata", "scopedOverview", "view"]);
|
|
assert.deepEqual((pgVerification.reconciliation as { target: Record<string, number> }).target, { runs: 228, findings: 242, payloads: 228, artifactLocators: 1, views: 1, runtimeMetadata: 234 });
|
|
await store.importSource({ manifestId: "sha256:orphan-metadata", sourceFingerprint: "sha256:orphan-metadata", rowCount: 0 }, [], [{ sentinelId: "fixture", node: "NC01", lane: "v03", key: "orphan", value: { unexpected: true }, updatedAt: snapshot.createdAt }]);
|
|
assert.equal((await verifyMonitorSqliteSnapshotAgainstStore(snapshot, store)).ok, false);
|
|
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");
|
|
});
|
|
|
|
test("metadata-only snapshot verifies scoped runtime metadata and rejects extras", async () => {
|
|
const directory = mkdtempSync(join(tmpdir(), "monitor-metadata-only-"));
|
|
try {
|
|
const sqlitePath = join(directory, "index.sqlite");
|
|
const snapshotPath = join(directory, "snapshot.json");
|
|
const database = new Database(sqlitePath);
|
|
database.run("CREATE TABLE metadata (key TEXT, value_json TEXT)");
|
|
database.run("INSERT INTO metadata VALUES (?, ?)", ["scheduler.heartbeat", JSON.stringify({ healthy: true })]);
|
|
database.close();
|
|
const source = { sentinelId: "fixture", node: "NC01", lane: "v03", path: sqlitePath, runtimeConfigRef: "fixture#runtime", stateRoot: directory, ownerPvc: "fixture-pvc", artifactSources: [{ ownerPvc: "fixture-pvc", mountPath: directory }] };
|
|
const snapshot = exportMonitorSqliteSnapshot(source, snapshotPath);
|
|
const store = createInMemoryMonitorCentralStore();
|
|
await importMonitorSqliteSnapshot(snapshot, store);
|
|
assert.equal((await verifyMonitorSqliteSnapshotAgainstStore(snapshot, store)).ok, true);
|
|
await store.importSource({ manifestId: "sha256:metadata-only-extra", sourceFingerprint: "sha256:metadata-only-extra", rowCount: 0 }, [], [{ sentinelId: "fixture", node: "NC01", lane: "v03", key: "orphan", value: true, updatedAt: snapshot.createdAt }]);
|
|
assert.equal((await verifyMonitorSqliteSnapshotAgainstStore(snapshot, store)).ok, false);
|
|
} finally {
|
|
rmSync(directory, { recursive: true, force: true });
|
|
}
|
|
});
|
|
|
|
test("WAL export freezes a serialized SQLite snapshot independent of later live writes", () => {
|
|
const directory = mkdtempSync(join(tmpdir(), "monitor-wal-snapshot-"));
|
|
try {
|
|
const sqlitePath = join(directory, "index.sqlite");
|
|
const snapshotPath = join(directory, "snapshot.json");
|
|
const database = new Database(sqlitePath);
|
|
database.exec("PRAGMA journal_mode = WAL");
|
|
database.run("CREATE TABLE metadata (key TEXT, value_json TEXT)");
|
|
database.run("INSERT INTO metadata VALUES (?, ?)", ["scheduler.heartbeat", JSON.stringify({ sequence: 1 })]);
|
|
const source = { sentinelId: "fixture", node: "NC01", lane: "v03", path: sqlitePath, runtimeConfigRef: "fixture#runtime", stateRoot: directory, ownerPvc: "fixture-pvc", artifactSources: [{ ownerPvc: "fixture-pvc", mountPath: directory }] };
|
|
const snapshot = exportMonitorSqliteSnapshot(source, snapshotPath);
|
|
database.run("INSERT INTO metadata VALUES (?, ?)", ["scheduler.heartbeat.late", JSON.stringify({ sequence: 2 })]);
|
|
database.close();
|
|
|
|
assert.equal(snapshot.artifactLocator.path, join(directory, "snapshot.sqlite"));
|
|
assert.equal(snapshot.tables.find((table) => table.name === "metadata")?.rowCount, 1);
|
|
assert.equal(verifyMonitorSqliteSnapshot(snapshot).ok, true);
|
|
} finally {
|
|
rmSync(directory, { recursive: true, force: true });
|
|
}
|
|
});
|
|
|
|
test("workspace artifact source verifies locator identity hash and size", async () => {
|
|
const directory = mkdtempSync(join(tmpdir(), "monitor-workspace-artifact-"));
|
|
try {
|
|
const sqlitePath = join(directory, "index.sqlite");
|
|
const snapshotPath = join(directory, "snapshot.json");
|
|
const workspaceRoot = join(directory, "hwlab-v03");
|
|
const stateDir = join(workspaceRoot, "web-observe", "run-1");
|
|
const artifacts = [
|
|
["analysis/report.json", Buffer.from('{"ok":true}\n')],
|
|
["analysis/details.json", Buffer.from('{"details":true}\n')],
|
|
["artifacts/request.txt", Buffer.from("request\n")],
|
|
["artifacts/response.txt", Buffer.from("response\n")],
|
|
["screenshots/page.png", Buffer.from([0x89, 0x50, 0x4e, 0x47])],
|
|
] as const;
|
|
const reportBytes = artifacts[0][1];
|
|
mkdirSync(join(stateDir, "analysis"), { recursive: true });
|
|
for (const [relativePath, bytes] of artifacts) {
|
|
const path = join(stateDir, relativePath);
|
|
mkdirSync(join(path, ".."), { recursive: true });
|
|
writeFileSync(path, bytes);
|
|
}
|
|
const reportHash = `sha256:${createHash("sha256").update(reportBytes).digest("hex")}`;
|
|
const database = new Database(sqlitePath);
|
|
database.run("CREATE TABLE runs (id TEXT, state_dir TEXT, artifact_count INTEGER, report_json_sha256 TEXT, updated_at TEXT)");
|
|
database.run("INSERT INTO runs VALUES (?, ?, ?, ?, ?)", ["run-1", stateDir, 5, reportHash, "2026-07-12T00:00:00.000Z"]);
|
|
database.close();
|
|
const source = { sentinelId: "fixture", node: "NC01", lane: "v03", path: sqlitePath, runtimeConfigRef: "fixture#runtime", stateRoot: directory, ownerPvc: "fixture-pvc", artifactSources: [{ ownerPvc: "fixture-pvc", mountPath: directory }, { ownerPvc: "legacy-web-observe-workspace-nc01", mountPath: workspaceRoot, hostPath: workspaceRoot }] };
|
|
const snapshot = exportMonitorSqliteSnapshot(source, snapshotPath);
|
|
const store = createInMemoryMonitorCentralStore();
|
|
await importMonitorSqliteSnapshot(snapshot, store);
|
|
const run = await store.run({ sentinelId: "fixture", node: "NC01", lane: "v03" }, "run-1");
|
|
assert.equal(run?.artifactLocators.length, 5);
|
|
assert.deepEqual(run?.artifactLocators.map((locator) => ({ ownerPvc: locator.ownerPvc, stateDir: locator.stateDir, relativePath: locator.relativePath, sha256: locator.sha256, sizeBytes: locator.sizeBytes })), artifacts.map(([relativePath, bytes]) => ({ ownerPvc: "legacy-web-observe-workspace-nc01", stateDir, relativePath, sha256: `sha256:${createHash("sha256").update(bytes).digest("hex")}`, sizeBytes: bytes.byteLength })).sort((left, right) => left.relativePath.localeCompare(right.relativePath)));
|
|
assert.equal((await verifyMonitorSqliteSnapshotAgainstStore(snapshot, store)).ok, true);
|
|
rmSync(join(stateDir, "artifacts", "response.txt"));
|
|
await assert.rejects(() => importMonitorSqliteSnapshot(snapshot, createInMemoryMonitorCentralStore()), (error: unknown) => { const migrationError = error as { code?: string; diagnostic?: Record<string, unknown> }; return migrationError.code === "monitor-migration-artifact-count-mismatch" && migrationError.diagnostic?.runId === "run-1" && migrationError.diagnostic?.stateDir === stateDir && migrationError.diagnostic?.expectedCount === 5 && migrationError.diagnostic?.observedCount === 4; });
|
|
} finally {
|
|
rmSync(directory, { recursive: true, force: true });
|
|
}
|
|
});
|
|
|
|
test("artifact count without a registry-backed locator fails closed", async () => {
|
|
const directory = mkdtempSync(join(tmpdir(), "monitor-artifact-count-"));
|
|
try {
|
|
const sqlitePath = join(directory, "index.sqlite");
|
|
const snapshotPath = join(directory, "snapshot.json");
|
|
const database = new Database(sqlitePath);
|
|
database.run("CREATE TABLE runs (id TEXT, artifact_count INTEGER, updated_at TEXT)");
|
|
database.run("INSERT INTO runs VALUES (?, ?, ?)", ["run-1", 1, "2026-07-12T00:00:00.000Z"]);
|
|
database.close();
|
|
const source = { sentinelId: "fixture", node: "NC01", lane: "v03", path: sqlitePath, runtimeConfigRef: "fixture#runtime", stateRoot: directory, ownerPvc: "fixture-pvc", artifactSources: [{ ownerPvc: "fixture-pvc", mountPath: directory }] };
|
|
const snapshot = exportMonitorSqliteSnapshot(source, snapshotPath);
|
|
await assert.rejects(() => importMonitorSqliteSnapshot(snapshot, createInMemoryMonitorCentralStore()), (error: unknown) => (error as { code?: string }).code === "monitor-migration-artifact-count-mismatch");
|
|
} finally {
|
|
rmSync(directory, { recursive: true, force: true });
|
|
}
|
|
});
|