Publish curated projectbrain source from 2713a23f6240bdabc1c6c45805a6264b9aaedad4
This commit is contained in:
Public Source Publisher committed
1 parent
98f7618af2
commit
86cfcd1d7a
17 files changed
+1388
-22
No files matched your search
@@ -4,6 +4,7 @@ const optionalEnvValue = <T extends z.ZodTypeAny>(schema: T) => z.preprocess(
|
||||
(value) => typeof value === "string" && value.trim() === "" ? undefined : value,
|
||||
schema.optional(),
|
||||
);
|
||||
const httpsUrl = z.string().url().refine((value) => new URL(value).protocol === "https:", "must use HTTPS");
|
||||
|
||||
const envSchema = z.object({
|
||||
NODE_ENV: z.enum(["development", "test", "production"]).default("development"),
|
||||
@@ -22,6 +23,8 @@ const envSchema = z.object({
|
||||
PROJECTBRAIN_GITEA_URL: z.string().url().optional(),
|
||||
PROJECTBRAIN_GITEA_TOKEN: z.string().min(20).optional(),
|
||||
PROJECTBRAIN_GITEA_WRITE_TOKEN: z.string().min(20).optional(),
|
||||
PROJECTBRAIN_APPOPS_MCP_URL: optionalEnvValue(httpsUrl),
|
||||
PROJECTBRAIN_APPOPS_MCP_TOKEN: optionalEnvValue(z.string().min(24).max(512)),
|
||||
PROJECTBRAIN_GITEA_WEBHOOK_SECRET: optionalEnvValue(z.string().min(24).max(512)),
|
||||
PROJECTBRAIN_MCP_ENABLED: z.enum(["true", "false"]).default("false").transform((value) => value === "true"),
|
||||
PROJECTBRAIN_MCP_RATE_LIMIT_PER_MINUTE: z.coerce.number().int().min(60).max(5_000).default(1_200),
|
||||
@@ -91,6 +94,8 @@ export const loadEnv = (
|
||||
PROJECTBRAIN_GITEA_URL: input.PROJECTBRAIN_GITEA_URL,
|
||||
PROJECTBRAIN_GITEA_TOKEN: input.PROJECTBRAIN_GITEA_TOKEN,
|
||||
PROJECTBRAIN_GITEA_WRITE_TOKEN: input.PROJECTBRAIN_GITEA_WRITE_TOKEN,
|
||||
PROJECTBRAIN_APPOPS_MCP_URL: input.PROJECTBRAIN_APPOPS_MCP_URL,
|
||||
PROJECTBRAIN_APPOPS_MCP_TOKEN: input.PROJECTBRAIN_APPOPS_MCP_TOKEN,
|
||||
PROJECTBRAIN_GITEA_WEBHOOK_SECRET: input.PROJECTBRAIN_GITEA_WEBHOOK_SECRET,
|
||||
PROJECTBRAIN_MCP_ENABLED: input.PROJECTBRAIN_MCP_ENABLED,
|
||||
PROJECTBRAIN_MCP_RATE_LIMIT_PER_MINUTE: input.PROJECTBRAIN_MCP_RATE_LIMIT_PER_MINUTE,
|
||||
@@ -149,6 +154,7 @@ export const loadEnv = (
|
||||
assertProductionSecret("PROJECTBRAIN_AUTH_SECRET", parsed.PROJECTBRAIN_AUTH_SECRET);
|
||||
if (parsed.PROJECTBRAIN_GITEA_WEBHOOK_SECRET) assertProductionSecret("PROJECTBRAIN_GITEA_WEBHOOK_SECRET", parsed.PROJECTBRAIN_GITEA_WEBHOOK_SECRET);
|
||||
if (parsed.PROJECTBRAIN_HUB_SERVICE_TOKEN) assertProductionSecret("PROJECTBRAIN_HUB_SERVICE_TOKEN", parsed.PROJECTBRAIN_HUB_SERVICE_TOKEN);
|
||||
if (parsed.PROJECTBRAIN_APPOPS_MCP_TOKEN) assertProductionSecret("PROJECTBRAIN_APPOPS_MCP_TOKEN", parsed.PROJECTBRAIN_APPOPS_MCP_TOKEN);
|
||||
if (parsed.PROJECTBRAIN_ENTRA_CLIENT_SECRET) assertProductionSecret("PROJECTBRAIN_ENTRA_CLIENT_SECRET", parsed.PROJECTBRAIN_ENTRA_CLIENT_SECRET);
|
||||
}
|
||||
|
||||
|
||||
@@ -6,6 +6,7 @@ import { getRequestId } from "../plugins/requestId.js";
|
||||
import { createGiteaActivityRuntimeLoader, isActiveRunnerJob, prioritizeRuntimeJobs, type CompactRunnerJob } from "../services/activityRuntime.js";
|
||||
import { loadAutonomyRuntime } from "../services/autonomyRuntime.js";
|
||||
import { evaluateRunnerClaimHealth } from "../services/runnerClaimHealth.js";
|
||||
import { createAppOpsRunnerStatisticsLoader, createGiteaRunnerStatisticsFallback } from "../services/runnerStatistics.js";
|
||||
import { errorEnvelope, successEnvelope } from "../utils/envelope.js";
|
||||
import { apiPaths } from "@projectbrain/contracts";
|
||||
|
||||
@@ -98,18 +99,29 @@ const loadDurableActivity = async (projectId: string): Promise<ActivityItem[]> =
|
||||
|
||||
export const registerActivityRoutes = async (app: FastifyInstance, options: { env: AppEnv }) => {
|
||||
const loadGiteaRuntime = createGiteaActivityRuntimeLoader(options.env);
|
||||
const loadAppOpsRunnerStatistics = createAppOpsRunnerStatisticsLoader(options.env);
|
||||
|
||||
app.get("/api/portfolio/runtime", async (request, reply) => {
|
||||
reply.header("Cache-Control", "private, no-store");
|
||||
const [snapshot, worker, autonomy] = await Promise.all([
|
||||
const [snapshot, worker, autonomy, appOpsStatistics] = await Promise.all([
|
||||
loadGiteaRuntime(),
|
||||
prisma.analysisWorkerHeartbeat.findFirst({ orderBy: { lastSeenAt: "desc" } }),
|
||||
loadAutonomyRuntime(options.env.PROJECTBRAIN_AUTONOMY_EXECUTOR_STALE_SECONDS),
|
||||
loadAppOpsRunnerStatistics(),
|
||||
]);
|
||||
const claimHealth = evaluateRunnerClaimHealth(snapshot, { stallSeconds: options.env.PROJECTBRAIN_RUNNER_CLAIM_STALL_SECONDS });
|
||||
const giteaStatisticsFallback = createGiteaRunnerStatisticsFallback(snapshot, {
|
||||
authoritativeProfiles: appOpsStatistics.snapshot.profiles,
|
||||
});
|
||||
return successEnvelope({
|
||||
...snapshot,
|
||||
claimHealth,
|
||||
runnerStatistics: {
|
||||
appOps: appOpsStatistics.snapshot,
|
||||
appOpsSource: appOpsStatistics.source,
|
||||
...(appOpsStatistics.warning ? { appOpsWarning: appOpsStatistics.warning } : {}),
|
||||
giteaFallback: giteaStatisticsFallback,
|
||||
},
|
||||
worker: worker ? { status: Date.now() - worker.lastSeenAt.getTime() < 30_000 ? (worker.activeJobId ? "working" : worker.status) : "stale", activeJobId: worker.activeJobId ?? undefined, detail: worker.detail ?? undefined, lastSeenAt: worker.lastSeenAt.toISOString() } : undefined,
|
||||
autonomy,
|
||||
pollAfterMs: snapshot.counts.running || snapshot.counts.runnable || snapshot.counts.dependencyBlocked || autonomy.counts.activeProjects || autonomy.counts.claimableProjects ? 8_000 : 30_000,
|
||||
|
||||
@@ -27,6 +27,7 @@ export type CompactRunner = {
|
||||
export type CompactRunnerJob = {
|
||||
id: number;
|
||||
runId: number;
|
||||
runAttempt: number;
|
||||
name: string;
|
||||
repository?: string;
|
||||
branch?: string;
|
||||
@@ -172,6 +173,7 @@ const compactJob = (job: GiteaActionJob): CompactRunnerJob => {
|
||||
return {
|
||||
id: job.id,
|
||||
runId: job.run_id,
|
||||
runAttempt: Number.isSafeInteger(job.run_attempt) && job.run_attempt > 0 ? job.run_attempt : 1,
|
||||
name: job.name,
|
||||
repository: repositoryFromActionJob(job),
|
||||
branch: job.head_branch,
|
||||
@@ -333,7 +335,7 @@ export const createGiteaActivityRuntimeLoader = (
|
||||
...(runnableResponse?.jobs ?? []),
|
||||
...(dependencyResponse?.jobs ?? []),
|
||||
...(recentResponse?.jobs ?? []),
|
||||
].map((job) => [job.id, job])).values()];
|
||||
].map((job) => [`${job.run_id}:${job.id}`, job])).values()];
|
||||
const derived = compactGiteaRuntime([], jobs, checkedAt.toISOString()).counts;
|
||||
const snapshot = compactGiteaRuntime(runners, jobs, checkedAt.toISOString(), {
|
||||
running: runningResponse?.total_count ?? derived.running,
|
||||
|
||||
@@ -0,0 +1,552 @@
|
||||
import { Client } from "@modelcontextprotocol/sdk/client/index.js";
|
||||
import { StreamableHTTPClientTransport } from "@modelcontextprotocol/sdk/client/streamableHttp.js";
|
||||
import type { FetchLike } from "@modelcontextprotocol/sdk/shared/transport.js";
|
||||
import { z } from "zod";
|
||||
import type { AppEnv } from "../config/env.js";
|
||||
import type { CompactRunnerJob, GiteaRuntimeSnapshot } from "./activityRuntime.js";
|
||||
|
||||
const runnerProfileIdSchema = z.enum(["checks", "backend", "frontend", "security", "browser", "build"]);
|
||||
export type RunnerProfileId = z.infer<typeof runnerProfileIdSchema>;
|
||||
const nullableCount = z.number().int().finite().min(0).max(1_000_000).nullable();
|
||||
const nullableSeconds = z.number().finite().min(0).max(31_536_000).nullable();
|
||||
const nullableCpu = z.number().finite().min(0).max(1_024).nullable();
|
||||
const nullableBytes = z.number().int().finite().min(0).max(1_125_899_906_842_624).nullable();
|
||||
const nullableTimestamp = z.string().datetime({ offset: true }).nullable();
|
||||
|
||||
export const runnerStatisticsProfileSchema = z.object({
|
||||
id: runnerProfileIdSchema,
|
||||
displayName: z.string().min(1).max(80),
|
||||
labels: z.array(z.string().min(1).max(120)).max(32),
|
||||
runnerName: z.string().min(1).max(120).nullable(),
|
||||
status: z.string().min(1).max(40),
|
||||
queueDepth: nullableCount,
|
||||
activeJobs: nullableCount,
|
||||
oldestQueuedSeconds: nullableSeconds,
|
||||
reason: z.string().max(240).nullable(),
|
||||
cpuLimit: nullableCpu,
|
||||
memoryLimitBytes: nullableBytes,
|
||||
cpuUsed: nullableCpu,
|
||||
memoryUsedBytes: nullableBytes,
|
||||
completedJobs: z.number().int().finite().min(0).max(1_000_000),
|
||||
p50WaitSeconds: nullableSeconds,
|
||||
p95WaitSeconds: nullableSeconds,
|
||||
p50DurationSeconds: nullableSeconds,
|
||||
p95DurationSeconds: nullableSeconds,
|
||||
lastDurationSeconds: nullableSeconds.optional(),
|
||||
}).superRefine((profile, context) => {
|
||||
if (profile.p50WaitSeconds !== null && profile.p95WaitSeconds !== null && profile.p95WaitSeconds < profile.p50WaitSeconds) {
|
||||
context.addIssue({ code: "custom", path: ["p95WaitSeconds"], message: "p95 wait must be at least p50 wait" });
|
||||
}
|
||||
if (profile.p50DurationSeconds !== null && profile.p95DurationSeconds !== null && profile.p95DurationSeconds < profile.p50DurationSeconds) {
|
||||
context.addIssue({ code: "custom", path: ["p95DurationSeconds"], message: "p95 duration must be at least p50 duration" });
|
||||
}
|
||||
if (profile.completedJobs === 0 && [profile.p50WaitSeconds, profile.p95WaitSeconds, profile.p50DurationSeconds, profile.p95DurationSeconds, profile.lastDurationSeconds].some((value) => value !== null && value !== undefined)) {
|
||||
context.addIssue({ code: "custom", message: "job percentiles and last duration require at least one completed job" });
|
||||
}
|
||||
});
|
||||
export type RunnerStatisticsProfile = z.infer<typeof runnerStatisticsProfileSchema>;
|
||||
|
||||
export const runnerStatisticsSchema = z.object({
|
||||
schemaVersion: z.literal(1),
|
||||
status: z.enum(["healthy", "degraded", "unavailable", "not_configured"]),
|
||||
generatedAt: z.string().datetime({ offset: true }),
|
||||
scope: z.literal("ci-fleet"),
|
||||
budget: z.object({
|
||||
cpuLimit: z.union([z.literal(8), z.null()]),
|
||||
memoryLimitBytes: z.union([z.literal(17_179_869_184), z.null()]),
|
||||
activeJobLimit: z.union([z.literal(2), z.null()]),
|
||||
heavyJobLimit: z.union([z.literal(1), z.null()]),
|
||||
reservedCpu: nullableCpu,
|
||||
reservedMemoryBytes: nullableBytes,
|
||||
activeJobs: nullableCount,
|
||||
heavyJobs: nullableCount,
|
||||
}),
|
||||
host: z.object({
|
||||
cpuBusyPercent: z.number().finite().min(0).max(100).nullable(),
|
||||
loadRatio: z.number().finite().min(0).max(1_000).nullable(),
|
||||
availableMemoryBytes: nullableBytes,
|
||||
cacheFreeBytes: nullableBytes,
|
||||
}),
|
||||
profiles: z.array(runnerStatisticsProfileSchema).max(6).superRefine((profiles, context) => {
|
||||
const ids = profiles.map((profile) => profile.id);
|
||||
if (new Set(ids).size !== ids.length) context.addIssue({ code: "custom", message: "profile ids must be unique" });
|
||||
}),
|
||||
window: z.object({
|
||||
startAt: nullableTimestamp,
|
||||
endAt: nullableTimestamp,
|
||||
completedJobs: z.number().int().finite().min(0).max(1_000_000),
|
||||
samples: z.number().int().finite().min(0).max(1_000_000),
|
||||
coverage: z.enum(["bounded", "complete", "unavailable"]),
|
||||
}).superRefine((window, context) => {
|
||||
if (window.startAt && window.endAt && Date.parse(window.startAt) > Date.parse(window.endAt)) {
|
||||
context.addIssue({ code: "custom", path: ["endAt"], message: "window end must not precede its start" });
|
||||
}
|
||||
}),
|
||||
history: z.array(z.object({
|
||||
at: z.string().datetime({ offset: true }),
|
||||
activeJobs: nullableCount,
|
||||
queuedJobs: nullableCount,
|
||||
cpuUsed: nullableCpu,
|
||||
memoryUsedBytes: nullableBytes,
|
||||
})).max(288).superRefine((history, context) => {
|
||||
for (let index = 1; index < history.length; index += 1) {
|
||||
if (Date.parse(history[index]!.at) <= Date.parse(history[index - 1]!.at)) {
|
||||
context.addIssue({ code: "custom", path: [index, "at"], message: "history timestamps must increase" });
|
||||
break;
|
||||
}
|
||||
}
|
||||
}),
|
||||
});
|
||||
export type RunnerStatisticsSnapshot = z.infer<typeof runnerStatisticsSchema>;
|
||||
|
||||
|
||||
export const APPOPS_RUNNER_STATISTICS_TOOL = "gitea.actions.runners.statistics";
|
||||
const REFRESH_INTERVAL_MS = 30_000;
|
||||
const MAX_RESPONSE_AGE_MS = 120_000;
|
||||
const MAX_FUTURE_SKEW_MS = 30_000;
|
||||
const MAX_HISTORY_AGE_MS = 24 * 60 * 60 * 1_000;
|
||||
const MCP_TIMEOUT_MS = 5_000;
|
||||
const MAX_MCP_RESPONSE_BYTES = 1_048_576;
|
||||
const DEFAULT_PROFILES: ReadonlyArray<{ id: RunnerProfileId; displayName: string; labels: string[] }> = [
|
||||
{ id: "checks", displayName: "ITWorx CI · Checks", labels: ["itworx-ci-checks"] },
|
||||
{ id: "backend", displayName: "ITWorx CI · Backend", labels: ["itworx-ci-backend"] },
|
||||
{ id: "frontend", displayName: "ITWorx CI · Frontend", labels: ["itworx-ci-frontend"] },
|
||||
{ id: "security", displayName: "ITWorx CI · Security", labels: ["itworx-ci-security"] },
|
||||
{ id: "browser", displayName: "ITWorx CI · Browser", labels: ["itworx-ci-browser"] },
|
||||
{ id: "build", displayName: "ITWorx CI · Build", labels: ["itworx-ci-build"] },
|
||||
];
|
||||
|
||||
export type AppOpsRunnerStatisticsResult = {
|
||||
snapshot: RunnerStatisticsSnapshot;
|
||||
source: "appops" | "not_configured";
|
||||
warning?: string;
|
||||
};
|
||||
|
||||
export type GiteaRunnerStatisticsProfile = {
|
||||
id: RunnerProfileId;
|
||||
displayName: string;
|
||||
labels: string[];
|
||||
queueDepth: number | null;
|
||||
queueObserved: number;
|
||||
activeJobsObserved: number;
|
||||
oldestQueuedSeconds: number | null;
|
||||
completedJobs: number;
|
||||
samples: number;
|
||||
p50WaitSeconds: number | null;
|
||||
p95WaitSeconds: number | null;
|
||||
p50DurationSeconds: number | null;
|
||||
p95DurationSeconds: number | null;
|
||||
lastDurationSeconds: number | null;
|
||||
window: { startAt: string | null; endAt: string | null; coverage: "bounded" | "unavailable" };
|
||||
};
|
||||
|
||||
export type GiteaRunnerStatisticsFallback = {
|
||||
source: "gitea";
|
||||
availability: GiteaRuntimeSnapshot["availability"];
|
||||
generatedAt: string;
|
||||
runners: { observed: number; busyObserved: number; disabledObserved: number; inventoryComplete: boolean };
|
||||
jobs: {
|
||||
runningObserved: number;
|
||||
queuedObserved: number;
|
||||
dependencyBlockedObserved: number;
|
||||
queuedTotal: number | null;
|
||||
queuedCoverage: "complete" | "bounded" | "unavailable";
|
||||
unattributedObserved: { running: number; queued: number; dependencyBlocked: number; completed: number; total: number };
|
||||
unattributedCoverage: "bounded" | "unavailable";
|
||||
};
|
||||
profiles: GiteaRunnerStatisticsProfile[];
|
||||
};
|
||||
|
||||
const emptySnapshot = (status: "unavailable" | "not_configured", now: Date): RunnerStatisticsSnapshot => ({
|
||||
schemaVersion: 1,
|
||||
status,
|
||||
generatedAt: now.toISOString(),
|
||||
scope: "ci-fleet",
|
||||
budget: { cpuLimit: null, memoryLimitBytes: null, activeJobLimit: null, heavyJobLimit: null, reservedCpu: null, reservedMemoryBytes: null, activeJobs: null, heavyJobs: null },
|
||||
host: { cpuBusyPercent: null, loadRatio: null, availableMemoryBytes: null, cacheFreeBytes: null },
|
||||
profiles: [],
|
||||
window: { startAt: null, endAt: null, completedJobs: 0, samples: 0, coverage: "unavailable" },
|
||||
history: [],
|
||||
});
|
||||
|
||||
const isRecord = (value: unknown): value is Record<string, unknown> => typeof value === "object" && value !== null && !Array.isArray(value);
|
||||
const sanitizeReason = (value: string, token?: string): string => {
|
||||
let reason = value
|
||||
.replace(/(bearer\s+)[^\s,;]+/gi, "$1[redacted]")
|
||||
.replace(/\b(token|secret|password|authorization|api[_-]?key)\b(\s*[:=]\s*)[^\s,;]+/gi, "$1$2[redacted]");
|
||||
if (token) reason = reason.split(token).join("[redacted]");
|
||||
return reason;
|
||||
};
|
||||
|
||||
const valueFromToolResult = (result: unknown): unknown => {
|
||||
if (!isRecord(result) || result.isError === true) throw new Error("AppOps statistics tool failed.");
|
||||
if (isRecord(result.structuredContent)) return result.structuredContent;
|
||||
if (Array.isArray(result.content)) {
|
||||
const textItems = result.content.filter((item) => isRecord(item) && item.type === "text" && typeof item.text === "string");
|
||||
if (textItems.length === 1) {
|
||||
try { return JSON.parse(textItems[0]!.text as string) as unknown; }
|
||||
catch { throw new Error("AppOps statistics tool returned invalid data."); }
|
||||
}
|
||||
}
|
||||
throw new Error("AppOps statistics tool returned no structured data.");
|
||||
};
|
||||
|
||||
const safeEndpoint = (value: string): URL => {
|
||||
const url = new URL(value);
|
||||
if (url.protocol !== "https:") throw new Error("AppOps MCP endpoint must use HTTPS.");
|
||||
if (url.username || url.password || url.search || url.hash) throw new Error("AppOps MCP endpoint configuration is invalid.");
|
||||
return url;
|
||||
};
|
||||
|
||||
const abortError = (): Error => new Error("AppOps statistics request was interrupted.");
|
||||
|
||||
const raceWithSignal = <T>(operation: Promise<T>, signal: AbortSignal): Promise<T> => {
|
||||
if (signal.aborted) return Promise.reject(abortError());
|
||||
return new Promise<T>((resolve, reject) => {
|
||||
let settled = false;
|
||||
const cleanup = () => signal.removeEventListener("abort", onAbort);
|
||||
const onAbort = () => {
|
||||
if (settled) return;
|
||||
settled = true;
|
||||
cleanup();
|
||||
reject(abortError());
|
||||
};
|
||||
signal.addEventListener("abort", onAbort, { once: true });
|
||||
operation.then(
|
||||
(value) => {
|
||||
if (settled) return;
|
||||
settled = true;
|
||||
cleanup();
|
||||
resolve(value);
|
||||
},
|
||||
(error: unknown) => {
|
||||
if (settled) return;
|
||||
settled = true;
|
||||
cleanup();
|
||||
reject(error);
|
||||
},
|
||||
);
|
||||
});
|
||||
};
|
||||
|
||||
export const createOriginBoundFetch = (
|
||||
endpoint: string,
|
||||
fetchImpl: FetchLike = globalThis.fetch,
|
||||
deadlineSignal?: AbortSignal,
|
||||
): FetchLike => {
|
||||
const target = safeEndpoint(endpoint);
|
||||
return async (input, init) => {
|
||||
let requestUrl: URL;
|
||||
try {
|
||||
requestUrl = new URL(input instanceof Request ? input.url : input.toString());
|
||||
} catch {
|
||||
throw new Error("AppOps MCP request target is invalid.");
|
||||
}
|
||||
if (requestUrl.origin !== target.origin || requestUrl.pathname !== target.pathname) {
|
||||
throw new Error("AppOps MCP request target was outside the configured endpoint.");
|
||||
}
|
||||
const signals = [init?.signal, deadlineSignal].filter((signal): signal is AbortSignal => signal !== undefined);
|
||||
const signal = signals.length > 1 ? AbortSignal.any(signals) : signals[0];
|
||||
const request = Promise.resolve(fetchImpl(input, {
|
||||
...init,
|
||||
...(signal ? { signal } : {}),
|
||||
credentials: "omit",
|
||||
redirect: "error",
|
||||
}));
|
||||
let response: Response;
|
||||
try {
|
||||
response = signal ? await raceWithSignal(request, signal) : await request;
|
||||
} catch (error) {
|
||||
// A custom fetch may ignore AbortSignal and resolve after the caller has
|
||||
// left. Release any response it eventually creates without waiting for it.
|
||||
void request.then((lateResponse) => lateResponse.body?.cancel()).catch(() => undefined);
|
||||
throw error;
|
||||
}
|
||||
if (!response.body) return response;
|
||||
const reader = response.body.getReader();
|
||||
let receivedBytes = 0;
|
||||
const cancelReader = () => { void reader.cancel().catch(() => undefined); };
|
||||
if (signal?.aborted) cancelReader();
|
||||
else signal?.addEventListener("abort", cancelReader, { once: true });
|
||||
const removeAbortListener = () => signal?.removeEventListener("abort", cancelReader);
|
||||
const boundedBody = new ReadableStream<Uint8Array>({
|
||||
async pull(controller) {
|
||||
try {
|
||||
const next = await reader.read();
|
||||
if (next.done) {
|
||||
removeAbortListener();
|
||||
controller.close();
|
||||
return;
|
||||
}
|
||||
receivedBytes += next.value.byteLength;
|
||||
if (receivedBytes > MAX_MCP_RESPONSE_BYTES) {
|
||||
await reader.cancel();
|
||||
removeAbortListener();
|
||||
controller.error(new Error("AppOps statistics response exceeded the size limit."));
|
||||
return;
|
||||
}
|
||||
controller.enqueue(next.value);
|
||||
} catch {
|
||||
removeAbortListener();
|
||||
controller.error(new Error("AppOps statistics response could not be read."));
|
||||
}
|
||||
},
|
||||
async cancel(reason) {
|
||||
removeAbortListener();
|
||||
await reader.cancel(reason);
|
||||
},
|
||||
});
|
||||
return new Response(boundedBody, { status: response.status, statusText: response.statusText, headers: response.headers });
|
||||
};
|
||||
};
|
||||
|
||||
export const readAppOpsTool = async (
|
||||
endpoint: string,
|
||||
token: string,
|
||||
options: { timeoutMs?: number; fetch?: FetchLike } = {},
|
||||
): Promise<unknown> => {
|
||||
const url = safeEndpoint(endpoint);
|
||||
const requestedTimeout = options.timeoutMs ?? MCP_TIMEOUT_MS;
|
||||
const timeoutMs = Number.isFinite(requestedTimeout)
|
||||
? Math.min(MCP_TIMEOUT_MS, Math.max(1, requestedTimeout))
|
||||
: MCP_TIMEOUT_MS;
|
||||
const deadline = new AbortController();
|
||||
let rejectDeadline: (reason: Error) => void = () => undefined;
|
||||
const deadlineReached = new Promise<never>((_resolve, reject) => { rejectDeadline = reject; });
|
||||
const onDeadline = () => rejectDeadline(new Error("AppOps statistics request timed out."));
|
||||
deadline.signal.addEventListener("abort", onDeadline, { once: true });
|
||||
const timeoutHandle = setTimeout(() => deadline.abort(), timeoutMs);
|
||||
let client: Client | undefined;
|
||||
try {
|
||||
const transport = new StreamableHTTPClientTransport(url, {
|
||||
requestInit: {
|
||||
headers: { Authorization: `Bearer ${token}` },
|
||||
credentials: "omit",
|
||||
redirect: "error",
|
||||
},
|
||||
fetch: createOriginBoundFetch(endpoint, options.fetch ?? globalThis.fetch, deadline.signal),
|
||||
reconnectionOptions: { maxRetries: 0, initialReconnectionDelay: 100, reconnectionDelayGrowFactor: 1, maxReconnectionDelay: 100 },
|
||||
});
|
||||
client = new Client({ name: "projectbrain-runner-statistics", version: "1.0.0" });
|
||||
const operation = (async () => {
|
||||
await client!.connect(transport);
|
||||
return valueFromToolResult(await client!.callTool({ name: APPOPS_RUNNER_STATISTICS_TOOL }));
|
||||
})();
|
||||
try {
|
||||
return await Promise.race([operation, deadlineReached]);
|
||||
} catch (error) {
|
||||
if (deadline.signal.aborted) throw new Error("AppOps statistics request timed out.", { cause: error });
|
||||
throw error;
|
||||
}
|
||||
} finally {
|
||||
clearTimeout(timeoutHandle);
|
||||
deadline.signal.removeEventListener("abort", onDeadline);
|
||||
try { await client?.close(); } catch { /* closing the read-only session has no bearing on the snapshot */ }
|
||||
}
|
||||
};
|
||||
|
||||
const validateSnapshot = (value: unknown, now: Date, token?: string): { snapshot: RunnerStatisticsSnapshot; warning?: string } => {
|
||||
const parsed = runnerStatisticsSchema.safeParse(value);
|
||||
if (!parsed.success) return { snapshot: emptySnapshot("unavailable", now), warning: "AppOps runner statistics returned an invalid snapshot." };
|
||||
const snapshot = parsed.data;
|
||||
const generatedAt = Date.parse(snapshot.generatedAt);
|
||||
const age = now.getTime() - generatedAt;
|
||||
if (!Number.isFinite(age) || age < -MAX_FUTURE_SKEW_MS || age > MAX_RESPONSE_AGE_MS) {
|
||||
return { snapshot: emptySnapshot("unavailable", now), warning: "AppOps runner statistics are outside the accepted freshness window." };
|
||||
}
|
||||
if ((snapshot.window.startAt && Date.parse(snapshot.window.startAt) > generatedAt)
|
||||
|| (snapshot.window.endAt && Date.parse(snapshot.window.endAt) > generatedAt)) {
|
||||
return { snapshot: emptySnapshot("unavailable", now), warning: "AppOps runner statistics contain a future window timestamp." };
|
||||
}
|
||||
if (snapshot.history.some((sample) => Date.parse(sample.at) > generatedAt)) {
|
||||
return { snapshot: emptySnapshot("unavailable", now), warning: "AppOps runner statistics contain a future history timestamp." };
|
||||
}
|
||||
const cutoff = generatedAt - MAX_HISTORY_AGE_MS;
|
||||
const history = snapshot.history.filter((sample) => Date.parse(sample.at) >= cutoff);
|
||||
const expectedIds = new Set(DEFAULT_PROFILES.map((profile) => profile.id));
|
||||
const observedIds = new Set(snapshot.profiles.map((profile) => profile.id));
|
||||
const missingProfiles = [...expectedIds].some((id) => !observedIds.has(id));
|
||||
const partialCoverage = snapshot.window.coverage === "unavailable";
|
||||
const missingBudget = [snapshot.budget.cpuLimit, snapshot.budget.memoryLimitBytes, snapshot.budget.activeJobLimit, snapshot.budget.heavyJobLimit].some((value) => value === null);
|
||||
const degraded = history.length !== snapshot.history.length || missingProfiles || partialCoverage || missingBudget;
|
||||
const sanitizedProfiles = snapshot.profiles.map((profile) => ({
|
||||
...profile,
|
||||
displayName: sanitizeReason(profile.displayName, token),
|
||||
labels: profile.labels.map((label) => sanitizeReason(label, token)),
|
||||
runnerName: profile.runnerName === null ? null : sanitizeReason(profile.runnerName, token),
|
||||
status: sanitizeReason(profile.status, token),
|
||||
reason: profile.reason === null ? null : sanitizeReason(profile.reason, token),
|
||||
}));
|
||||
return {
|
||||
snapshot: {
|
||||
...snapshot,
|
||||
status: snapshot.status === "healthy" && degraded ? "degraded" : snapshot.status,
|
||||
profiles: sanitizedProfiles,
|
||||
history,
|
||||
},
|
||||
...(degraded ? { warning: "AppOps runner statistics have bounded, stale, or partial sample coverage." } : {}),
|
||||
};
|
||||
};
|
||||
|
||||
export const createAppOpsRunnerStatisticsLoader = (
|
||||
env: Pick<AppEnv, "PROJECTBRAIN_APPOPS_MCP_URL" | "PROJECTBRAIN_APPOPS_MCP_TOKEN">,
|
||||
dependencies: { now?: () => Date; read?: () => Promise<unknown> } = {},
|
||||
) => {
|
||||
const now = dependencies.now ?? (() => new Date());
|
||||
const read = dependencies.read ?? (() => {
|
||||
if (!env.PROJECTBRAIN_APPOPS_MCP_URL || !env.PROJECTBRAIN_APPOPS_MCP_TOKEN) throw new Error("AppOps MCP is not configured.");
|
||||
return readAppOpsTool(env.PROJECTBRAIN_APPOPS_MCP_URL, env.PROJECTBRAIN_APPOPS_MCP_TOKEN);
|
||||
});
|
||||
let cachedAt = 0;
|
||||
let cached: AppOpsRunnerStatisticsResult | undefined;
|
||||
let inFlight: Promise<AppOpsRunnerStatisticsResult> | undefined;
|
||||
|
||||
return async (): Promise<AppOpsRunnerStatisticsResult> => {
|
||||
const checkedAt = now();
|
||||
if (!env.PROJECTBRAIN_APPOPS_MCP_URL || !env.PROJECTBRAIN_APPOPS_MCP_TOKEN) {
|
||||
const completelyAbsent = !env.PROJECTBRAIN_APPOPS_MCP_URL && !env.PROJECTBRAIN_APPOPS_MCP_TOKEN;
|
||||
if (completelyAbsent && (!cached || cached.source !== "not_configured")) {
|
||||
cached = { snapshot: emptySnapshot("not_configured", checkedAt), source: "not_configured", warning: "AppOps runner statistics are not configured." };
|
||||
}
|
||||
if (completelyAbsent) return structuredClone(cached!);
|
||||
return { snapshot: emptySnapshot("unavailable", checkedAt), source: "appops", warning: "AppOps runner statistics are only partially configured." };
|
||||
}
|
||||
const age = checkedAt.getTime() - cachedAt;
|
||||
if (cached && age >= 0 && age < REFRESH_INTERVAL_MS) return structuredClone(cached);
|
||||
if (inFlight) return inFlight.then((value) => structuredClone(value));
|
||||
|
||||
inFlight = read()
|
||||
.then((value) => {
|
||||
const payload = isRecord(value) && "schemaVersion" in value ? value : valueFromToolResult(value);
|
||||
const validated = validateSnapshot(payload, checkedAt, env.PROJECTBRAIN_APPOPS_MCP_TOKEN);
|
||||
cached = { snapshot: validated.snapshot, source: "appops", ...(validated.warning ? { warning: validated.warning } : {}) };
|
||||
return cached;
|
||||
})
|
||||
.catch(() => {
|
||||
cached = { snapshot: emptySnapshot("unavailable", checkedAt), source: "appops", warning: "AppOps runner statistics could not be loaded." };
|
||||
return cached;
|
||||
})
|
||||
.finally(() => { cachedAt = checkedAt.getTime(); inFlight = undefined; });
|
||||
return inFlight.then((value) => structuredClone(value));
|
||||
};
|
||||
};
|
||||
|
||||
const percentile = (values: number[], p: number): number | null => {
|
||||
if (!values.length) return null;
|
||||
const sorted = [...values].sort((left, right) => left - right);
|
||||
return sorted[Math.max(0, Math.ceil(p * sorted.length) - 1)] ?? null;
|
||||
};
|
||||
|
||||
const parseSampleTime = (value: string | undefined, nowMs: number): number | null => {
|
||||
if (!value || value.startsWith("1970-01-01")) return null;
|
||||
const parsed = Date.parse(value);
|
||||
return Number.isFinite(parsed) && parsed > 0 && parsed <= nowMs + MAX_FUTURE_SKEW_MS ? parsed : null;
|
||||
};
|
||||
|
||||
const deduplicateAttempts = (jobs: CompactRunnerJob[]): CompactRunnerJob[] => {
|
||||
const attempts = new Map<string, CompactRunnerJob>();
|
||||
for (const job of jobs) {
|
||||
const key = `${job.runId}:${job.id}`;
|
||||
const previous = attempts.get(key);
|
||||
if (!previous || job.runAttempt > previous.runAttempt || (job.runAttempt === previous.runAttempt && Boolean(job.completedAt) && !previous.completedAt)) attempts.set(key, job);
|
||||
}
|
||||
return [...attempts.values()];
|
||||
};
|
||||
|
||||
const profileForJob = (job: CompactRunnerJob, profiles: RunnerStatisticsProfile[]): RunnerProfileId[] => {
|
||||
const labels = job.labels.map((label) => label.toLowerCase());
|
||||
const profileLabels = labels.filter((label) => label.startsWith("itworx-ci-"));
|
||||
if (profileLabels.length > 0) {
|
||||
if (profileLabels.length !== 1) return [];
|
||||
const explicit = DEFAULT_PROFILES.find((profile) => profile.labels.includes(profileLabels[0]!));
|
||||
return explicit ? [explicit.id] : [];
|
||||
}
|
||||
if ((job.queueState !== "running" && job.queueState !== "completed") || !job.runnerName) return [];
|
||||
const fixedRunnerProfiles = DEFAULT_PROFILES.filter((definition) => profiles.some((profile) => profile.id === definition.id && profile.runnerName === job.runnerName));
|
||||
return fixedRunnerProfiles.length === 1 ? [fixedRunnerProfiles[0]!.id] : [];
|
||||
};
|
||||
|
||||
const safeCompletedSample = (job: CompactRunnerJob, nowMs: number) => {
|
||||
if (job.queueState !== "completed") return undefined;
|
||||
const created = parseSampleTime(job.createdAt, nowMs);
|
||||
const started = parseSampleTime(job.startedAt, nowMs);
|
||||
const completed = parseSampleTime(job.completedAt, nowMs);
|
||||
if (created === null || started === null || completed === null || started < created || completed < started) return undefined;
|
||||
return { created, started, completed, wait: (started - created) / 1_000, duration: (completed - started) / 1_000 };
|
||||
};
|
||||
|
||||
export const createGiteaRunnerStatisticsFallback = (
|
||||
runtime: GiteaRuntimeSnapshot,
|
||||
options: { now?: () => Date; authoritativeProfiles?: RunnerStatisticsProfile[] } = {},
|
||||
): GiteaRunnerStatisticsFallback => {
|
||||
const nowMs = (options.now ?? (() => new Date()))().getTime();
|
||||
const snapshotTime = Date.parse(runtime.generatedAt);
|
||||
const observedAtMs = Number.isFinite(snapshotTime) && snapshotTime > 0 && snapshotTime <= nowMs + MAX_FUTURE_SKEW_MS ? snapshotTime : nowMs;
|
||||
const jobs = deduplicateAttempts(runtime.jobs);
|
||||
const mappedProfile = new Map(jobs.map((job) => [job, profileForJob(job, options.authoritativeProfiles ?? [])]));
|
||||
const hasUnattributedRunnableJob = jobs.some((job) => job.queueState === "runnable" && (mappedProfile.get(job)?.length ?? 0) === 0);
|
||||
const profiles = DEFAULT_PROFILES.map((definition): GiteaRunnerStatisticsProfile => {
|
||||
const matchedJobs = jobs.filter((job) => mappedProfile.get(job)?.includes(definition.id));
|
||||
const queued = matchedJobs.filter((job) => job.queueState === "runnable");
|
||||
const running = matchedJobs.filter((job) => job.queueState === "running");
|
||||
const queueAges = queued.flatMap((job) => {
|
||||
const created = parseSampleTime(job.createdAt, observedAtMs);
|
||||
return created === null ? [] : [Math.max(0, (observedAtMs - created) / 1_000)];
|
||||
});
|
||||
const completed = matchedJobs.flatMap((job) => {
|
||||
const sample = safeCompletedSample(job, observedAtMs);
|
||||
return sample ? [{ job, ...sample }] : [];
|
||||
});
|
||||
const latest = [...completed].sort((left, right) => right.completed - left.completed)[0];
|
||||
const completeQueueScan = runtime.availability === "live" && runtime.runnableJobScan.complete;
|
||||
return {
|
||||
...definition,
|
||||
queueDepth: completeQueueScan && !hasUnattributedRunnableJob ? queued.length : null,
|
||||
queueObserved: queued.length,
|
||||
activeJobsObserved: running.length,
|
||||
oldestQueuedSeconds: queueAges.length ? Math.max(...queueAges) : null,
|
||||
completedJobs: completed.length,
|
||||
samples: completed.length,
|
||||
p50WaitSeconds: percentile(completed.map((sample) => sample.wait), 0.5),
|
||||
p95WaitSeconds: percentile(completed.map((sample) => sample.wait), 0.95),
|
||||
p50DurationSeconds: percentile(completed.map((sample) => sample.duration), 0.5),
|
||||
p95DurationSeconds: percentile(completed.map((sample) => sample.duration), 0.95),
|
||||
lastDurationSeconds: latest?.duration ?? null,
|
||||
window: {
|
||||
startAt: completed.length ? new Date(Math.min(...completed.map((sample) => sample.created))).toISOString() : null,
|
||||
endAt: latest ? new Date(latest.completed).toISOString() : null,
|
||||
coverage: runtime.availability === "unavailable" ? "unavailable" : "bounded",
|
||||
},
|
||||
};
|
||||
});
|
||||
const observed = jobs.filter((job) => job.queueState !== "completed");
|
||||
const unattributed = jobs.filter((job) => (mappedProfile.get(job)?.length ?? 0) === 0);
|
||||
const queuedTotal = runtime.runnableJobScan.total;
|
||||
return {
|
||||
source: "gitea",
|
||||
availability: runtime.availability,
|
||||
generatedAt: runtime.generatedAt,
|
||||
runners: {
|
||||
observed: runtime.runners.length,
|
||||
busyObserved: runtime.runners.filter((runner) => runner.busy).length,
|
||||
disabledObserved: runtime.runners.filter((runner) => runner.disabled).length,
|
||||
inventoryComplete: runtime.runnerInventoryComplete,
|
||||
},
|
||||
jobs: {
|
||||
runningObserved: observed.filter((job) => job.queueState === "running").length,
|
||||
queuedObserved: observed.filter((job) => job.queueState === "runnable").length,
|
||||
dependencyBlockedObserved: observed.filter((job) => job.queueState === "dependency-blocked").length,
|
||||
queuedTotal,
|
||||
queuedCoverage: runtime.availability === "unavailable" ? "unavailable" : runtime.runnableJobScan.complete ? "complete" : "bounded",
|
||||
unattributedObserved: {
|
||||
running: unattributed.filter((job) => job.queueState === "running").length,
|
||||
queued: unattributed.filter((job) => job.queueState === "runnable").length,
|
||||
dependencyBlocked: unattributed.filter((job) => job.queueState === "dependency-blocked").length,
|
||||
completed: unattributed.filter((job) => job.queueState === "completed").length,
|
||||
total: unattributed.length,
|
||||
},
|
||||
unattributedCoverage: runtime.availability === "unavailable" ? "unavailable" : "bounded",
|
||||
},
|
||||
profiles,
|
||||
};
|
||||
};
|
||||
@@ -0,0 +1,410 @@
|
||||
import { describe, expect, it, vi } from "vitest";
|
||||
import type { FetchLike } from "@modelcontextprotocol/sdk/shared/transport.js";
|
||||
import { loadEnv } from "../src/config/env.js";
|
||||
import type { CompactRunnerJob, GiteaRuntimeSnapshot } from "../src/services/activityRuntime.js";
|
||||
import { createAppOpsRunnerStatisticsLoader, createGiteaRunnerStatisticsFallback, createOriginBoundFetch, readAppOpsTool, type RunnerStatisticsSnapshot } from "../src/services/runnerStatistics.js";
|
||||
|
||||
const now = new Date("2026-10-01T09:00:00.000Z");
|
||||
const env = (input: NodeJS.ProcessEnv = {}) => loadEnv({
|
||||
NODE_ENV: "test",
|
||||
DATABASE_URL: "postgresql://example.invalid/projectbrain",
|
||||
PROJECTBRAIN_DATA_SOURCE: "database",
|
||||
PROJECTBRAIN_OWNER_USERNAME: "owner",
|
||||
PROJECTBRAIN_OWNER_PASSWORD: "correct-horse-battery-staple",
|
||||
PROJECTBRAIN_AUTH_SECRET: "test-secret-with-at-least-thirty-two-characters",
|
||||
...input,
|
||||
});
|
||||
|
||||
const profile = (id: "checks" | "backend" | "frontend" | "security" | "browser" | "build", input: Partial<RunnerStatisticsSnapshot["profiles"][number]> = {}) => ({
|
||||
id,
|
||||
displayName: `ITWorx CI · ${id[0]!.toUpperCase()}${id.slice(1)}`,
|
||||
labels: [`itworx-ci-${id}`],
|
||||
runnerName: `${id}-1`,
|
||||
status: "ready",
|
||||
queueDepth: 2,
|
||||
activeJobs: 1,
|
||||
oldestQueuedSeconds: 60,
|
||||
reason: null,
|
||||
cpuLimit: 2,
|
||||
memoryLimitBytes: 4_294_967_296,
|
||||
cpuUsed: 1.25,
|
||||
memoryUsedBytes: 1_073_741_824,
|
||||
completedJobs: 4,
|
||||
p50WaitSeconds: 30,
|
||||
p95WaitSeconds: 120,
|
||||
p50DurationSeconds: 90,
|
||||
p95DurationSeconds: 300,
|
||||
lastDurationSeconds: 100,
|
||||
...input,
|
||||
});
|
||||
|
||||
const stats = (generatedAt = now.toISOString()): RunnerStatisticsSnapshot => ({
|
||||
schemaVersion: 1,
|
||||
status: "healthy",
|
||||
generatedAt,
|
||||
scope: "ci-fleet",
|
||||
budget: { cpuLimit: 8, memoryLimitBytes: 17_179_869_184, activeJobLimit: 2, heavyJobLimit: 1, reservedCpu: 4, reservedMemoryBytes: 8_589_934_592, activeJobs: 1, heavyJobs: 0 },
|
||||
host: { cpuBusyPercent: 20, loadRatio: 0.5, availableMemoryBytes: 8_589_934_592, cacheFreeBytes: 17_179_869_184 },
|
||||
profiles: ["checks", "backend", "frontend", "security", "browser", "build"].map((id) => profile(id as RunnerStatisticsSnapshot["profiles"][number]["id"])),
|
||||
window: { startAt: "2026-10-01T08:00:00.000Z", endAt: now.toISOString(), completedJobs: 24, samples: 120, coverage: "complete" },
|
||||
history: [{ at: "2026-10-01T08:59:00.000Z", activeJobs: 1, queuedJobs: 2, cpuUsed: 1.25, memoryUsedBytes: 1_073_741_824 }],
|
||||
});
|
||||
|
||||
const runtimeJob = (input: Partial<CompactRunnerJob> = {}): CompactRunnerJob => ({
|
||||
id: 1,
|
||||
runId: 10,
|
||||
runAttempt: 1,
|
||||
name: "Backend tests",
|
||||
status: "completed",
|
||||
queueState: "completed",
|
||||
conclusion: "success",
|
||||
labels: ["itworx-ci-backend"],
|
||||
completedSteps: 4,
|
||||
totalSteps: 4,
|
||||
url: "https://gitea.example.com/Jens/projectbrain/actions/runs/10/jobs/1",
|
||||
createdAt: "2026-10-01T08:55:00.000Z",
|
||||
startedAt: "2026-10-01T08:56:00.000Z",
|
||||
completedAt: "2026-10-01T08:58:00.000Z",
|
||||
...input,
|
||||
});
|
||||
|
||||
const runtime = (jobs: CompactRunnerJob[]): GiteaRuntimeSnapshot => ({
|
||||
availability: "live",
|
||||
runners: [{ id: 1, name: "backend-1", status: "online", busy: false, disabled: false, labels: ["itworx-ci-backend"] }],
|
||||
jobs,
|
||||
counts: { running: 0, runnable: 0, dependencyBlocked: 0 },
|
||||
runnableJobScan: { complete: true, observed: 0, total: 0 },
|
||||
runnerInventoryComplete: true,
|
||||
generatedAt: now.toISOString(),
|
||||
stateVersion: "snapshot-v1",
|
||||
});
|
||||
|
||||
describe("runner statistics bridge and Gitea fallback", () => {
|
||||
it("accepts only HTTPS AppOps endpoints while keeping the optional integration unset by default", () => {
|
||||
expect(env().PROJECTBRAIN_APPOPS_MCP_URL).toBeUndefined();
|
||||
expect(env({ PROJECTBRAIN_APPOPS_MCP_URL: "" }).PROJECTBRAIN_APPOPS_MCP_URL).toBeUndefined();
|
||||
expect(env({ PROJECTBRAIN_APPOPS_MCP_URL: "https://appops.example.com/mcp" }).PROJECTBRAIN_APPOPS_MCP_URL)
|
||||
.toBe("https://appops.example.com/mcp");
|
||||
expect(() => env({ PROJECTBRAIN_APPOPS_MCP_URL: "http://appops.example.com/mcp" })).toThrow();
|
||||
});
|
||||
|
||||
it("rejects an HTTP MCP endpoint before the statistics transport can send its credential", async () => {
|
||||
const httpsFetch = vi.fn().mockResolvedValue(new Response(null, { status: 500 }));
|
||||
await expect(readAppOpsTool("https://appops.example.com/mcp", "test-readonly-token-for-statistics", { fetch: httpsFetch })).rejects.toThrow();
|
||||
expect(httpsFetch).toHaveBeenCalled();
|
||||
|
||||
const httpFetch = vi.fn().mockResolvedValue(new Response(null, { status: 500 }));
|
||||
await expect(readAppOpsTool("http://appops.example.com/mcp", "test-readonly-token-for-statistics", { fetch: httpFetch }))
|
||||
.rejects.toThrow(/HTTPS/i);
|
||||
expect(httpFetch).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("reports not configured without calling the optional MCP bridge", async () => {
|
||||
const read = vi.fn();
|
||||
const load = createAppOpsRunnerStatisticsLoader(env(), { now: () => now, read });
|
||||
|
||||
const result = await load();
|
||||
|
||||
expect(result.snapshot.status).toBe("not_configured");
|
||||
expect(result.snapshot.host.availableMemoryBytes).toBeNull();
|
||||
expect(read).not.toHaveBeenCalled();
|
||||
|
||||
const partiallyConfigured = createAppOpsRunnerStatisticsLoader(env({ PROJECTBRAIN_APPOPS_MCP_URL: "https://appops.example.com/mcp" }), { now: () => now, read });
|
||||
expect((await partiallyConfigured()).snapshot.status).toBe("unavailable");
|
||||
expect(read).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("keeps complete v1 data, but degrades partial profile and window coverage", async () => {
|
||||
const read = vi.fn().mockResolvedValue(stats());
|
||||
const load = createAppOpsRunnerStatisticsLoader(env({
|
||||
PROJECTBRAIN_APPOPS_MCP_URL: "https://appops.example.com/mcp",
|
||||
PROJECTBRAIN_APPOPS_MCP_TOKEN: "scoped-readonly-machine-token-123",
|
||||
}), { now: () => now, read });
|
||||
const complete = await load();
|
||||
expect(complete.snapshot.status).toBe("healthy");
|
||||
expect(complete.snapshot.profiles[0]?.displayName).toBe("ITWorx CI · Checks");
|
||||
expect(complete.snapshot.profiles[0]?.lastDurationSeconds).toBe(100);
|
||||
|
||||
const partial = createAppOpsRunnerStatisticsLoader(env({
|
||||
PROJECTBRAIN_APPOPS_MCP_URL: "https://appops.example.com/mcp",
|
||||
PROJECTBRAIN_APPOPS_MCP_TOKEN: "scoped-readonly-machine-token-123",
|
||||
}), { now: () => now, read: async () => ({ ...stats(), status: "healthy", profiles: stats().profiles.slice(0, 4), window: { ...stats().window, coverage: "unavailable" } }) });
|
||||
expect((await partial()).snapshot.status).toBe("degraded");
|
||||
});
|
||||
|
||||
it("reads structured MCP content and strips fields outside the public statistics schema", async () => {
|
||||
const load = createAppOpsRunnerStatisticsLoader(env({
|
||||
PROJECTBRAIN_APPOPS_MCP_URL: "https://appops.example.com/mcp",
|
||||
PROJECTBRAIN_APPOPS_MCP_TOKEN: "scoped-readonly-machine-token-123",
|
||||
}), { now: () => now, read: async () => ({ isError: false, structuredContent: { ...stats(), secretEcho: "must-not-survive" } }) });
|
||||
const result = await load();
|
||||
|
||||
expect(result.snapshot.status).toBe("healthy");
|
||||
expect(JSON.stringify(result.snapshot)).not.toContain("must-not-survive");
|
||||
});
|
||||
|
||||
it("rejects stale, future, out-of-range, and internally inconsistent metric samples", async () => {
|
||||
const configuredEnv = env({
|
||||
PROJECTBRAIN_APPOPS_MCP_URL: "https://appops.example.com/mcp",
|
||||
PROJECTBRAIN_APPOPS_MCP_TOKEN: "scoped-readonly-machine-token-123",
|
||||
});
|
||||
const loadFor = (value: unknown) => createAppOpsRunnerStatisticsLoader(configuredEnv, { now: () => now, read: async () => value });
|
||||
expect((await loadFor(stats("2026-10-01T08:57:59.000Z"))()).snapshot.status).toBe("unavailable");
|
||||
expect((await loadFor(stats("2026-10-01T09:02:00.000Z"))()).snapshot.status).toBe("unavailable");
|
||||
expect((await loadFor({ ...stats(), host: { ...stats().host, availableMemoryBytes: 9e20 } })()).snapshot.status).toBe("unavailable");
|
||||
expect((await loadFor({ ...stats(), profiles: [profile("backend", { p50DurationSeconds: 300, p95DurationSeconds: 120 })] })()).snapshot.status).toBe("unavailable");
|
||||
expect((await loadFor({ ...stats(), history: [{ ...stats().history[0]!, at: "2026-10-01T09:00:01.000Z" }] })()).snapshot.status).toBe("unavailable");
|
||||
|
||||
const aged = await loadFor({ ...stats(), history: [{ ...stats().history[0]!, at: "2026-09-29T08:00:00.000Z" }] })();
|
||||
expect(aged.snapshot.status).toBe("degraded");
|
||||
expect(aged.snapshot.history).toEqual([]);
|
||||
});
|
||||
|
||||
it("rejects future measurement-window bounds while accepting unknown bounds", async () => {
|
||||
const configuredEnv = env({
|
||||
PROJECTBRAIN_APPOPS_MCP_URL: "https://appops.example.com/mcp",
|
||||
PROJECTBRAIN_APPOPS_MCP_TOKEN: "scoped-readonly-machine-token-123",
|
||||
});
|
||||
const loadFor = (value: unknown) => createAppOpsRunnerStatisticsLoader(configuredEnv, { now: () => now, read: async () => value });
|
||||
|
||||
const futureStart = await loadFor({
|
||||
...stats(),
|
||||
window: { ...stats().window, startAt: "2026-10-01T09:00:01.000Z", endAt: null },
|
||||
})();
|
||||
expect(futureStart.snapshot.status).toBe("unavailable");
|
||||
|
||||
const futureEnd = await loadFor({
|
||||
...stats(),
|
||||
window: { ...stats().window, endAt: "2026-10-01T09:00:01.000Z" },
|
||||
})();
|
||||
expect(futureEnd.snapshot.status).toBe("unavailable");
|
||||
|
||||
const unknownBounds = await loadFor({
|
||||
...stats(),
|
||||
window: { ...stats().window, startAt: null, endAt: null, coverage: "bounded" },
|
||||
})();
|
||||
expect(unknownBounds.snapshot.status).toBe("healthy");
|
||||
});
|
||||
|
||||
it("coalesces readers, enforces the 30-second minimum interval, and never returns a tool error or secret", async () => {
|
||||
let clock = now;
|
||||
const secret = "scoped-readonly-machine-token-123";
|
||||
const read = vi.fn().mockRejectedValue(new Error(`Authorization: Bearer ${secret}`, { cause: new Error(secret) }));
|
||||
const load = createAppOpsRunnerStatisticsLoader(env({
|
||||
PROJECTBRAIN_APPOPS_MCP_URL: "https://appops.example.com/mcp",
|
||||
PROJECTBRAIN_APPOPS_MCP_TOKEN: secret,
|
||||
}), { now: () => clock, read });
|
||||
|
||||
const [first, simultaneous] = await Promise.all([load(), load()]);
|
||||
expect(first).toEqual(simultaneous);
|
||||
expect(first.snapshot.status).toBe("unavailable");
|
||||
expect(first.warning).toBe("AppOps runner statistics could not be loaded.");
|
||||
expect(JSON.stringify(first)).not.toContain(secret);
|
||||
expect(read).toHaveBeenCalledTimes(1);
|
||||
clock = new Date(now.getTime() + 29_999);
|
||||
await load();
|
||||
expect(read).toHaveBeenCalledTimes(1);
|
||||
clock = new Date(now.getTime() + 30_000);
|
||||
await load();
|
||||
expect(read).toHaveBeenCalledTimes(2);
|
||||
});
|
||||
|
||||
it("prevents authorization from crossing origins or following a redirect", async () => {
|
||||
const fetcher = vi.fn().mockResolvedValue(new Response("ok"));
|
||||
const safeFetch = createOriginBoundFetch("https://appops.example.com/mcp", fetcher);
|
||||
await expect(safeFetch("https://evil.example/mcp", { headers: { Authorization: "Bearer secret" } })).rejects.toThrow();
|
||||
await expect(safeFetch("https://appops.example.com/other", { headers: { Authorization: "Bearer secret" } })).rejects.toThrow();
|
||||
await safeFetch("https://appops.example.com/mcp", { redirect: "follow" });
|
||||
expect(fetcher).toHaveBeenCalledTimes(1);
|
||||
expect(fetcher.mock.calls[0]?.[1]?.redirect).toBe("error");
|
||||
});
|
||||
|
||||
it.each(["slow POST", "stalled JSON body", "stalled SSE body"] as const)(
|
||||
"bounds the MCP operation and cancels a %s before cleanup completes",
|
||||
async (mode) => {
|
||||
let toolCallStarted = false;
|
||||
let observedSignal: AbortSignal | undefined;
|
||||
let bodyCancelled = false;
|
||||
const fetchImpl: FetchLike = async (input, init) => {
|
||||
const method = init?.method ?? (input instanceof Request ? input.method : "GET");
|
||||
if (method === "GET") return new Response(null, { status: 405 });
|
||||
const requestBody = typeof init?.body === "string" ? init.body : "";
|
||||
const message = JSON.parse(requestBody) as { id?: number; method?: string };
|
||||
if (message.method === "initialize") {
|
||||
return new Response(JSON.stringify({
|
||||
jsonrpc: "2.0",
|
||||
id: message.id,
|
||||
result: {
|
||||
protocolVersion: "2025-06-18",
|
||||
capabilities: { tools: {} },
|
||||
serverInfo: { name: "AppOps test", version: "1.0.0" },
|
||||
},
|
||||
}), { headers: { "content-type": "application/json" } });
|
||||
}
|
||||
if (message.method === "notifications/initialized") return new Response(null, { status: 202 });
|
||||
if (message.method !== "tools/call") return new Response(null, { status: 400 });
|
||||
|
||||
toolCallStarted = true;
|
||||
observedSignal = init?.signal ?? undefined;
|
||||
if (mode === "slow POST") {
|
||||
return await new Promise<Response>((_resolve, reject) => {
|
||||
const signal = init?.signal;
|
||||
if (signal?.aborted) reject(new Error("request aborted"));
|
||||
else signal?.addEventListener("abort", () => reject(new Error("request aborted")), { once: true });
|
||||
});
|
||||
}
|
||||
|
||||
const body = new ReadableStream<Uint8Array>({
|
||||
pull: () => new Promise<void>(() => {}),
|
||||
cancel: () => { bodyCancelled = true; },
|
||||
});
|
||||
return new Response(body, {
|
||||
headers: { "content-type": mode === "stalled SSE body" ? "text/event-stream" : "application/json" },
|
||||
});
|
||||
};
|
||||
|
||||
let rejection: unknown;
|
||||
const outcome = await Promise.race([
|
||||
readAppOpsTool("https://appops.example.com/mcp", "scoped-readonly-machine-token-123", { timeoutMs: 30, fetch: fetchImpl })
|
||||
.then(() => "resolved", (error: unknown) => { rejection = error; return "rejected"; }),
|
||||
new Promise<string>((resolve) => setTimeout(() => resolve("still pending"), 250)),
|
||||
]);
|
||||
|
||||
expect(outcome).toBe("rejected");
|
||||
expect(rejection).toBeInstanceOf(Error);
|
||||
expect((rejection as Error).message).toBe("AppOps statistics request timed out.");
|
||||
expect((rejection as Error).cause).toBeInstanceOf(Error);
|
||||
expect(((rejection as Error).cause as Error).message).toBe("AppOps statistics request timed out.");
|
||||
expect(toolCallStarted).toBe(true);
|
||||
expect(observedSignal?.aborted).toBe(true);
|
||||
if (mode !== "slow POST") expect(bodyCancelled).toBe(true);
|
||||
},
|
||||
1_000,
|
||||
);
|
||||
|
||||
it("redacts a configured machine credential echoed in a profile reason", async () => {
|
||||
const secret = "scoped-readonly-machine-token-123";
|
||||
const payload = { ...stats(), profiles: [profile("backend", { runnerName: secret, reason: `Bearer ${secret}` })] };
|
||||
const load = createAppOpsRunnerStatisticsLoader(env({
|
||||
PROJECTBRAIN_APPOPS_MCP_URL: "https://appops.example.com/mcp",
|
||||
PROJECTBRAIN_APPOPS_MCP_TOKEN: secret,
|
||||
}), { now: () => now, read: async () => payload });
|
||||
const result = await load();
|
||||
|
||||
expect(JSON.stringify(result)).not.toContain(secret);
|
||||
expect(result.snapshot.profiles[0]?.reason).toBe("Bearer [redacted]");
|
||||
expect(result.snapshot.profiles[0]?.runnerName).toBe("[redacted]");
|
||||
});
|
||||
|
||||
it("deduplicates retried job attempts and excludes invalid timestamps from bounded percentiles", () => {
|
||||
const fallback = createGiteaRunnerStatisticsFallback(runtime([
|
||||
runtimeJob({ id: 1, runAttempt: 1, createdAt: "2026-10-01T08:50:00.000Z", startedAt: "2026-10-01T08:51:00.000Z", completedAt: "2026-10-01T08:53:00.000Z" }),
|
||||
runtimeJob({ id: 1, runAttempt: 2, createdAt: "2026-10-01T08:55:00.000Z", startedAt: "2026-10-01T08:56:00.000Z", completedAt: "2026-10-01T08:58:00.000Z" }),
|
||||
runtimeJob({ id: 2, runId: 11, createdAt: "2026-10-01T08:40:00.000Z", startedAt: "2026-10-01T08:42:00.000Z", completedAt: "2026-10-01T08:46:00.000Z" }),
|
||||
runtimeJob({ id: 3, runId: 12, createdAt: "2026-10-01T08:40:00.000Z", startedAt: "not-a-date", completedAt: "2026-10-01T08:46:00.000Z" }),
|
||||
]), { now: () => now });
|
||||
|
||||
const backend = fallback.profiles.find((item) => item.id === "backend");
|
||||
expect(backend).toMatchObject({ completedJobs: 2, samples: 2, p50WaitSeconds: 60, p95WaitSeconds: 120, p50DurationSeconds: 120, p95DurationSeconds: 240, lastDurationSeconds: 120, window: { coverage: "bounded" } });
|
||||
expect(backend?.window.startAt).toBe("2026-10-01T08:40:00.000Z");
|
||||
expect(backend?.window.endAt).toBe("2026-10-01T08:58:00.000Z");
|
||||
});
|
||||
|
||||
it("keeps profile queue depth unknown when a complete queue scan has an unattributed queued job", () => {
|
||||
const fallback = createGiteaRunnerStatisticsFallback({
|
||||
...runtime([
|
||||
runtimeJob({ id: 10, status: "queued", queueState: "runnable", labels: ["itworx-ci-backend"], runnerId: undefined, runnerName: undefined }),
|
||||
runtimeJob({ id: 11, runId: 11, status: "queued", queueState: "runnable", labels: ["ubuntu-latest"], runnerId: undefined, runnerName: undefined }),
|
||||
]),
|
||||
counts: { running: 0, runnable: 2, dependencyBlocked: 0 },
|
||||
runnableJobScan: { complete: true, observed: 2, total: 2 },
|
||||
}, { now: () => now });
|
||||
|
||||
expect(fallback.jobs).toMatchObject({
|
||||
queuedObserved: 2,
|
||||
queuedTotal: 2,
|
||||
queuedCoverage: "complete",
|
||||
unattributedObserved: { queued: 1, total: 1 },
|
||||
});
|
||||
expect(fallback.profiles.find((item) => item.id === "backend")).toMatchObject({ queueDepth: null, queueObserved: 1 });
|
||||
expect(fallback.profiles.find((item) => item.id === "frontend")).toMatchObject({ queueDepth: null, queueObserved: 0 });
|
||||
});
|
||||
|
||||
it("keeps exact profile queue depth when all queued jobs are attributed, regardless of other unattributed states", () => {
|
||||
const allQueuedJobsAttributed = createGiteaRunnerStatisticsFallback({
|
||||
...runtime([
|
||||
runtimeJob({ id: 12, status: "queued", queueState: "runnable", labels: ["itworx-ci-backend"], runnerId: undefined, runnerName: undefined }),
|
||||
]),
|
||||
counts: { running: 0, runnable: 1, dependencyBlocked: 0 },
|
||||
runnableJobScan: { complete: true, observed: 1, total: 1 },
|
||||
}, { now: () => now });
|
||||
expect(allQueuedJobsAttributed.profiles.find((item) => item.id === "backend")).toMatchObject({ queueDepth: 1, queueObserved: 1 });
|
||||
expect(allQueuedJobsAttributed.profiles.find((item) => item.id === "frontend")).toMatchObject({ queueDepth: 0, queueObserved: 0 });
|
||||
|
||||
const onlyUnattributedNonqueuedJobs = createGiteaRunnerStatisticsFallback({
|
||||
...runtime([
|
||||
runtimeJob({ id: 13, status: "in_progress", queueState: "running", labels: ["ubuntu-latest"], runnerId: undefined, runnerName: undefined }),
|
||||
runtimeJob({ id: 14, runId: 14, status: "pending", queueState: "dependency-blocked", labels: ["linux-validation"], runnerId: undefined, runnerName: undefined }),
|
||||
runtimeJob({ id: 15, runId: 15, labels: ["ubuntu-latest"], runnerId: undefined, runnerName: undefined }),
|
||||
]),
|
||||
counts: { running: 1, runnable: 0, dependencyBlocked: 1 },
|
||||
runnableJobScan: { complete: true, observed: 0, total: 0 },
|
||||
}, { now: () => now });
|
||||
expect(onlyUnattributedNonqueuedJobs.jobs).toMatchObject({
|
||||
queuedTotal: 0,
|
||||
queuedCoverage: "complete",
|
||||
unattributedObserved: { running: 1, dependencyBlocked: 1, completed: 1, total: 3 },
|
||||
});
|
||||
expect(onlyUnattributedNonqueuedJobs.profiles.find((item) => item.id === "backend")).toMatchObject({ queueDepth: 0, queueObserved: 0 });
|
||||
});
|
||||
|
||||
it("does not guess profiles from shared legacy labels and exposes unattributed jobs", () => {
|
||||
const fallback = createGiteaRunnerStatisticsFallback({
|
||||
...runtime([]),
|
||||
jobs: [
|
||||
runtimeJob({ id: 4, status: "queued", queueState: "runnable", labels: ["ubuntu-latest"], runnerId: undefined, runnerName: undefined }),
|
||||
runtimeJob({ id: 5, runId: 11, status: "queued", queueState: "runnable", labels: ["linux-validation"], runnerId: undefined, runnerName: undefined }),
|
||||
runtimeJob({ id: 6, runId: 12, status: "queued", queueState: "runnable", labels: ["ubuntu-latest", "itworx-ci-backend"], runnerId: undefined, runnerName: undefined }),
|
||||
],
|
||||
runnableJobScan: { complete: false, observed: 3, total: 50 },
|
||||
}, { now: () => now });
|
||||
|
||||
expect(fallback.jobs.queuedCoverage).toBe("bounded");
|
||||
expect(fallback.profiles.find((item) => item.id === "frontend")).toMatchObject({ queueDepth: null, queueObserved: 0 });
|
||||
expect(fallback.profiles.find((item) => item.id === "backend")).toMatchObject({ queueDepth: null, queueObserved: 1 });
|
||||
expect(fallback.profiles.find((item) => item.id === "security")?.queueObserved).toBe(0);
|
||||
expect(fallback.jobs.unattributedObserved).toMatchObject({ queued: 2, total: 2 });
|
||||
expect(fallback.jobs.unattributedCoverage).toBe("bounded");
|
||||
});
|
||||
|
||||
it("maps a fixed runner name only for assigned running/completed jobs and rejects conflicting specialized labels", () => {
|
||||
const fallback = createGiteaRunnerStatisticsFallback(runtime([
|
||||
runtimeJob({ id: 7, status: "in_progress", queueState: "running", runnerName: "backend-fixed", labels: ["linux-validation"] }),
|
||||
runtimeJob({ id: 8, runId: 13, runnerName: "backend-fixed", labels: ["itworx-ci-backend", "itworx-ci-frontend"] }),
|
||||
]), {
|
||||
now: () => now,
|
||||
authoritativeProfiles: [profile("backend", { runnerName: "backend-fixed" })],
|
||||
});
|
||||
expect(fallback.profiles.find((item) => item.id === "backend")?.activeJobsObserved).toBe(1);
|
||||
expect(fallback.profiles.find((item) => item.id === "frontend")?.activeJobsObserved).toBe(0);
|
||||
expect(fallback.jobs.unattributedObserved).toMatchObject({ running: 0, completed: 1, total: 1 });
|
||||
});
|
||||
|
||||
it("limits AppOps response streams to one mebibyte", async () => {
|
||||
const safeFetch = createOriginBoundFetch("https://appops.example.com/mcp", vi.fn().mockResolvedValue(new Response("x".repeat(1_048_577))));
|
||||
const response = await safeFetch("https://appops.example.com/mcp");
|
||||
await expect(response.text()).rejects.toThrow();
|
||||
});
|
||||
|
||||
it("returns independent snapshots from the 30-second cache", async () => {
|
||||
const read = vi.fn().mockResolvedValue(stats());
|
||||
const load = createAppOpsRunnerStatisticsLoader(env({
|
||||
PROJECTBRAIN_APPOPS_MCP_URL: "https://appops.example.com/mcp",
|
||||
PROJECTBRAIN_APPOPS_MCP_TOKEN: "scoped-readonly-machine-token-123",
|
||||
}), { now: () => now, read });
|
||||
const first = await load();
|
||||
first.snapshot.profiles[0]!.displayName = "poisoned";
|
||||
expect((await load()).snapshot.profiles[0]?.displayName).toBe("ITWorx CI · Checks");
|
||||
expect(read).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
});
|
||||
Reference in new issue
Block a user