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

fix: address monitor workbench review blockers

Browse Source
This commit is contained in:
AgentRun Codex
2026-07-13 03:49:01 +00:00
parent 6de541221e
commit 1b3eced7f5
8 changed files with 151 additions and 135 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scripts/src/monitor-central-persistence.test.ts
+14
View File
@@ -121,6 +121,20 @@ test("monitor workspace composes runs while runtime facts remain unavailable wit
assert.equal(body.capability, undefined);
});
test("monitor workspace filters before paging and counts findings beyond page size", async () => {
const target = createMonitorCentralService({ store: createInMemoryMonitorCentralStore(), pageSize: 2 });
await target.ingest({ ...input("new-1", "2026-07-12T13:00:00.000Z"), status: "succeeded", findings: [] });
await target.ingest({ ...input("new-2", "2026-07-12T12:00:00.000Z"), status: "succeeded", findings: [] });
await target.ingest({ ...input("older-match", "2026-07-12T11:00:00.000Z"), status: "failed", findings: [{ id: "f1" }, { id: "f2" }, { id: "f3" }] });
const body = await (await target.fetch(new Request("http://local/api/monitor/workspace?sentinelId=sentinel-a&node=NC01&lane=v03&status=failed"))).json() as any;
assert.equal(body.runs.items[0].runId, "older-match");
assert.equal(body.runs.items[0].findingCount, 3);
assert.equal(body.summary.totalRuns, 1);
assert.equal(body.summary.findingCount, 3);
assert.deepEqual(body.summary.statusCounts, { failed: 1 });
assert.equal(body.runs.nextCursor, null);
});
test("monitor workspace projects cadence freshness and health only from runtime metadata", async () => {
const store = createInMemoryMonitorCentralStore();
const normalized = normalizeMonitorCentralIngest(input("run-metadata"));
scripts/src/monitor-central-persistence.ts
+50 -10
View File
@@ -56,6 +56,7 @@ export interface MonitorCentralStore {
runtimeMetadata(scope: Required<MonitorScope>): Promise<readonly MonitorRuntimeMetadata[]>;
updateArtifactAvailability(scope: Required<MonitorScope>, runId: string, locatorIndex: number, reachable: boolean): Promise<void>;
overview(scope: MonitorScope): Promise<Record<string, unknown>>;
monitorWorkspace(scope: MonitorScope, options: MonitorWorkspaceQueryOptions): Promise<MonitorWorkspaceQueryResult>;
runs(scope: MonitorScope, options: RunQueryOptions): Promise<readonly MonitorRun[]>;
run(scope: Required<MonitorScope>, runId: string): Promise<Record<string, unknown> | null>;
view(scope: Required<MonitorScope>, runId: string, view: string): Promise<Record<string, unknown> | null>;
@@ -97,6 +98,15 @@ export interface FindingQueryOptions {
readonly cursor: FindingCursor | null;
}
interface MonitorWorkspaceQueryOptions extends RunQueryOptions {
readonly status?: string;
readonly from?: string;
readonly to?: string;
}
interface MonitorWorkspaceRun extends MonitorRun { readonly findingCount: number; }
interface MonitorWorkspaceQueryResult { readonly items: readonly MonitorWorkspaceRun[]; readonly nextCursor: string | null; readonly totalRuns: number; readonly findingCount: number; readonly statusCounts: Readonly<Record<string, number>>; readonly latestSuccessAt: string | null; readonly latestFailureAt: string | null; }
export interface MonitorCentralService {
ingest(input: MonitorCentralIngest): Promise<Record<string, unknown>>;
health(): Promise<Record<string, unknown>>;
@@ -189,16 +199,10 @@ export function createMonitorCentralService(options: MonitorCentralServiceOption
async function monitorWorkspace(store: MonitorCentralStore, url: URL, pageSize: number): Promise<Record<string, unknown>> {
const scope = scopeFrom(url);
const options = runQueryOptions(url, pageSize);
const status = optional(url, "status");
const from = optionalDate(url, "from");
const to = optionalDate(url, "to");
const source = await store.runs(scope, { ...options, limit: pageSize });
const items = source.filter((run) => (!status || run.status === status) && (!from || run.updatedAt >= from) && (!to || run.updatedAt <= to));
const findings = await store.findings(scope, { limit: pageSize, cursor: null });
const findingCounts = new Map<string, number>();
for (const item of findings) findingCounts.set(String(item.runId), (findingCounts.get(String(item.runId)) ?? 0) + 1);
const runs = items.map((run) => ({ ...run, findingCount: findingCounts.get(run.runId) ?? 0 }));
const result = await store.monitorWorkspace(scope, { ...runQueryOptions(url, pageSize), status, from, to });
const runtime = await monitorRuntimeProjection(store, scope);
return {
ok: true,
@@ -206,9 +210,9 @@ async function monitorWorkspace(store: MonitorCentralStore, url: URL, pageSize:
health: runtime.health,
freshness: runtime.freshness,
cadence: runtime.cadence,
summary: { totalRuns: runs.length, statusCounts: countBy(runs, (run) => run.status), findingCount: runs.reduce((total, run) => total + run.findingCount, 0) },
runs: { items: runs, nextCursor: source.length === pageSize && source.at(-1) ? cursor(source.at(-1)) : null },
selection: { defaultRunId: runs[0]?.runId ?? null },
summary: { totalRuns: result.totalRuns, statusCounts: result.statusCounts, findingCount: result.findingCount, latestSuccessAt: result.latestSuccessAt, latestFailureAt: result.latestFailureAt },
runs: { items: result.items, nextCursor: result.nextCursor },
selection: { defaultRunId: result.items[0]?.runId ?? null },
valuesRedacted: true,
};
}
@@ -313,6 +317,7 @@ export function createPostgresMonitorCentralStore(databaseUrl: string, options:
const rows = await queryRuns(sql, scope, { limit: 1, cursor: null });
return { scope: publicScope(scope), runCount: await countRuns(sql, scope), latest: rows[0] ?? null };
},
async monitorWorkspace(scope, options) { return queryMonitorWorkspace(sql, scope, options); },
async runs(scope, options) { return queryRuns(sql, scope, options); },
async run(scope, runId) {
const where = identityWhere(scope, runId);
@@ -447,6 +452,16 @@ export function createInMemoryMonitorCentralStore(): MonitorCentralStore {
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 };
},
async monitorWorkspace(scope, options) {
const filteredRecords = [...records.values()].filter((item) => matches(scope, item) && monitorWorkspaceMatches(runFrom(item), options));
const all = sortRuns(filteredRecords.map(runFrom));
const findingCounts = new Map(filteredRecords.map((item) => [identityKey(item), item.findings.length]));
const candidates = all.filter((run) => afterRunCursor(run, options.cursor)).slice(0, options.limit + 1);
const hasMore = candidates.length > options.limit;
const items = candidates.slice(0, options.limit).map((run) => ({ ...run, findingCount: findingCounts.get(identityKey(run)) ?? 0 }));
const last = items.at(-1);
return { items, nextCursor: hasMore && last ? cursor(last) : null, totalRuns: all.length, findingCount: filteredRecords.reduce((total, item) => total + item.findings.length, 0), statusCounts: countBy(all, (run) => run.status), latestSuccessAt: all.find((run) => run.status === "succeeded")?.updatedAt ?? null, latestFailureAt: all.find((run) => run.status !== "succeeded")?.updatedAt ?? null };
},
async runs(scope, options) { return sortRuns([...records.values()].filter((item) => matches(scope, item)).map(runFrom)).filter((run) => afterRunCursor(run, options.cursor)).slice(0, options.limit); },
async run(scope, runId) {
const value = records.get(identityKey({ ...scope, runId }));
@@ -500,6 +515,7 @@ function countBy<T>(items: readonly T[], key: (item: T) => string): Record<strin
function recordOrEmpty(value: unknown): Record<string, unknown> { return value && typeof value === "object" && !Array.isArray(value) ? value as Record<string, unknown> : {}; }
function arrayOrEmpty(value: unknown): readonly unknown[] { return Array.isArray(value) ? value : []; }
function objectId(value: unknown, fallback: string): string { const record = recordOrEmpty(value); return String(record.id ?? record.errorId ?? fallback); }
function dateOrNull(value: unknown): string | null { return value ? new Date(String(value)).toISOString() : null; }
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; }
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) }; }
@@ -514,12 +530,36 @@ function matches(scope: MonitorScope, value: MonitorScope): boolean { return (!s
function runFrom(value: NormalizedMonitorCentralIngest): MonitorRun { return { sentinelId: value.sentinelId, node: value.node, lane: value.lane, runId: value.runId, updatedAt: value.updatedAt, status: value.status, payloadHash: value.payloadHash }; }
function sortRuns(runs: readonly MonitorRun[]): MonitorRun[] { return [...runs].sort((left, right) => right.updatedAt.localeCompare(left.updatedAt) || right.runId.localeCompare(left.runId)); }
function afterRunCursor(run: MonitorRun, cursorValue: RunCursor | null): boolean { return !cursorValue || run.updatedAt < cursorValue.updatedAt || (run.updatedAt === cursorValue.updatedAt && run.runId < cursorValue.runId); }
function monitorWorkspaceMatches(run: MonitorRun, options: MonitorWorkspaceQueryOptions): boolean { return (!options.status || run.status === options.status) && (!options.from || run.updatedAt >= options.from) && (!options.to || run.updatedAt <= options.to); }
function compareFindings(left: Record<string, unknown>, right: Record<string, unknown>): number { return String(right.updatedAt).localeCompare(String(left.updatedAt)) || String(right.runId).localeCompare(String(left.runId)) || Number(right.findingIndex) - Number(left.findingIndex); }
function afterFindingCursor(finding: Record<string, unknown>, cursorValue: FindingCursor | null): boolean { if (!cursorValue) return true; return String(finding.updatedAt) < cursorValue.updatedAt || (finding.updatedAt === cursorValue.updatedAt && (String(finding.runId) < cursorValue.runId || (finding.runId === cursorValue.runId && Number(finding.findingIndex) < cursorValue.findingIndex))); }
export function monitorCentralStableJson(value: unknown): string { if (value === undefined) return "null"; if (Array.isArray(value)) return `[${value.map(monitorCentralStableJson).join(",")}]`; if (value && typeof value === "object") return `{${Object.entries(value as Record<string, unknown>).filter(([, item]) => item !== undefined).sort(([left], [right]) => left.localeCompare(right)).map(([key, item]) => `${JSON.stringify(key)}:${monitorCentralStableJson(item)}`).join(",")}}`; const json = JSON.stringify(value); return json === undefined ? "null" : json; }
function scopeWhere(scope: MonitorScope): { readonly text: string; readonly values: readonly string[] } { const entries = Object.entries({ sentinel_id: scope.sentinelId, node: scope.node, lane: scope.lane }).filter(([, value]) => value); return { text: entries.length ? entries.map(([key], index) => `r.${key} = $${index + 1}`).join(" and ") : "true", values: entries.map(([, value]) => String(value)) }; }
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 queryMonitorWorkspace(sql: postgres.Sql, scope: MonitorScope, options: MonitorWorkspaceQueryOptions): Promise<MonitorWorkspaceQueryResult> {
const where = scopeWhere(scope);
const values: unknown[] = [...where.values];
const filters: string[] = [where.text];
if (options.status) { values.push(options.status); filters.push(`r.status = $${values.length}`); }
if (options.from) { values.push(options.from); filters.push(`r.updated_at >= $${values.length}`); }
if (options.to) { values.push(options.to); filters.push(`r.updated_at <= $${values.length}`); }
const baseValues = [...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 + 1);
const predicate = filters.join(" and ");
const [pageRows, summaryRows, statusRows] = await Promise.all([
sql.unsafe(`select r.sentinel_id, r.node, r.lane, r.run_id, r.updated_at, r.status, r.payload_hash, count(f.finding_index)::int as finding_count from monitor.runs r left join monitor.findings f using (sentinel_id, node, lane, run_id) where ${predicate} ${cursorClause} group by r.sentinel_id, r.node, r.lane, r.run_id, r.updated_at, r.status, r.payload_hash order by r.updated_at desc, r.run_id desc limit $${values.length}`, values),
sql.unsafe(`select count(*)::int as total_runs, coalesce(sum(finding_count),0)::int as finding_count, max(updated_at) filter (where status = 'succeeded') as latest_success_at, max(updated_at) filter (where status <> 'succeeded') as latest_failure_at from (select r.status, r.updated_at, count(f.finding_index)::int as finding_count from monitor.runs r left join monitor.findings f using (sentinel_id, node, lane, run_id) where ${predicate} group by r.sentinel_id, r.node, r.lane, r.run_id, r.status, r.updated_at) filtered`, baseValues),
sql.unsafe(`select r.status, count(*)::int as status_count from monitor.runs r where ${predicate} group by r.status`, baseValues),
]);
const hasMore = pageRows.length > options.limit;
const items = pageRows.slice(0, options.limit).map((row) => ({ ...runRow(row), findingCount: Number(row.finding_count ?? 0) }));
const summary = summaryRows[0] ?? {};
const last = items.at(-1);
return { items, nextCursor: hasMore && last ? cursor(last) : null, totalRuns: Number(summary.total_runs ?? 0), findingCount: Number(summary.finding_count ?? 0), statusCounts: Object.fromEntries(statusRows.map((row) => [String(row.status), Number(row.status_count)])), latestSuccessAt: dateOrNull(summary.latest_success_at), latestFailureAt: dateOrNull(summary.latest_failure_at) };
}
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: NormalizedMonitorCentralIngest): Promise<void> {
for (const [index, locator] of input.artifactLocators.entries()) await transaction`
scripts/src/monitor-workbench-route.test.ts
+8
View File
@@ -16,3 +16,11 @@ test("monitor route rejects unsupported paths and filter changes clear selection
const next = withMonitorRoute(readMonitorRoute("https://monitor.example.test/monitor/runs/run-1"), { kind: "workspace", runId: "", filters: { status: "failed" } });
assert.equal(monitorRouteHref(next), "/monitor?status=failed");
});
test("monitor route rejects invalid tab status time range and malformed encoding", () => {
assert.equal(readMonitorRoute("https://monitor.example.test/monitor?tab=raw").reason, "invalid_tab");
assert.equal(readMonitorRoute("https://monitor.example.test/monitor?status=surprise").reason, "invalid_status");
assert.equal(readMonitorRoute("https://monitor.example.test/monitor?from=invalid").reason, "invalid_time_range");
assert.equal(readMonitorRoute("https://monitor.example.test/monitor?from=2026-07-13T05:00:00Z&to=2026-07-13T04:00:00Z").reason, "invalid_time_range");
assert.equal(readMonitorRoute("https://monitor.example.test/monitor/runs/%E0%A4%A").reason, "malformed_url_encoding");
});