Codex
785 lines
33 KiB
TypeScript
785 lines
33 KiB
TypeScript
import { spawn, spawnSync } from "node:child_process";
|
|
import { existsSync, mkdirSync, readFileSync, readdirSync, statSync, writeFileSync } from "node:fs";
|
|
import { join } from "node:path";
|
|
import { repoRoot, rootPath } from "./config";
|
|
import { runCommandToFiles, tailFile } from "./command";
|
|
import { hwlabDefaultRuntimeTarget } from "./hwlab-node-lanes";
|
|
|
|
export type JobStatus = "queued" | "running" | "succeeded" | "failed" | "canceled";
|
|
|
|
export interface JobRecord {
|
|
id: string;
|
|
name: string;
|
|
status: JobStatus;
|
|
command: string[];
|
|
cwd: string;
|
|
runner: "local" | "docker";
|
|
runnerPid?: number | null;
|
|
runnerContainer?: string | null;
|
|
createdAt: string;
|
|
startedAt: string | null;
|
|
finishedAt: string | null;
|
|
exitCode: number | null;
|
|
stdoutFile: string;
|
|
stderrFile: string;
|
|
note: string;
|
|
}
|
|
|
|
export interface JobProgressSummary {
|
|
kind: "hwlab-v02-trigger" | "hwlab-runtime-lane-trigger" | "git-mirror" | "generic";
|
|
stage: string | null;
|
|
stageStatus: string | null;
|
|
sourceCommit: string | null;
|
|
pipelineRun: string | null;
|
|
pipelineCreated: boolean | null;
|
|
elapsedSeconds: number | null;
|
|
stageElapsedSeconds: number | null;
|
|
lastEventAt: string | null;
|
|
lastEventAgeSeconds: number | null;
|
|
eventsObserved: number;
|
|
slow: boolean;
|
|
warnings: string[];
|
|
diagnostics?: Record<string, unknown> | null;
|
|
timings: Record<string, number>;
|
|
summary: string;
|
|
nextCommand: string | null;
|
|
}
|
|
|
|
export interface StartJobOptions {
|
|
runner?: "local" | "docker";
|
|
dockerImage?: string;
|
|
}
|
|
|
|
function jobsDir(): string {
|
|
const dir = rootPath(".state", "jobs");
|
|
mkdirSync(dir, { recursive: true });
|
|
return dir;
|
|
}
|
|
|
|
function jobPath(id: string): string {
|
|
return join(jobsDir(), `${id}.json`);
|
|
}
|
|
|
|
function writeJob(job: JobRecord): void {
|
|
writeFileSync(jobPath(job.id), `${JSON.stringify(job, null, 2)}\n`, "utf8");
|
|
}
|
|
|
|
export function readJob(id: string): JobRecord {
|
|
const path = jobPath(id);
|
|
if (!existsSync(path)) throw new Error(`job not found: ${id}`);
|
|
return JSON.parse(readFileSync(path, "utf8")) as JobRecord;
|
|
}
|
|
|
|
export function cancelJob(id: string): unknown {
|
|
const job = readJob(id);
|
|
if (job.status !== "queued" && job.status !== "running") {
|
|
return { ok: true, action: "cancel-job", id, alreadyTerminal: true, job };
|
|
}
|
|
const actions: Array<Record<string, unknown>> = [];
|
|
if (job.runnerPid !== null && job.runnerPid !== undefined) {
|
|
try {
|
|
process.kill(-job.runnerPid, "SIGTERM");
|
|
actions.push({ signal: "SIGTERM", target: "process-group", pid: job.runnerPid, ok: true });
|
|
} catch (error) {
|
|
actions.push({ signal: "SIGTERM", target: "process-group", pid: job.runnerPid, ok: false, error: error instanceof Error ? error.message : String(error) });
|
|
try {
|
|
process.kill(job.runnerPid, "SIGTERM");
|
|
actions.push({ signal: "SIGTERM", target: "process", pid: job.runnerPid, ok: true });
|
|
} catch (innerError) {
|
|
actions.push({ signal: "SIGTERM", target: "process", pid: job.runnerPid, ok: false, error: innerError instanceof Error ? innerError.message : String(innerError) });
|
|
}
|
|
}
|
|
}
|
|
job.status = "canceled";
|
|
job.finishedAt = new Date().toISOString();
|
|
job.exitCode = 130;
|
|
writeFileSync(job.stderrFile, `${tailFile(job.stderrFile, 512_000)}${JSON.stringify({ event: "unidesk.job.cancel", at: job.finishedAt, jobId: id, actions })}\n`, "utf8");
|
|
writeJob(job);
|
|
return { ok: true, action: "cancel-job", id, actions, job };
|
|
}
|
|
|
|
export function listJobs(): JobRecord[] {
|
|
return readdirSync(jobsDir())
|
|
.filter((name) => name.endsWith(".json"))
|
|
.map((name) => JSON.parse(readFileSync(join(jobsDir(), name), "utf8")) as JobRecord)
|
|
.sort((a, b) => b.createdAt.localeCompare(a.createdAt));
|
|
}
|
|
|
|
export function startJob(name: string, command: string[], note: string, options: StartJobOptions = {}): JobRecord {
|
|
const id = `${name}_${new Date().toISOString().replace(/[-:.TZ]/g, "")}_${Math.random().toString(16).slice(2, 8)}`;
|
|
const stdoutFile = rootPath(".state", "jobs", `${id}.stdout.log`);
|
|
const stderrFile = rootPath(".state", "jobs", `${id}.stderr.log`);
|
|
const runner = options.runner ?? "local";
|
|
const job: JobRecord = {
|
|
id,
|
|
name,
|
|
status: "queued",
|
|
command,
|
|
cwd: repoRoot,
|
|
runner,
|
|
runnerPid: null,
|
|
runnerContainer: null,
|
|
createdAt: new Date().toISOString(),
|
|
startedAt: null,
|
|
finishedAt: null,
|
|
exitCode: null,
|
|
stdoutFile,
|
|
stderrFile,
|
|
note,
|
|
};
|
|
writeJob(job);
|
|
if (runner === "docker") {
|
|
const containerName = `unidesk-job-runner-${id}`.replace(/[^A-Za-z0-9_.-]/g, "-").slice(0, 120);
|
|
job.runnerContainer = containerName;
|
|
writeJob(job);
|
|
const dockerArgs = [
|
|
"run",
|
|
"-d",
|
|
"--rm",
|
|
"--name",
|
|
containerName,
|
|
"-v",
|
|
"/var/run/docker.sock:/var/run/docker.sock",
|
|
"-v",
|
|
`${repoRoot}:${repoRoot}`,
|
|
"-w",
|
|
repoRoot,
|
|
"--entrypoint",
|
|
"bun",
|
|
options.dockerImage ?? "unidesk-code-queue:latest",
|
|
rootPath("scripts", "cli.ts"),
|
|
"internal",
|
|
"run-job",
|
|
id,
|
|
];
|
|
const result = spawnSync("docker", dockerArgs, { cwd: repoRoot, encoding: "utf8" });
|
|
if (result.status !== 0) {
|
|
job.status = "failed";
|
|
job.startedAt = new Date().toISOString();
|
|
job.finishedAt = new Date().toISOString();
|
|
job.exitCode = result.status ?? 127;
|
|
writeFileSync(stderrFile, result.stderr || result.error?.message || "failed to start docker job runner\n", "utf8");
|
|
writeFileSync(stdoutFile, result.stdout || "", "utf8");
|
|
writeJob(job);
|
|
}
|
|
return job;
|
|
}
|
|
const child = spawn(process.execPath, [rootPath("scripts", "cli.ts"), "internal", "run-job", id], {
|
|
cwd: repoRoot,
|
|
detached: true,
|
|
stdio: "ignore",
|
|
env: process.env,
|
|
});
|
|
job.runnerPid = child.pid ?? null;
|
|
writeJob(job);
|
|
child.unref();
|
|
return job;
|
|
}
|
|
|
|
export async function runJob(id: string): Promise<JobRecord> {
|
|
const job = readJob(id);
|
|
job.status = "running";
|
|
job.runnerPid = process.pid;
|
|
job.startedAt = new Date().toISOString();
|
|
writeJob(job);
|
|
const exitCode = await runCommandToFiles(job.command, job.cwd, job.stdoutFile, job.stderrFile);
|
|
job.exitCode = exitCode;
|
|
job.status = exitCode === 0 ? "succeeded" : "failed";
|
|
job.finishedAt = new Date().toISOString();
|
|
writeJob(job);
|
|
return job;
|
|
}
|
|
|
|
export function jobWithTail(job: JobRecord, maxBytes = 12000): JobRecord & {
|
|
progress: JobProgressSummary;
|
|
tailPolicy: {
|
|
requestedTailBytes: number;
|
|
stdoutBytes: number;
|
|
stderrBytes: number;
|
|
stdoutTruncated: boolean;
|
|
stderrTruncated: boolean;
|
|
fullLogPaths: { stdoutFile: string; stderrFile: string };
|
|
};
|
|
stdoutTail: string;
|
|
stderrTail: string;
|
|
} {
|
|
const stdoutBytes = existsSync(job.stdoutFile) ? statSync(job.stdoutFile).size : 0;
|
|
const stderrBytes = existsSync(job.stderrFile) ? statSync(job.stderrFile).size : 0;
|
|
const progressTailBytes = Math.max(maxBytes, 96_000);
|
|
const stdoutProgressTail = tailFile(job.stdoutFile, progressTailBytes);
|
|
const stderrProgressTail = tailFile(job.stderrFile, progressTailBytes);
|
|
return {
|
|
...job,
|
|
progress: summarizeJobProgress(job, progressTailBytes, { stdoutTail: stdoutProgressTail, stderrTail: stderrProgressTail }),
|
|
tailPolicy: {
|
|
requestedTailBytes: maxBytes,
|
|
stdoutBytes,
|
|
stderrBytes,
|
|
stdoutTruncated: stdoutBytes > maxBytes,
|
|
stderrTruncated: stderrBytes > maxBytes,
|
|
fullLogPaths: { stdoutFile: job.stdoutFile, stderrFile: job.stderrFile },
|
|
},
|
|
stdoutTail: tailTextByBytes(stdoutProgressTail, maxBytes),
|
|
stderrTail: tailTextByBytes(stderrProgressTail, maxBytes),
|
|
};
|
|
}
|
|
|
|
function summarizeJobProgress(job: JobRecord, maxBytes = 96_000, tails?: { stdoutTail: string; stderrTail: string }): JobProgressSummary {
|
|
const nowMs = Date.now();
|
|
const knownWorkflow = job.name === "hwlab_g14_v02_trigger_current";
|
|
const v02PrMonitorWorkflow = job.name === "hwlab_g14_v02_pr_monitor";
|
|
const runtimeLaneTriggerWorkflow = /^hwlab_nodes_v[0-9]{2}_control-plane_trigger-current$/u.test(job.name);
|
|
const gitMirrorWorkflow = job.name === "hwlab_g14_git_mirror_sync" || job.name === "hwlab_g14_git_mirror_flush" || job.name === "agentrun_v01_git_mirror_sync" || job.name === "agentrun_v01_git_mirror_flush";
|
|
const ciInstallWorkflow = job.name === "ci_install";
|
|
if (ciInstallWorkflow) return summarizeCiInstallJobProgress(job, tails?.stderrTail, nowMs);
|
|
if (!knownWorkflow && !v02PrMonitorWorkflow && !runtimeLaneTriggerWorkflow && !gitMirrorWorkflow) return genericJobProgress(job, tails?.stderrTail);
|
|
const progressTailBytes = Math.max(4096, Math.floor(maxBytes));
|
|
const stderrTail = tails?.stderrTail ?? tailFile(job.stderrFile, progressTailBytes);
|
|
const stdoutTail = tails?.stdoutTail ?? tailFile(job.stdoutFile, progressTailBytes);
|
|
if (gitMirrorWorkflow) return summarizeGitMirrorJobProgress(job, stdoutTail, stderrTail, nowMs);
|
|
if (runtimeLaneTriggerWorkflow) return summarizeRuntimeLaneTriggerJobProgress(job, stdoutTail, stderrTail, nowMs);
|
|
const events = v02PrMonitorWorkflow
|
|
? [
|
|
...parseJsonLineEvents(stderrTail, "hwlab.v02.pr-monitor.progress"),
|
|
...parseJsonLineEvents(stderrTail, "hwlab.v02.trigger.progress"),
|
|
].sort((left, right) => String(left.at ?? "").localeCompare(String(right.at ?? "")))
|
|
: parseJsonLineEvents(stderrTail, "hwlab.v02.trigger.progress");
|
|
const lastEvent = events.at(-1) ?? {};
|
|
const stage = stringField(lastEvent.stage);
|
|
const stageStatus = stringField(lastEvent.status);
|
|
const sourceCommit = stringField(lastEvent.sourceCommit) ?? firstMatch(stdoutTail, /"sourceCommit"\s*:\s*"([0-9a-f]{40})"/iu);
|
|
const pipelineRun = stringField(lastEvent.pipelineRun) ?? firstMatch(stdoutTail, /"pipelineRun"\s*:\s*"([^"]+)"/u);
|
|
const pipelineCreated = /pipelinerun\.tekton\.dev\/[^ \n]+ created/u.test(stdoutTail)
|
|
? true
|
|
: stage === "create-pipelinerun" && stageStatus === "failed"
|
|
? false
|
|
: null;
|
|
const lastEventAt = stringField(lastEvent.at);
|
|
const kind = events.length > 0 || knownWorkflow || v02PrMonitorWorkflow ? "hwlab-v02-trigger" : "generic";
|
|
const elapsedSeconds = jobElapsedSeconds(job, nowMs);
|
|
const stageElapsedSeconds = currentStageElapsedSeconds(events, stage, stageStatus, job, nowMs);
|
|
const lastEventAgeSeconds = lastEventAt === null ? null : secondsSince(lastEventAt, job.finishedAt ?? nowMs);
|
|
const timings = extractGitMirrorTimings(`${stderrTail}\n${stdoutTail}`);
|
|
const warnings = jobProgressWarnings({
|
|
job,
|
|
eventsObserved: events.length,
|
|
elapsedSeconds,
|
|
stage,
|
|
stageStatus,
|
|
stageElapsedSeconds,
|
|
lastEventAgeSeconds,
|
|
});
|
|
const slow = warnings.length > 0;
|
|
const nextCommand = pipelineRun
|
|
? `bun scripts/cli.ts hwlab g14 control-plane status --lane v02`
|
|
: job.status === "running"
|
|
? `bun scripts/cli.ts job status ${job.id} --tail-bytes 12000`
|
|
: null;
|
|
const summary = kind === "hwlab-v02-trigger"
|
|
? [
|
|
job.status,
|
|
stage ? `${stage}${stageStatus ? `:${stageStatus}` : ""}` : "stage:unknown",
|
|
sourceCommit ? `source=${sourceCommit.slice(0, 12)}` : null,
|
|
pipelineRun ? `pipelineRun=${pipelineRun}` : null,
|
|
pipelineCreated === true ? "created" : pipelineCreated === false ? "create-failed" : null,
|
|
elapsedSeconds !== null ? `elapsed=${elapsedSeconds}s` : null,
|
|
stageElapsedSeconds !== null && job.status === "running" ? `stageElapsed=${stageElapsedSeconds}s` : null,
|
|
lastEventAgeSeconds !== null && job.status === "running" ? `lastEventAge=${lastEventAgeSeconds}s` : null,
|
|
slow ? "visibility-warning" : null,
|
|
].filter(Boolean).join(" ")
|
|
: `${job.status}${job.exitCode === null ? "" : ` exit=${job.exitCode}`}`;
|
|
return {
|
|
kind,
|
|
stage,
|
|
stageStatus,
|
|
sourceCommit,
|
|
pipelineRun,
|
|
pipelineCreated,
|
|
elapsedSeconds,
|
|
stageElapsedSeconds,
|
|
lastEventAt,
|
|
lastEventAgeSeconds,
|
|
eventsObserved: events.length,
|
|
slow,
|
|
warnings,
|
|
timings,
|
|
summary,
|
|
nextCommand,
|
|
};
|
|
}
|
|
|
|
function summarizeGitMirrorJobProgress(job: JobRecord, stdoutTail: string, stderrTail: string, nowMs = Date.now()): JobProgressSummary {
|
|
const action = job.name.endsWith("_flush") ? "flush" : job.name.endsWith("_sync") ? "sync" : "unknown";
|
|
const elapsedSeconds = jobElapsedSeconds(job, nowMs);
|
|
const timings = extractGitMirrorTimings(`${stderrTail}\n${stdoutTail}`);
|
|
const status =
|
|
firstMatch(stdoutTail, new RegExp(`"event"\\s*:\\s*"git-mirror-${action}"[\\s\\S]{0,240}?"status"\\s*:\\s*"([^"]+)"`, "u")) ??
|
|
firstMatch(stdoutTail.replaceAll('\\"', '"').replaceAll("\\n", "\n"), new RegExp(`"event"\\s*:\\s*"git-mirror-${action}"[\\s\\S]{0,240}?"status"\\s*:\\s*"([^"]+)"`, "u"));
|
|
const pendingFlush = firstMatch(stdoutTail.replaceAll('\\"', '"'), /"pendingFlush"\s*:\s*(true|false)/u);
|
|
const jobName = firstMatch(stdoutTail, /"jobName"\s*:\s*"([^"]+)"/u);
|
|
const warnings = jobProgressWarnings({
|
|
job,
|
|
eventsObserved: 0,
|
|
elapsedSeconds,
|
|
stage: action,
|
|
stageStatus: job.status === "running" ? "running" : status,
|
|
stageElapsedSeconds: job.status === "running" ? elapsedSeconds : null,
|
|
lastEventAgeSeconds: null,
|
|
});
|
|
const timingSummary = Object.entries(timings)
|
|
.filter(([key]) => key.startsWith(`gitMirror.${action}.`) || key === "sshRuntimeMaxMs")
|
|
.map(([key, value]) => `${key}=${value}ms`);
|
|
return {
|
|
kind: "git-mirror",
|
|
stage: action,
|
|
stageStatus: status ?? (job.status === "running" ? "running" : null),
|
|
sourceCommit: null,
|
|
pipelineRun: null,
|
|
pipelineCreated: null,
|
|
elapsedSeconds,
|
|
stageElapsedSeconds: job.status === "running" ? elapsedSeconds : null,
|
|
lastEventAt: null,
|
|
lastEventAgeSeconds: null,
|
|
eventsObserved: 0,
|
|
slow: warnings.length > 0,
|
|
warnings,
|
|
timings,
|
|
summary: [
|
|
job.status,
|
|
`git-mirror-${action}${status ? `:${status}` : ""}`,
|
|
jobName ? `job=${jobName}` : null,
|
|
pendingFlush !== null ? `pendingFlush=${pendingFlush}` : null,
|
|
elapsedSeconds !== null ? `elapsed=${elapsedSeconds}s` : null,
|
|
...timingSummary,
|
|
warnings.length > 0 ? "visibility-warning" : null,
|
|
].filter(Boolean).join(" "),
|
|
nextCommand: job.status === "running"
|
|
? `bun scripts/cli.ts job status ${job.id} --tail-bytes 12000`
|
|
: job.name.startsWith("agentrun_")
|
|
? "bun scripts/cli.ts agentrun git-mirror status"
|
|
: "bun scripts/cli.ts hwlab g14 git-mirror status",
|
|
};
|
|
}
|
|
|
|
function summarizeCiInstallJobProgress(job: JobRecord, stderrTailOverride?: string, nowMs = Date.now()): JobProgressSummary {
|
|
const stderrTail = stderrTailOverride ?? tailFile(job.stderrFile, 96_000);
|
|
const events = parseJsonLineEvents(stderrTail, "ci.install.progress")
|
|
.sort((left, right) => String(left.at ?? "").localeCompare(String(right.at ?? "")));
|
|
const lastEvent = events.at(-1) ?? {};
|
|
const stage = stringField(lastEvent.stage);
|
|
const stageStatus = stringField(lastEvent.status);
|
|
const lastEventAt = stringField(lastEvent.at);
|
|
const elapsedSeconds = jobElapsedSeconds(job, nowMs);
|
|
const stageElapsedSeconds = currentStageElapsedSeconds(events, stage, stageStatus, job, nowMs);
|
|
const lastEventAgeSeconds = lastEventAt === null ? null : secondsSince(lastEventAt, job.finishedAt ?? nowMs);
|
|
const warnings = jobProgressWarnings({
|
|
job,
|
|
eventsObserved: events.length,
|
|
elapsedSeconds,
|
|
stage,
|
|
stageStatus,
|
|
stageElapsedSeconds,
|
|
lastEventAgeSeconds,
|
|
});
|
|
return {
|
|
kind: "generic",
|
|
stage,
|
|
stageStatus,
|
|
sourceCommit: null,
|
|
pipelineRun: null,
|
|
pipelineCreated: null,
|
|
elapsedSeconds,
|
|
stageElapsedSeconds,
|
|
lastEventAt,
|
|
lastEventAgeSeconds,
|
|
eventsObserved: events.length,
|
|
slow: warnings.length > 0,
|
|
warnings,
|
|
timings: {},
|
|
summary: [
|
|
job.status,
|
|
stage ? `${stage}${stageStatus ? `:${stageStatus}` : ""}` : "stage:unknown",
|
|
typeof lastEvent.providerId === "string" ? `provider=${lastEvent.providerId}` : null,
|
|
typeof lastEvent.manifest === "string" ? `manifest=${lastEvent.manifest}` : null,
|
|
elapsedSeconds !== null ? `elapsed=${elapsedSeconds}s` : null,
|
|
stageElapsedSeconds !== null && job.status === "running" ? `stageElapsed=${stageElapsedSeconds}s` : null,
|
|
lastEventAgeSeconds !== null && job.status === "running" ? `lastEventAge=${lastEventAgeSeconds}s` : null,
|
|
warnings.length > 0 ? "visibility-warning" : null,
|
|
].filter(Boolean).join(" "),
|
|
nextCommand: job.status === "running"
|
|
? `bun scripts/cli.ts ci install-status ${job.id}`
|
|
: "bun scripts/cli.ts ci status",
|
|
};
|
|
}
|
|
|
|
function summarizeRuntimeLaneTriggerJobProgress(job: JobRecord, stdoutTail: string, stderrTail: string, nowMs = Date.now()): JobProgressSummary {
|
|
const events = parseJsonLineEvents(stderrTail, "hwlab.runtime-lane.trigger.progress")
|
|
.sort((left, right) => String(left.at ?? "").localeCompare(String(right.at ?? "")));
|
|
const lastEvent = events.at(-1) ?? {};
|
|
const stage = stringField(lastEvent.stage);
|
|
const stageStatus = stringField(lastEvent.status);
|
|
const sourceCommit = stringField(lastEvent.sourceCommit) ?? firstMatch(stdoutTail, /"sourceCommit"\s*:\s*"([0-9a-f]{40})"/iu);
|
|
const pipelineRun = stringField(lastEvent.pipelineRun) ?? firstMatch(stdoutTail, /"pipelineRun"\s*:\s*"([^"]+)"/u);
|
|
const pipelineCreated = /pipelinerun\.tekton\.dev\/[^ \n]+ created/u.test(stdoutTail)
|
|
? true
|
|
: stage === "create-pipelinerun" && stageStatus === "failed"
|
|
? false
|
|
: null;
|
|
const lastEventAt = stringField(lastEvent.at);
|
|
const elapsedSeconds = jobElapsedSeconds(job, nowMs);
|
|
const stageElapsedSeconds = currentStageElapsedSeconds(events, stage, stageStatus, job, nowMs);
|
|
const lastEventAgeSeconds = lastEventAt === null ? null : secondsSince(lastEventAt, job.finishedAt ?? nowMs);
|
|
const timings = extractGitMirrorTimings(`${stderrTail}\n${stdoutTail}`);
|
|
const warnings = jobProgressWarnings({
|
|
job,
|
|
eventsObserved: events.length,
|
|
elapsedSeconds,
|
|
stage,
|
|
stageStatus,
|
|
stageElapsedSeconds,
|
|
lastEventAgeSeconds,
|
|
});
|
|
const nextStatusCommand = hwlabRuntimeLaneStatusNextCommand(pipelineRun, lastEvent);
|
|
if (nextStatusCommand.warning !== null) warnings.push(nextStatusCommand.warning);
|
|
const tcpPoolDiagnostics = job.status === "succeeded" ? null : sshTcpPoolDiagnosticsFromJobText(`${stderrTail}\n${stdoutTail}`);
|
|
const slow = warnings.length > 0;
|
|
return {
|
|
kind: "hwlab-runtime-lane-trigger",
|
|
stage,
|
|
stageStatus,
|
|
sourceCommit,
|
|
pipelineRun,
|
|
pipelineCreated,
|
|
elapsedSeconds,
|
|
stageElapsedSeconds,
|
|
lastEventAt,
|
|
lastEventAgeSeconds,
|
|
eventsObserved: events.length,
|
|
slow,
|
|
warnings,
|
|
diagnostics: tcpPoolDiagnostics,
|
|
timings,
|
|
summary: [
|
|
job.status,
|
|
stage ? `${stage}${stageStatus ? `:${stageStatus}` : ""}` : "stage:unknown",
|
|
sourceCommit ? `source=${sourceCommit.slice(0, 12)}` : null,
|
|
pipelineRun ? `pipelineRun=${pipelineRun}` : null,
|
|
pipelineCreated === true ? "created" : pipelineCreated === false ? "create-failed" : null,
|
|
elapsedSeconds !== null ? `elapsed=${elapsedSeconds}s` : null,
|
|
stageElapsedSeconds !== null && job.status === "running" ? `stageElapsed=${stageElapsedSeconds}s` : null,
|
|
lastEventAgeSeconds !== null && job.status === "running" ? `lastEventAge=${lastEventAgeSeconds}s` : null,
|
|
tcpPoolDiagnostics !== null ? `sshPool=${String(tcpPoolDiagnostics.failureKind ?? "transient")}` : null,
|
|
slow ? "visibility-warning" : null,
|
|
].filter(Boolean).join(" "),
|
|
nextCommand: pipelineRun
|
|
? nextStatusCommand.command
|
|
: job.status === "running"
|
|
? `bun scripts/cli.ts job status ${job.id} --tail-bytes 12000`
|
|
: null,
|
|
};
|
|
}
|
|
|
|
function hwlabRuntimeLaneStatusNextCommand(pipelineRun: string | null, lastEvent: Record<string, unknown>): { command: string | null; warning: string | null } {
|
|
if (pipelineRun === null) return { command: null, warning: null };
|
|
const eventNode = stringField(lastEvent.node);
|
|
const eventLane = stringField(lastEvent.lane);
|
|
const defaults = hwlabDefaultRuntimeTarget();
|
|
const node = eventNode ?? defaults.node;
|
|
const lane = eventLane ?? defaults.lane;
|
|
const missing = [
|
|
eventNode === null ? "node" : null,
|
|
eventLane === null ? "lane" : null,
|
|
].filter((value): value is string => value !== null);
|
|
return {
|
|
command: `bun scripts/cli.ts hwlab nodes control-plane status --node ${node} --lane ${lane} --pipeline-run ${pipelineRun}`,
|
|
warning: missing.length === 0
|
|
? null
|
|
: `runtime lane progress event missing ${missing.join("/")} for nextCommand; used YAML default ${defaults.node}/${defaults.lane} from config/hwlab-node-lanes.yaml`,
|
|
};
|
|
}
|
|
|
|
function sshTcpPoolDiagnosticsFromJobText(text: string): Record<string, unknown> | null {
|
|
const hint = lastJsonLinePayload(text, "UNIDESK_SSH_TCP_POOL_HINT");
|
|
if (hint !== null) return hint;
|
|
const error = lastJsonLinePayload(text, "UNIDESK_SSH_ERROR");
|
|
if (error !== null && typeof error.failureKind === "string" && error.failureKind.startsWith("provider-data-")) {
|
|
return {
|
|
code: "ssh-tcp-pool-transient",
|
|
failureKind: error.failureKind,
|
|
providerId: error.providerId ?? null,
|
|
dataChannelId: error.dataChannelId ?? null,
|
|
dataPool: error.dataPool ?? null,
|
|
diagnostics: { fullHealth: "bun scripts/cli.ts debug health" },
|
|
};
|
|
}
|
|
if (/ssh tcp data channel closed/iu.test(text)) {
|
|
return { code: "ssh-tcp-pool-transient", failureKind: "provider-data-channel-closed", diagnostics: { fullHealth: "bun scripts/cli.ts debug health" } };
|
|
}
|
|
if (/requested ssh tcp data channel is not ready|ssh tcp data channel is not available/iu.test(text)) {
|
|
return { code: "ssh-tcp-pool-transient", failureKind: "provider-data-channel-missing", diagnostics: { fullHealth: "bun scripts/cli.ts debug health" } };
|
|
}
|
|
if (/provider ssh tcp data pool has no idle channel/iu.test(text)) {
|
|
return { code: "ssh-tcp-pool-transient", failureKind: "provider-data-pool-exhausted", diagnostics: { fullHealth: "bun scripts/cli.ts debug health" } };
|
|
}
|
|
return null;
|
|
}
|
|
|
|
function lastJsonLinePayload(text: string, prefix: string): Record<string, unknown> | null {
|
|
const lines = text.split(/\r?\n/u);
|
|
for (let index = lines.length - 1; index >= 0; index -= 1) {
|
|
const line = lines[index]?.trim() ?? "";
|
|
if (!line.startsWith(`${prefix} `)) continue;
|
|
try {
|
|
const parsed = JSON.parse(line.slice(prefix.length + 1)) as unknown;
|
|
if (typeof parsed === "object" && parsed !== null && !Array.isArray(parsed)) return parsed as Record<string, unknown>;
|
|
} catch {
|
|
return null;
|
|
}
|
|
}
|
|
return null;
|
|
}
|
|
|
|
function genericJobProgress(job: JobRecord, stderrTailOverride?: string): JobProgressSummary {
|
|
const nowMs = Date.now();
|
|
const stderrTail = stderrTailOverride ?? tailFile(job.stderrFile, 96_000);
|
|
const downloadProgressEvents = parseJsonLineEvents(stderrTail, "unidesk.ssh.download.progress");
|
|
const downloadRetryEvents = [
|
|
...parseJsonLineEvents(stderrTail, "unidesk.ssh.download.empty-read-retry"),
|
|
...parseJsonLineEvents(stderrTail, "unidesk.ssh.download.short-read-retry"),
|
|
];
|
|
const downloadEvents = [...downloadProgressEvents, ...downloadRetryEvents].sort((left, right) => Number(left.blockIndex ?? -1) - Number(right.blockIndex ?? -1));
|
|
const lastDownload = downloadEvents.at(-1) ?? null;
|
|
const lastDownloadAt = stringField(lastDownload?.at);
|
|
const lastEventAgeSeconds = lastDownloadAt === null ? null : secondsSince(lastDownloadAt, job.finishedAt ?? nowMs);
|
|
const elapsedSeconds = jobElapsedSeconds(job, nowMs);
|
|
const warnings = jobProgressWarnings({
|
|
job,
|
|
eventsObserved: downloadEvents.length,
|
|
elapsedSeconds,
|
|
stage: lastDownload === null ? null : "ssh-download",
|
|
stageStatus: lastDownload === null ? null : "running",
|
|
stageElapsedSeconds: null,
|
|
lastEventAgeSeconds,
|
|
});
|
|
const downloadSummary = lastDownload === null
|
|
? null
|
|
: [
|
|
` ssh-download ${Number(lastDownload.actualBytes ?? lastDownload.bytes ?? 0)}/${Number(lastDownload.expectedBytes ?? lastDownload.totalBytes ?? 0)} bytes`,
|
|
`chunks=${Number(lastDownload.chunks ?? 0)}`,
|
|
`remaining=${Number(lastDownload.remainingBytes ?? 0)}`,
|
|
`throughput=${Number(lastDownload.throughputBytesPerSecond ?? 0)}Bps`,
|
|
`elapsedMs=${Number(lastDownload.elapsedMs ?? 0)}`,
|
|
downloadRetryEvents.length === 0 ? null : `retries=${downloadRetryEvents.length} lastRetry=${String(lastDownload.reason ?? "unknown")}`,
|
|
].filter(Boolean).join(" ");
|
|
return {
|
|
kind: "generic",
|
|
stage: lastDownload === null ? null : "ssh-download",
|
|
stageStatus: lastDownload === null ? null : "running",
|
|
sourceCommit: null,
|
|
pipelineRun: null,
|
|
pipelineCreated: null,
|
|
elapsedSeconds,
|
|
stageElapsedSeconds: null,
|
|
lastEventAt: lastDownloadAt,
|
|
lastEventAgeSeconds,
|
|
eventsObserved: downloadEvents.length,
|
|
slow: warnings.length > 0,
|
|
warnings,
|
|
timings: {},
|
|
summary: `${job.status}${job.exitCode === null ? "" : ` exit=${job.exitCode}`}${downloadSummary ?? ""}`,
|
|
nextCommand: job.status === "running" ? `bun scripts/cli.ts job status ${job.id} --tail-bytes 12000` : null,
|
|
};
|
|
}
|
|
|
|
function jobElapsedSeconds(job: JobRecord, nowMs = Date.now()): number | null {
|
|
const startMs = timestampMs(job.startedAt ?? job.createdAt);
|
|
if (startMs === null) return null;
|
|
const endMs = timestampMs(job.finishedAt) ?? nowMs;
|
|
if (endMs < startMs) return null;
|
|
return Math.round((endMs - startMs) / 1000);
|
|
}
|
|
|
|
function secondsSince(value: string, end: string | number = Date.now()): number | null {
|
|
const startMs = timestampMs(value);
|
|
const endMs = typeof end === "number" ? end : timestampMs(end);
|
|
if (startMs === null || endMs === null || endMs < startMs) return null;
|
|
return Math.round((endMs - startMs) / 1000);
|
|
}
|
|
|
|
function timestampMs(value: unknown): number | null {
|
|
if (typeof value !== "string" || value.trim().length === 0) return null;
|
|
const parsed = Date.parse(value);
|
|
return Number.isFinite(parsed) ? parsed : null;
|
|
}
|
|
|
|
function currentStageElapsedSeconds(events: Record<string, unknown>[], stage: string | null, stageStatus: string | null, job: JobRecord, nowMs = Date.now()): number | null {
|
|
if (stage === null) return null;
|
|
const stageEvents = events.filter((event) => stringField(event.stage) === stage);
|
|
const startEvent = stageEvents.findLast((event) => stringField(event.status) === "started");
|
|
const startAt = stringField(startEvent?.at);
|
|
if (startAt === null) return null;
|
|
const endEvent = stageStatus === "started"
|
|
? null
|
|
: stageEvents.findLast((event) => {
|
|
const status = stringField(event.status);
|
|
return status !== null && status !== "started";
|
|
});
|
|
const endAt = stringField(endEvent?.at) ?? job.finishedAt ?? (job.status === "running" ? new Date(nowMs).toISOString() : null);
|
|
return endAt === null ? null : secondsSince(startAt, endAt);
|
|
}
|
|
|
|
function jobProgressWarnings(input: {
|
|
job: JobRecord;
|
|
eventsObserved: number;
|
|
elapsedSeconds: number | null;
|
|
stage: string | null;
|
|
stageStatus: string | null;
|
|
stageElapsedSeconds: number | null;
|
|
lastEventAgeSeconds: number | null;
|
|
}): string[] {
|
|
if (input.job.status !== "running") return [];
|
|
if (input.stage === "idle" && input.stageStatus === "waiting") return [];
|
|
const warnings: string[] = [];
|
|
if (input.eventsObserved === 0 && (input.elapsedSeconds ?? 0) >= 30) {
|
|
warnings.push(`running for ${input.elapsedSeconds}s without progress events; inspect stdoutTail/stderrTail and fullLogPaths`);
|
|
}
|
|
if ((input.lastEventAgeSeconds ?? 0) >= 30) {
|
|
warnings.push(`no progress event for ${input.lastEventAgeSeconds}s; current stage may be blocked or waiting on remote I/O`);
|
|
}
|
|
if (input.stage !== null && input.stageStatus !== null && input.stageStatus !== "succeeded" && input.stageStatus !== "failed" && (input.stageElapsedSeconds ?? 0) >= 30) {
|
|
warnings.push(`stage ${input.stage}:${input.stageStatus} has been active for ${input.stageElapsedSeconds}s`);
|
|
}
|
|
return warnings;
|
|
}
|
|
|
|
function tailTextByBytes(text: string, maxBytes: number): string {
|
|
const safeMaxBytes = Math.max(0, Math.floor(maxBytes));
|
|
if (safeMaxBytes === 0) return "";
|
|
const buffer = Buffer.from(text, "utf8");
|
|
if (buffer.length <= safeMaxBytes) return text;
|
|
return buffer.subarray(buffer.length - safeMaxBytes).toString("utf8");
|
|
}
|
|
|
|
function parseJsonLineEvents(text: string, eventName: string): Record<string, unknown>[] {
|
|
const events: Record<string, unknown>[] = [];
|
|
for (const line of text.split(/\r?\n/u)) {
|
|
const trimmed = line.trim();
|
|
if (!trimmed.startsWith("{")) continue;
|
|
try {
|
|
const parsed = JSON.parse(trimmed) as unknown;
|
|
if (typeof parsed === "object" && parsed !== null && !Array.isArray(parsed) && (parsed as Record<string, unknown>).event === eventName) {
|
|
events.push(parsed as Record<string, unknown>);
|
|
}
|
|
} catch {
|
|
// Ignore non-JSON stderr lines; the raw tail remains available in stderrTail.
|
|
}
|
|
}
|
|
return events;
|
|
}
|
|
|
|
function extractGitMirrorTimings(text: string): Record<string, number> {
|
|
const timings: Record<string, number> = {};
|
|
for (const normalizedText of [text, text.replaceAll('\\"', '"').replaceAll("\\n", "\n")]) {
|
|
extractGitMirrorTimingJsonLines(normalizedText, timings);
|
|
extractGitMirrorTimingRegex(normalizedText, timings);
|
|
extractSshRuntimeTimingRegex(normalizedText, timings);
|
|
}
|
|
return timings;
|
|
}
|
|
|
|
function extractGitMirrorTimingJsonLines(text: string, timings: Record<string, number>): void {
|
|
for (const line of text.split(/\r?\n/u)) {
|
|
const trimmed = line.trim();
|
|
if (!trimmed.startsWith("{")) continue;
|
|
try {
|
|
const parsed = JSON.parse(trimmed) as unknown;
|
|
if (typeof parsed !== "object" || parsed === null || Array.isArray(parsed)) continue;
|
|
const event = parsed as Record<string, unknown>;
|
|
const operation = event.event === "git-mirror-sync-timing"
|
|
? "sync"
|
|
: event.event === "git-mirror-flush-timing"
|
|
? "flush"
|
|
: null;
|
|
if (operation === null) continue;
|
|
const phase = stringField(event.phase);
|
|
const durationMs = typeof event.durationMs === "number" && Number.isFinite(event.durationMs) ? event.durationMs : null;
|
|
if (phase !== null && durationMs !== null) {
|
|
timings[`gitMirror.${operation}.${phase}Ms`] = Math.round(durationMs);
|
|
if (operation === "sync") timings[`gitMirror.${phase}Ms`] = Math.round(durationMs);
|
|
}
|
|
} catch {
|
|
// Ignore non-JSON lines; raw logs remain available through job status tails.
|
|
}
|
|
}
|
|
}
|
|
|
|
function extractGitMirrorTimingRegex(text: string, timings: Record<string, number>): void {
|
|
const pattern = /"event"\s*:\s*"git-mirror-(sync|flush)-timing"[\s\S]{0,240}?"phase"\s*:\s*"([^"]+)"[\s\S]{0,240}?"durationMs"\s*:\s*(\d+)/gu;
|
|
for (const match of text.matchAll(pattern)) {
|
|
const operation = match[1];
|
|
const phase = match[2];
|
|
const durationMs = Number(match[3]);
|
|
if (operation !== undefined && phase !== undefined && Number.isFinite(durationMs)) {
|
|
timings[`gitMirror.${operation}.${phase}Ms`] = Math.round(durationMs);
|
|
if (operation === "sync") timings[`gitMirror.${phase}Ms`] = Math.round(durationMs);
|
|
}
|
|
}
|
|
}
|
|
|
|
function extractSshRuntimeTimingRegex(text: string, timings: Record<string, number>): void {
|
|
const pattern = /"code"\s*:\s*"ssh-runtime-timing"[\s\S]{0,320}?"elapsedMs"\s*:\s*(\d+)/gu;
|
|
const values = [...text.matchAll(pattern)]
|
|
.map((match) => Number(match[1]))
|
|
.filter((value) => Number.isFinite(value));
|
|
if (values.length > 0) timings.sshRuntimeMaxMs = Math.max(...values);
|
|
}
|
|
|
|
function stringField(value: unknown): string | null {
|
|
return typeof value === "string" && value.trim().length > 0 ? value.trim() : null;
|
|
}
|
|
|
|
function firstMatch(text: string, pattern: RegExp): string | null {
|
|
const match = pattern.exec(text);
|
|
return typeof match?.[1] === "string" && match[1].length > 0 ? match[1] : null;
|
|
}
|
|
|
|
export interface JobListOptions {
|
|
limit?: number;
|
|
includeCommand?: boolean;
|
|
}
|
|
|
|
export function listJobsSummary(options: JobListOptions = {}): unknown {
|
|
const limit = Math.max(1, Math.floor(options.limit ?? 50));
|
|
const jobs = listJobs();
|
|
const returned = jobs.slice(0, limit).map((job) => ({
|
|
id: job.id,
|
|
name: job.name,
|
|
status: job.status,
|
|
progress: summarizeJobProgress(job, 32_000),
|
|
runner: job.runner,
|
|
runnerPid: job.runnerPid ?? null,
|
|
runnerContainer: job.runnerContainer ?? null,
|
|
createdAt: job.createdAt,
|
|
startedAt: job.startedAt,
|
|
finishedAt: job.finishedAt,
|
|
exitCode: job.exitCode,
|
|
note: job.note,
|
|
stdoutFile: job.stdoutFile,
|
|
stderrFile: job.stderrFile,
|
|
...(options.includeCommand === true ? { command: job.command, cwd: job.cwd } : {}),
|
|
}));
|
|
return {
|
|
jobs: returned,
|
|
total: jobs.length,
|
|
returned: returned.length,
|
|
limit,
|
|
truncated: jobs.length > returned.length,
|
|
disclosure: {
|
|
defaultLimit: 50,
|
|
nextCommand: jobs.length > returned.length ? `bun scripts/cli.ts job list --limit ${Math.min(jobs.length, limit * 2)}` : null,
|
|
includeCommandCommand: "bun scripts/cli.ts job list --include-command",
|
|
statusCommand: "bun scripts/cli.ts job status <jobId> --tail-bytes 12000",
|
|
},
|
|
};
|
|
}
|