From e5bf0b225642caf7068fc0b29b54551b32901447 Mon Sep 17 00:00:00 2001 From: AgentRun Codex Date: Sun, 12 Jul 2026 20:22:59 +0000 Subject: [PATCH] fix(monitor): finalize persistence review contracts --- scripts/monitor-central-service.ts | 4 +- .../src/monitor-central-persistence.test.ts | 15 ++++-- scripts/src/monitor-central-persistence.ts | 51 ++++++++++++++++--- 3 files changed, 59 insertions(+), 11 deletions(-) diff --git a/scripts/monitor-central-service.ts b/scripts/monitor-central-service.ts index 790b8e17..fa46f80d 100644 --- a/scripts/monitor-central-service.ts +++ b/scripts/monitor-central-service.ts @@ -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; } diff --git a/scripts/src/monitor-central-persistence.test.ts b/scripts/src/monitor-central-persistence.test.ts index 5578f9e5..30557664 100644 --- a/scripts/src/monitor-central-persistence.test.ts +++ b/scripts/src/monitor-central-persistence.test.ts @@ -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:// failed" }); +}); diff --git a/scripts/src/monitor-central-persistence.ts b/scripts/src/monitor-central-persistence.ts index 79188964..24524307 100644 --- a/scripts/src/monitor-central-persistence.ts +++ b/scripts/src/monitor-central-persistence.ts @@ -49,7 +49,9 @@ export interface MonitorCentralStore { migrate(): Promise; ping(): Promise; schemaVersion(): Promise; + capability(): Promise<{ readonly ingest: boolean; readonly query: boolean }>; ingest(input: NormalizedIngest): Promise<{ readonly disposition: "inserted" | "idempotent" }>; + updateArtifactAvailability(scope: Required, runId: string, locatorIndex: number, reachable: boolean): Promise; overview(scope: MonitorScope): Promise>; runs(scope: MonitorScope, options: RunQueryOptions): Promise; run(scope: Required, runId: string): Promise | 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 => { 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 { 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[]): Record, 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 { return error instanceof MonitorCentralConfigurationError ? error.toJSON().error as Record : { code: (error as { code?: string })?.code ?? "invalid_request", message: errorMessage(error) }; } +export function structuredMonitorCentralError(error: unknown): Record { return error instanceof MonitorCentralConfigurationError ? error.toJSON().error as Record : { code: (error as { code?: string })?.code ?? "invalid_request", message: errorMessage(error) }; } +function structuredError(error: unknown): Record { 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://") : "unknown error"; } function identityOf(value: MonitorScope & { readonly runId: string }): Record { 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, runId: string): { readonly text: string; readonly values: readonly string[] } { return { text: "r.sentinel_id = $1 and r.node = $2 and r.lane = $3 and r.run_id = $4", values: [scope.sentinelId, scope.node, scope.lane, runId] }; } async function queryRuns(sql: postgres.Sql, scope: MonitorScope, options: RunQueryOptions): Promise { const where = scopeWhere(scope); const values: unknown[] = [...where.values]; let cursorClause = ""; if (options.cursor) { values.push(options.cursor.updatedAt, options.cursor.runId); cursorClause = `and (r.updated_at, r.run_id) < ($${values.length - 1}, $${values.length})`; } values.push(options.limit); const rows = await sql.unsafe(`select r.sentinel_id, r.node, r.lane, r.run_id, r.updated_at, r.status, r.payload_hash from monitor.runs r where ${where.text} ${cursorClause} order by r.updated_at desc, r.run_id desc limit $${values.length}`, values); return rows.map(runRow); } async function countRuns(sql: postgres.Sql, scope: MonitorScope): Promise { const where = scopeWhere(scope); const rows = await sql.unsafe(`select count(*)::int as count from monitor.runs r where ${where.text}`, where.values); return Number(rows[0]?.count ?? 0); } +async function updatePostgresArtifactAvailability(transaction: postgres.TransactionSql, input: NormalizedIngest): Promise { + 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): { 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): 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): Record { return { ...runRow(row), findingIndex: Number(row.finding_index), finding: row.finding }; }