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

fix(monitor): harden central ingest contracts

Browse Source
This commit is contained in:
AgentRun Codex
2026-07-12 20:16:59 +00:00
parent a9587fcae2
commit 3a4e41a248
3 changed files with 252 additions and 204 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scripts/src/monitor-central-persistence.test.ts
+62 -46
View File
@@ -2,26 +2,42 @@ import assert from "node:assert/strict";
import { spawnSync } from "node:child_process";
import { test } from "bun:test";
import { createInMemoryMonitorCentralStore, createMonitorCentralService, MONITOR_CENTRAL_MIGRATIONS, MonitorCentralConfigurationError } from "./monitor-central-persistence";
import { createInMemoryMonitorCentralStore, createMonitorCentralService, mapPostgresFindingRow, MONITOR_CENTRAL_MIGRATIONS, MonitorCentralConfigurationError } from "./monitor-central-persistence";
function service() { return createMonitorCentralService({ store: createInMemoryMonitorCentralStore() }); }
async function ingest(target: ReturnType<typeof service>, runId: string, updatedAt: string, scope: Record<string, string> = {}) {
return target.ingest({ sentinelId: scope.sentinelId ?? "sentinel-a", node: scope.node ?? "NC01", lane: scope.lane ?? "v03", runId, updatedAt, status: "succeeded", payload: { runId, views: { summary: { state: "ready" } } }, findings: [{ id: `finding-${runId}` }], artifactLocators: [{ locator: `s3://evidence/${runId}`, sha256: "sha256:abc", owner: "runner", sizeBytes: 12, kind: "report" }], timeline: [{ phase: "done" }] });
}
function service() { return createMonitorCentralService({ store: createInMemoryMonitorCentralStore(), pageSize: 50 }); }
function locator(runId: string) { return { ownerNode: "NC01", ownerLane: "v03", ownerPvc: "sentinel-pvc", stateDir: `/state/${runId}`, relativePath: "analysis/report.json", sha256: "sha256:abc", sizeBytes: 12, kind: "report", reachable: true }; }
function input(runId: string, updatedAt = "2026-07-12T10:00:00.000Z", scope: Record<string, string> = {}) { return { sentinelId: scope.sentinelId ?? "sentinel-a", node: scope.node ?? "NC01", lane: scope.lane ?? "v03", runId, updatedAt, status: "succeeded", payload: { runId, views: { summary: { state: "ready" } } }, findings: [{ id: `${runId}-0` }, { id: `${runId}-1` }], artifactLocators: [locator(runId)], timeline: [{ phase: "done" }] }; }
async function ingest(target: ReturnType<typeof service>, runId: string, updatedAt?: string, scope?: Record<string, string>) { return target.ingest(input(runId, updatedAt, scope)); }
test("schema covers central persistence objects and does not store artifact blobs", () => {
test("schema covers runner FK, locator reachability, and atomic insert contract", () => {
const schema = MONITOR_CENTRAL_MIGRATIONS.join("\n");
for (const name of ["schema_migrations", "runners", "runs", "findings", "run_payloads", "artifact_locators", "timeline_events", "import_manifests"]) assert.match(schema, new RegExp(`monitor\\.${name}`));
assert.doesNotMatch(schema, /blob|bytea/i);
assert.match(schema, /foreign key \(sentinel_id, node, lane\) references monitor\.runners/u);
assert.match(schema, /owner_pvc.*state_dir.*relative_path.*reachable/u);
assert.doesNotMatch(schema, /blob|bytea/iu);
});
test("ingest is transactional in contract, idempotent by payload hash, and rejects conflicts", async () => {
test("HTTP terminal ingest is unbound-safe, ignores caller hash, and maps conflict to 409", async () => {
const target = service();
const first = await ingest(target, "run-1", "2026-07-12T10:00:00.000Z");
const repeated = await ingest(target, "run-1", "2026-07-12T10:00:00.000Z");
assert.equal(first.disposition, "inserted");
assert.equal(repeated.disposition, "idempotent");
await assert.rejects(() => target.ingest({ sentinelId: "sentinel-a", node: "NC01", lane: "v03", runId: "run-1", payload: { changed: true } }), { code: "payload_hash_conflict" });
const fetch = target.fetch;
const first = await fetch(new Request("http://local/api/ingest", { method: "POST", body: JSON.stringify(input("run-1")) }));
const repeated = await fetch(new Request("http://local/api/ingest", { method: "POST", body: JSON.stringify({ ...input("run-1"), payloadHash: "sha256:forged" }) }));
const conflict = await fetch(new Request("http://local/api/ingest", { method: "POST", body: JSON.stringify({ ...input("run-1"), status: "blocked", payloadHash: "sha256:forged" }) }));
assert.equal(first.status, 202);
assert.equal((await first.json() as any).disposition, "inserted");
assert.equal((await repeated.json() as any).disposition, "idempotent");
assert.equal(conflict.status, 409);
assert.equal((await conflict.json() as any).error.code, "payload_hash_conflict");
assert.equal((await fetch(new Request("http://local/health"))).status, 200);
});
test("full terminal envelope controls hash and concurrent same identity converges", async () => {
const target = service();
const base = { ...input("run-envelope"), payloadHash: "sha256:caller-controlled" };
const results = await Promise.all([target.ingest(base), target.ingest(base)]);
assert.deepEqual(new Set(results.map((item) => item.disposition)), new Set(["inserted", "idempotent"]));
await assert.rejects(() => target.ingest({ ...base, findings: [{ id: "changed" }] }), { code: "payload_hash_conflict" });
await assert.rejects(() => target.ingest({ ...base, artifactLocators: [{ ...locator("run-envelope"), reachable: false }] }), { code: "payload_hash_conflict" });
});
test("global and scoped latest use updatedAt then runId descending", async () => {
@@ -35,47 +51,47 @@ test("global and scoped latest use updatedAt then runId descending", async () =>
assert.equal(scoped.overview.latest.runId, "run-c");
});
test("runs and findings are bounded, cursor-paged, and scope-isolated", async () => {
test("detail and view require full composite identity", async () => {
const target = service();
await ingest(target, "run-1", "2026-07-12T10:00:00.000Z");
await ingest(target, "run-2", "2026-07-12T11:00:00.000Z");
await ingest(target, "run-3", "2026-07-12T12:00:00.000Z", { lane: "v04" });
const first = await (await target.fetch(new Request("http://local/api/runs?lane=v03&limit=1"))).json() as any;
const second = await (await target.fetch(new Request(`http://local/api/runs?lane=v03&limit=1000&cursor=${first.nextCursor}`))).json() as any;
const findings = await (await target.fetch(new Request("http://local/api/findings?lane=v03"))).json() as any;
assert.equal(first.items.length, 1);
assert.equal(first.items[0].runId, "run-2");
assert.equal(second.items.length, 1);
assert.equal(second.items[0].runId, "run-1");
assert.equal(findings.items.length, 2);
await ingest(target, "shared", undefined, { sentinelId: "sentinel-a" });
await ingest(target, "shared", undefined, { sentinelId: "sentinel-b" });
const missing = await target.fetch(new Request("http://local/api/runs/shared"));
const detail = await (await target.fetch(new Request("http://local/api/runs/shared?sentinelId=sentinel-b&node=NC01&lane=v03"))).json() as any;
const view = await (await target.fetch(new Request("http://local/api/runs/shared/views/summary?sentinelId=sentinel-b&node=NC01&lane=v03"))).json() as any;
assert.equal(missing.status, 400);
assert.equal((await missing.json() as any).error.code, "scope_required");
assert.equal(detail.run.sentinelId, "sentinel-b");
assert.equal(view.view.value.state, "ready");
});
test("run detail returns payload, locator metadata, timeline, and no secret values", async () => {
test("findings cursor contains finding index and does not skip a same-run page", async () => {
const target = service();
await ingest(target, "run-detail", "2026-07-12T10:00:00.000Z");
const response = await (await target.fetch(new Request("http://local/api/runs/run-detail"))).json() as any;
assert.equal(response.run.payload.runId, "run-detail");
assert.deepEqual(response.run.artifactLocators[0], { locator: "s3://evidence/run-detail", sha256: "sha256:abc", owner: "runner", sizeBytes: 12, kind: "report" });
await ingest(target, "run-findings");
const first = await (await target.fetch(new Request("http://local/api/findings?limit=1"))).json() as any;
const second = await (await target.fetch(new Request(`http://local/api/findings?limit=1&cursor=${first.nextCursor}`))).json() as any;
assert.equal(first.items[0].findingIndex, 1);
assert.equal(second.items[0].findingIndex, 0);
});
test("PG finding mapping preserves selected status and payload hash", () => {
const row = mapPostgresFindingRow({ sentinel_id: "sentinel-a", node: "NC01", lane: "v03", run_id: "run-pg", updated_at: "2026-07-12T10:00:00.000Z", status: "blocked", payload_hash: "sha256:terminal", finding_index: 2, finding: { id: "pg" } });
assert.deepEqual(row, { sentinelId: "sentinel-a", node: "NC01", lane: "v03", runId: "run-pg", updatedAt: "2026-07-12T10:00:00.000Z", status: "blocked", payloadHash: "sha256:terminal", findingIndex: 2, finding: { id: "pg" } });
});
test("run detail returns normalized locator metadata without artifact bytes", async () => {
const target = service();
await ingest(target, "run-detail");
const response = await (await target.fetch(new Request("http://local/api/runs/run-detail?sentinelId=sentinel-a&node=NC01&lane=v03"))).json() as any;
assert.deepEqual(response.run.artifactLocators[0], locator("run-detail"));
assert.equal(response.valuesRedacted, true);
});
test("health is redacted and missing DATABASE_URL fails structurally", async () => {
test("configuration is explicit and missing environment projection fails structurally", async () => {
const target = service();
const health = await target.health();
assert.equal(health.ok, true);
assert.equal(health.valuesRedacted, true);
assert.throws(() => createMonitorCentralService({ databaseUrl: "" }), MonitorCentralConfigurationError);
});
test("view contract reshapes an existing report payload without reading runner artifacts", async () => {
const target = service();
await ingest(target, "run-view", "2026-07-12T10:00:00.000Z");
const response = await (await target.fetch(new Request("http://local/api/runs/run-view/views/summary"))).json() as any;
assert.deepEqual(response.view, { runId: "run-view", name: "summary", value: { state: "ready" } });
});
test("service entry exits with a structured redacted error when DATABASE_URL is absent", () => {
assert.equal((await target.health()).ok, true);
assert.throws(() => createMonitorCentralService({ databaseUrl: "", pageSize: 50 }), MonitorCentralConfigurationError);
assert.throws(() => createMonitorCentralService({ store: createInMemoryMonitorCentralStore() }), { code: "monitor_central_page_size_missing" });
const result = spawnSync("bun", ["scripts/monitor-central-service.ts"], { cwd: process.cwd(), encoding: "utf8", env: { PATH: process.env.PATH ?? "" } });
assert.equal(result.status, 2);
assert.deepEqual(JSON.parse(result.stdout), { ok: false, error: { code: "monitor_central_database_url_missing", message: "Monitor central service requires an explicit DATABASE_URL" }, valuesRedacted: true });
assert.equal(JSON.parse(result.stdout).error.code, "monitor_central_runtime_config_missing");
});