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

fix(monitor): finalize persistence review contracts

Browse Source
This commit is contained in:
AgentRun Codex
2026-07-12 20:22:59 +00:00
parent 3a4e41a248
commit e5bf0b2256
3 changed files with 59 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scripts/monitor-central-service.ts
+2 -2
View File
@@ -1,6 +1,6 @@
// SPEC: PJ2026-01060508 Web哨兵 draft-2026-07-12-p17-monitor-central-persistence.
// Responsibility: Environment-projected bootstrap for the independent central Monitor HTTP service.
import { createMonitorCentralService, createPostgresMonitorCentralStore, MonitorCentralConfigurationError } from "./src/monitor-central-persistence";
import { createMonitorCentralService, createPostgresMonitorCentralStore, structuredMonitorCentralError } from "./src/monitor-central-persistence";
function requiredString(name: string): string {
const value = process.env[name];
@@ -22,7 +22,7 @@ try {
const server = Bun.serve({ hostname: requiredString("MONITOR_CENTRAL_HOST"), port: requiredPositiveInteger("MONITOR_CENTRAL_PORT"), fetch: service.fetch });
console.log(JSON.stringify({ ok: true, service: "monitor-central", port: server.port, valuesRedacted: true }));
} catch (error) {
const structured = error instanceof MonitorCentralConfigurationError ? error.toJSON() : { ok: false, error: { code: (error as { code?: string })?.code ?? "monitor_central_runtime_config_invalid", message: error instanceof Error ? error.message : "invalid configuration" }, valuesRedacted: true };
const structured = { ok: false, error: structuredMonitorCentralError(error), valuesRedacted: true };
console.log(JSON.stringify(structured));
process.exitCode = 2;
}
scripts/src/monitor-central-persistence.test.ts
+12 -3
View File
@@ -2,7 +2,7 @@ import assert from "node:assert/strict";
import { spawnSync } from "node:child_process";
import { test } from "bun:test";
import { createInMemoryMonitorCentralStore, createMonitorCentralService, mapPostgresFindingRow, MONITOR_CENTRAL_MIGRATIONS, MonitorCentralConfigurationError } from "./monitor-central-persistence";
import { createInMemoryMonitorCentralStore, createMonitorCentralService, mapPostgresFindingRow, MONITOR_CENTRAL_MIGRATIONS, MonitorCentralConfigurationError, structuredMonitorCentralError } from "./monitor-central-persistence";
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 }; }
@@ -23,8 +23,9 @@ test("HTTP terminal ingest is unbound-safe, ignores caller hash, and maps confli
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(first.status, 201);
assert.equal((await first.json() as any).disposition, "inserted");
assert.equal(repeated.status, 200);
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");
@@ -37,7 +38,10 @@ test("full terminal envelope controls hash and concurrent same identity converge
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" });
const availability = await target.ingest({ ...base, artifactLocators: [{ ...locator("run-envelope"), reachable: false }] });
assert.equal(availability.disposition, "idempotent");
const detail = await (await target.fetch(new Request("http://local/api/runs/run-envelope?sentinelId=sentinel-a&node=NC01&lane=v03"))).json() as any;
assert.equal(detail.run.artifactLocators[0].reachable, false);
});
test("global and scoped latest use updatedAt then runId descending", async () => {
@@ -95,3 +99,8 @@ test("configuration is explicit and missing environment projection fails structu
assert.equal(result.status, 2);
assert.equal(JSON.parse(result.stdout).error.code, "monitor_central_runtime_config_missing");
});
test("structured error projection redacts database credentials", () => {
const error = new Error("connect postgres://user:secret@example.test/monitor failed");
assert.deepEqual(structuredMonitorCentralError(error), { code: "invalid_request", message: "connect postgres://<redacted> failed" });
});
scripts/src/monitor-central-persistence.ts
+45 -6
View File
@@ -49,7 +49,9 @@ export interface MonitorCentralStore {
migrate(): Promise<void>;
ping(): Promise<void>;
schemaVersion(): Promise<string | null>;
capability(): Promise<{ readonly ingest: boolean; readonly query: boolean }>;
ingest(input: NormalizedIngest): Promise<{ 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[]>;
run(scope: Required<MonitorScope>, runId: string): Promise<Record<string, unknown> | null>;
@@ -126,8 +128,9 @@ export function createMonitorCentralService(options: MonitorCentralServiceOption
checks.push({ name: "postgres", ok: true, valuesRedacted: true });
const schemaVersion = await store.schemaVersion();
checks.push({ name: "schema", ok: schemaVersion === MONITOR_CENTRAL_SCHEMA_VERSION, version: schemaVersion, valuesRedacted: true });
checks.push({ name: "ingest", ok: true, valuesRedacted: true });
checks.push({ name: "query", ok: true, valuesRedacted: true });
const capability = await store.capability();
checks.push({ name: "ingest", ok: capability.ingest, valuesRedacted: true });
checks.push({ name: "query", ok: capability.query, valuesRedacted: true });
} catch (error) {
checks.push({ name: "postgres", ok: false, code: "postgres_unavailable", message: errorMessage(error), valuesRedacted: true });
}
@@ -136,7 +139,10 @@ export function createMonitorCentralService(options: MonitorCentralServiceOption
const fetch = async (request: Request): Promise<Response> => {
const url = new URL(request.url);
try {
if (request.method === "POST" && url.pathname === "/api/ingest") return json(await ingest(await requestRecord(request)), 202);
if (request.method === "POST" && url.pathname === "/api/ingest") {
const result = await ingest(await requestRecord(request));
return json(result, result.disposition === "inserted" ? 201 : 200);
}
if (request.method === "GET" && url.pathname === "/health") return json(await health());
if (request.method === "GET" && url.pathname === "/api/overview") return json({ ok: true, overview: await store.overview(scopeFrom(url)), valuesRedacted: true });
if (request.method === "GET" && url.pathname === "/api/runs") return json(runPage(await store.runs(scopeFrom(url), runQueryOptions(url, pageSize))));
@@ -172,6 +178,13 @@ export function createPostgresMonitorCentralStore(databaseUrl: string, options:
const rows = await sql<{ version: string }[]>`select version from monitor.schema_migrations order by applied_at desc limit 1`;
return rows[0]?.version ?? null;
},
async capability() {
const rows = await sql<{ can_ingest: boolean; can_query: boolean }[]>`
select
has_table_privilege(current_user, 'monitor.runs', 'insert') as can_ingest,
has_table_privilege(current_user, 'monitor.runs', 'select') as can_query`;
return { ingest: rows[0]?.can_ingest === true, query: rows[0]?.can_query === true };
},
async ingest(input) {
return sql.begin(async (transaction) => {
await transaction`
@@ -187,7 +200,10 @@ export function createPostgresMonitorCentralStore(databaseUrl: string, options:
const existing = await transaction<{ payload_hash: string }[]>`
select payload_hash from monitor.runs
where sentinel_id = ${input.sentinelId} and node = ${input.node} and lane = ${input.lane} and run_id = ${input.runId}`;
if (existing[0]?.payload_hash === input.payloadHash) return { disposition: "idempotent" as const };
if (existing[0]?.payload_hash === input.payloadHash) {
await updatePostgresArtifactAvailability(transaction, input);
return { disposition: "idempotent" as const };
}
throw conflict(input);
}
await transaction`
@@ -205,6 +221,12 @@ export function createPostgresMonitorCentralStore(databaseUrl: string, options:
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 };
@@ -263,16 +285,25 @@ export function createInMemoryMonitorCentralStore(): MonitorCentralStore {
async migrate() {},
async ping() {},
async schemaVersion() { return MONITOR_CENTRAL_SCHEMA_VERSION; },
async capability() { return { ingest: true, query: true }; },
async ingest(input) {
const key = identityKey(input);
const existing = records.get(key);
if (existing) {
if (existing.payloadHash !== input.payloadHash) throw conflict(input);
records.set(key, { ...existing, artifactLocators: input.artifactLocators });
return { disposition: "idempotent" };
}
records.set(key, structuredClone(input));
return { disposition: "inserted" };
},
async updateArtifactAvailability(scope, runId, locatorIndex, reachable) {
const key = identityKey({ ...scope, runId });
const existing = records.get(key);
if (!existing?.artifactLocators[locatorIndex]) return;
const artifactLocators = existing.artifactLocators.map((locator, index) => index === locatorIndex ? { ...locator, reachable } : locator);
records.set(key, { ...existing, artifactLocators });
},
async overview(scope) {
const runs = await this.runs(scope, { limit: 1, cursor: null });
return { scope: publicScope(scope), runCount: [...records.values()].filter((item) => matches(scope, item)).length, latest: runs[0] ?? null };
@@ -303,8 +334,9 @@ function requirePositiveInteger(value: number | undefined, code: string): number
function normalizeIngest(input: MonitorCentralIngest): NormalizedIngest {
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, timeline: normalized.timeline })).digest("hex")}` };
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")}` };
}
function immutableLocator(locator: MonitorArtifactLocator): Omit<MonitorArtifactLocator, "reachable"> { const { reachable: _reachable, ...immutable } = locator; return immutable; }
function normalizeLocator(value: MonitorArtifactLocator): MonitorArtifactLocator {
for (const key of ["ownerNode", "ownerLane", "ownerPvc", "stateDir", "relativePath", "sha256", "kind"] as const) if (!value[key]) throw invalid(`artifact locator must include ${key}`);
if (!Number.isSafeInteger(value.sizeBytes) || value.sizeBytes < 0) throw invalid("artifact locator sizeBytes must be a non-negative integer");
@@ -326,7 +358,8 @@ function findingPage(items: readonly Record<string, unknown>[]): Record<string,
function cursor(value: unknown): string { return Buffer.from(JSON.stringify(value)).toString("base64url"); }
function json(body: Record<string, unknown>, status = 200): Response { return Response.json(body, { status }); }
function httpStatus(error: unknown): number { return (error as { code?: string })?.code === "payload_hash_conflict" ? 409 : error instanceof MonitorCentralConfigurationError ? 503 : 400; }
function structuredError(error: unknown): Record<string, unknown> { return error instanceof MonitorCentralConfigurationError ? error.toJSON().error as Record<string, unknown> : { code: (error as { code?: string })?.code ?? "invalid_request", message: errorMessage(error) }; }
export function structuredMonitorCentralError(error: unknown): Record<string, unknown> { return error instanceof MonitorCentralConfigurationError ? error.toJSON().error as Record<string, unknown> : { code: (error as { code?: string })?.code ?? "invalid_request", message: errorMessage(error) }; }
function structuredError(error: unknown): Record<string, unknown> { return structuredMonitorCentralError(error); }
function invalid(message: string): Error & { code: string } { return Object.assign(new Error(message), { code: "invalid_request" }); }
function errorMessage(error: unknown): string { return error instanceof Error ? error.message.replace(/postgres(?:ql)?:\/\/[^\s]+/giu, "postgres://<redacted>") : "unknown error"; }
function identityOf(value: MonitorScope & { readonly runId: string }): Record<string, unknown> { return { sentinelId: value.sentinelId, node: value.node, lane: value.lane, runId: value.runId }; }
@@ -344,6 +377,12 @@ function scopeWhere(scope: MonitorScope): { readonly text: string; readonly valu
function identityWhere(scope: Required<MonitorScope>, 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<MonitorRun[]> { 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<number> { 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<void> {
for (const [index, locator] of input.artifactLocators.entries()) await transaction`
update monitor.artifact_locators
set reachable = ${locator.reachable}
where sentinel_id = ${input.sentinelId} and node = ${input.node} and lane = ${input.lane} and run_id = ${input.runId} and locator_index = ${index}`;
}
function rowIdentity(row: Record<string, unknown>): { readonly text: string; readonly values: readonly string[] } { return { text: "sentinel_id = $1 and node = $2 and lane = $3 and run_id = $4", values: [String(row.sentinel_id), String(row.node), String(row.lane), String(row.run_id)] }; }
function runRow(row: Record<string, unknown>): MonitorRun { return { sentinelId: String(row.sentinel_id), node: String(row.node), lane: String(row.lane), runId: String(row.run_id), updatedAt: new Date(String(row.updated_at)).toISOString(), status: String(row.status), payloadHash: String(row.payload_hash) }; }
export function mapPostgresFindingRow(row: Record<string, unknown>): Record<string, unknown> { return { ...runRow(row), findingIndex: Number(row.finding_index), finding: row.finding }; }