mirror of
https://github.com/PR0M3TH3AN/TORCH.git
synced 2026-10-27 07:51:52 +00:00
[scheduler] daily/decompose-agent: Decompose scripts/agent/run-scheduler-cycle.mjs
- Executed daily scheduler cycle. - Selected `decompose-agent` (next in roster). - Decomposed `scripts/agent/run-scheduler-cycle.mjs` (1127 lines) into `scripts/agent/scheduler-utils.mjs` (195 lines) and `scripts/agent/scheduler-lock.mjs` (173 lines). - Resulting main file size: 788 lines (339 lines reduction). - Fixed lint warnings. - Updated `test/scheduler-preflight-lock.e2e.test.mjs`, `test/run-scheduler-cycle-memory-policy.test.mjs`, `test/scheduler-lock-failure-schema.contract.test.mjs` to copy new modules into test fixtures. - Updated `test/scheduler-ratchet.test.mjs` to verify moved functions in `scheduler-lock.mjs`. - Note: `scripts/check-file-size.mjs` was not updated as it lacks the baseline feature referenced in the prompt. - Created artifacts: `src/context/CONTEXT_...`, `src/decisions/DECISIONS_...`, `src/test_logs/TEST_LOG_...`, `src/todo/TODO_...`. - Created scheduler log: `task-logs/daily/2026-02-17T22-29-00Z__decompose-agent__completed.md`. Co-authored-by: PR0M3TH3AN <[email protected]>
This commit is contained in:
1 parent
7f45ff84b4
commit
9e993800fa
13 files changed
+712
-359
No files matched your search
@@ -3,9 +3,28 @@
|
||||
// implemented by this script; step 3 (policy-file read) is best-effort and non-fatal.
|
||||
import fs from 'node:fs/promises';
|
||||
import path from 'node:path';
|
||||
import { spawn } from 'node:child_process';
|
||||
import { randomUUID } from 'node:crypto';
|
||||
|
||||
import {
|
||||
runCommand,
|
||||
parseJsonFromOutput,
|
||||
readJson,
|
||||
normalizeStringList,
|
||||
parseNonNegativeInt,
|
||||
parseBooleanFlag,
|
||||
excerptText,
|
||||
getRunDateKey,
|
||||
toYamlScalar,
|
||||
} from './scheduler-utils.mjs';
|
||||
|
||||
import {
|
||||
classifyLockBackendError,
|
||||
buildLockBackendRemediation,
|
||||
runLockHealthPreflight,
|
||||
acquireLockWithRetry,
|
||||
summarizeLockFailureReasons,
|
||||
} from './scheduler-lock.mjs';
|
||||
|
||||
const VALID_CADENCES = new Set(['daily', 'weekly']);
|
||||
const ALL_EXCLUDED_REASON = 'All roster tasks currently claimed by other agents';
|
||||
const FAILURE_CATEGORY = {
|
||||
@@ -15,8 +34,6 @@ const FAILURE_CATEGORY = {
|
||||
EXECUTION: 'execution_error',
|
||||
};
|
||||
|
||||
const LOCK_INCIDENT_LINK = 'docs/agent-handoffs/learnings/2026-02-15-relay-health-preflight-job.md';
|
||||
|
||||
function parseArgs(argv) {
|
||||
const args = { cadence: null, platform: process.env.AGENT_PLATFORM || 'codex', model: process.env.AGENT_MODEL || null };
|
||||
for (let i = 0; i < argv.length; i += 1) {
|
||||
@@ -46,180 +63,6 @@ function ts() {
|
||||
return new Date().toISOString().replace(/\.\d{3}Z$/, 'Z').replace(/:/g, '-');
|
||||
}
|
||||
|
||||
function parseJsonFromOutput(text) {
|
||||
const trimmed = String(text || '').trim();
|
||||
if (!trimmed) return null;
|
||||
const lines = trimmed.split(/\r?\n/).map((line) => line.trim()).filter(Boolean);
|
||||
for (let i = lines.length - 1; i >= 0; i -= 1) {
|
||||
const line = lines[i];
|
||||
if (!line.startsWith('{') && !line.startsWith('[')) continue;
|
||||
try {
|
||||
return JSON.parse(line);
|
||||
} catch {
|
||||
// Keep scanning from the end.
|
||||
}
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
function parseJsonEventsFromOutput(text) {
|
||||
const events = [];
|
||||
for (const line of String(text || '').split(/\r?\n/)) {
|
||||
const trimmed = line.trim();
|
||||
if (!trimmed.startsWith('{')) continue;
|
||||
try {
|
||||
const parsed = JSON.parse(trimmed);
|
||||
if (parsed && typeof parsed === 'object' && !Array.isArray(parsed)) {
|
||||
events.push(parsed);
|
||||
}
|
||||
} catch {
|
||||
// Ignore non-JSON lines.
|
||||
}
|
||||
}
|
||||
return events;
|
||||
}
|
||||
|
||||
const ALLOWED_ENV_KEYS = new Set([
|
||||
'PATH', 'Path', // Windows compatibility
|
||||
'HOME', 'USER', 'SHELL', 'TERM', 'LANG', 'LC_ALL', 'LC_CTYPE', 'TZ',
|
||||
'NODE_ENV', 'NODE_OPTIONS', 'NODE_PATH',
|
||||
'APPDATA', 'LOCALAPPDATA', 'PROGRAMDATA', 'SystemRoot', 'windir', 'ComSpec', 'PATHEXT',
|
||||
'TMPDIR', 'TEMP', 'TMP',
|
||||
'EDITOR', 'VISUAL',
|
||||
// CI variables
|
||||
'CI', 'GITHUB_ACTIONS', 'GITHUB_REF', 'GITHUB_HEAD_REF', 'GITHUB_BASE_REF', 'GITHUB_EVENT_NAME', 'GITHUB_SHA',
|
||||
]);
|
||||
|
||||
function getSafeEnv() {
|
||||
const safeEnv = {};
|
||||
for (const key in process.env) {
|
||||
if (ALLOWED_ENV_KEYS.has(key) ||
|
||||
key.startsWith('npm_') ||
|
||||
key.startsWith('NOSTR_') ||
|
||||
key.startsWith('TORCH_') ||
|
||||
key.startsWith('SCHEDULER_') ||
|
||||
key.startsWith('AGENT_')) {
|
||||
safeEnv[key] = process.env[key];
|
||||
}
|
||||
}
|
||||
return safeEnv;
|
||||
}
|
||||
|
||||
/**
|
||||
* Spawns a child process with a sanitized environment by default.
|
||||
* @param {string} command
|
||||
* @param {string[]} args
|
||||
* @param {object} options
|
||||
* @param {object} [options.env] - Additional environment variables to merge.
|
||||
* @param {boolean} [options.inheritProcessEnv=false] - If true, use full process.env. If false (default), use sanitized env.
|
||||
*/
|
||||
async function runCommand(command, args = [], options = {}) {
|
||||
const baseEnv = options.inheritProcessEnv ? process.env : getSafeEnv();
|
||||
return new Promise((resolve) => {
|
||||
const child = spawn(command, args, {
|
||||
cwd: process.cwd(),
|
||||
env: { ...baseEnv, ...(options.env || {}) },
|
||||
stdio: ['ignore', 'pipe', 'pipe'],
|
||||
});
|
||||
|
||||
let stdout = '';
|
||||
let stderr = '';
|
||||
|
||||
const OUTPUT_LIMIT = 20000; // 20KB
|
||||
let stdoutWritten = 0;
|
||||
let stderrWritten = 0;
|
||||
let stdoutTruncated = false;
|
||||
let stderrTruncated = false;
|
||||
|
||||
child.stdout.on('data', (chunk) => {
|
||||
const text = chunk.toString();
|
||||
stdout += text;
|
||||
|
||||
if (stdoutWritten < OUTPUT_LIMIT) {
|
||||
const remaining = OUTPUT_LIMIT - stdoutWritten;
|
||||
if (text.length <= remaining) {
|
||||
process.stdout.write(text);
|
||||
stdoutWritten += text.length;
|
||||
} else {
|
||||
process.stdout.write(text.slice(0, remaining));
|
||||
stdoutWritten += remaining;
|
||||
}
|
||||
} else if (!stdoutTruncated) {
|
||||
process.stdout.write('\n...[stdout truncated]...\n');
|
||||
stdoutTruncated = true;
|
||||
}
|
||||
});
|
||||
|
||||
child.stderr.on('data', (chunk) => {
|
||||
const text = chunk.toString();
|
||||
stderr += text;
|
||||
|
||||
if (stderrWritten < OUTPUT_LIMIT) {
|
||||
const remaining = OUTPUT_LIMIT - stderrWritten;
|
||||
if (text.length <= remaining) {
|
||||
process.stderr.write(text);
|
||||
stderrWritten += text.length;
|
||||
} else {
|
||||
process.stderr.write(text.slice(0, remaining));
|
||||
stderrWritten += remaining;
|
||||
}
|
||||
} else if (!stderrTruncated) {
|
||||
process.stderr.write('\n...[stderr truncated]...\n');
|
||||
stderrTruncated = true;
|
||||
}
|
||||
});
|
||||
|
||||
child.on('close', (code) => {
|
||||
resolve({ code: code ?? 1, stdout, stderr });
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
async function readJson(filePath, fallback) {
|
||||
try {
|
||||
const raw = await fs.readFile(filePath, 'utf8');
|
||||
return JSON.parse(raw);
|
||||
} catch {
|
||||
return fallback;
|
||||
}
|
||||
}
|
||||
|
||||
function normalizeStringList(value, fallback = []) {
|
||||
if (!Array.isArray(value)) return fallback;
|
||||
return value.filter((item) => typeof item === 'string' && item.trim()).map((item) => item.trim());
|
||||
}
|
||||
|
||||
function parseNonNegativeInt(value, fallback) {
|
||||
const parsed = Number.parseInt(String(value ?? ''), 10);
|
||||
if (Number.isFinite(parsed) && parsed >= 0) {
|
||||
return parsed;
|
||||
}
|
||||
return fallback;
|
||||
}
|
||||
|
||||
function parseBooleanFlag(value, fallback = false) {
|
||||
if (value === undefined || value === null || value === '') return fallback;
|
||||
const normalized = String(value).trim().toLowerCase();
|
||||
if (['1', 'true', 'yes', 'on'].includes(normalized)) return true;
|
||||
if (['0', 'false', 'no', 'off'].includes(normalized)) return false;
|
||||
return fallback;
|
||||
}
|
||||
|
||||
function redactSensitive(text) {
|
||||
if (!text) return '';
|
||||
return String(text)
|
||||
.replace(/\b(BEARER\s+)[A-Za-z0-9._~+/=-]+/gi, '$1[REDACTED]')
|
||||
.replace(/\b(token|api[_-]?key|secret(?:[_-]?key)?|password|passwd|authorization)\s*[:=]\s*[^\s,;]+/gi, '$1=[REDACTED]')
|
||||
.replace(/\b(sk|pk|ghp|xoxb|xoxp)_[A-Za-z0-9_-]+\b/g, '[REDACTED]');
|
||||
}
|
||||
|
||||
function excerptText(text, maxChars = 600) {
|
||||
const clean = redactSensitive(String(text || '').trim());
|
||||
if (!clean) return '';
|
||||
if (clean.length <= maxChars) return clean;
|
||||
return `${clean.slice(0, maxChars)}…`;
|
||||
}
|
||||
|
||||
function categorizeFailureMetadata(failureCategory, extraMetadata = {}) {
|
||||
return {
|
||||
failure_category: failureCategory,
|
||||
@@ -227,26 +70,6 @@ function categorizeFailureMetadata(failureCategory, extraMetadata = {}) {
|
||||
};
|
||||
}
|
||||
|
||||
function formatDurationMs(ms) {
|
||||
if (!Number.isFinite(ms) || ms < 0) return 'unknown';
|
||||
const minutes = Math.round(ms / 60000);
|
||||
return `${minutes} minute(s)`;
|
||||
}
|
||||
|
||||
function buildLockBackendRemediation({ cadence, retryWindowMs, maxDeferrals, incidentSignalId = null }) {
|
||||
const steps = [
|
||||
'Likely backend/relay connectivity issue during lock acquisition.',
|
||||
`Retry window: ${formatDurationMs(retryWindowMs)} (max deferrals: ${maxDeferrals}).`,
|
||||
`Retry command: npm run scheduler:${cadence}`,
|
||||
`Run health check: npm run lock:health -- --cadence ${cadence}`,
|
||||
`Review incident runbook: ${LOCK_INCIDENT_LINK}`,
|
||||
];
|
||||
if (incidentSignalId) {
|
||||
steps.push(`Incident signal: ${incidentSignalId}`);
|
||||
}
|
||||
return `Recommended auto-remediation: ${steps.join(' ')}`;
|
||||
}
|
||||
|
||||
async function validatePromptFile(promptPath) {
|
||||
let content;
|
||||
try {
|
||||
@@ -274,47 +97,6 @@ async function validatePromptFile(promptPath) {
|
||||
return { ok: true };
|
||||
}
|
||||
|
||||
function classifyLockBackendError(outputText) {
|
||||
const text = String(outputText || '').toLowerCase();
|
||||
if (!text.trim()) return 'unknown_backend_error';
|
||||
|
||||
const errorCategoryMatch = text.match(/error_category=([a-z0-9_]+)/);
|
||||
if (errorCategoryMatch?.[1]) {
|
||||
return errorCategoryMatch[1];
|
||||
}
|
||||
|
||||
if ((text.includes('relay') || text.includes('query')) && text.includes('timeout')) {
|
||||
return 'relay_query_timeout';
|
||||
}
|
||||
|
||||
if (text.includes('publish failed to all relays') || text.includes('failed to publish to any relay')) {
|
||||
return 'relay_publish_quorum_failure';
|
||||
}
|
||||
|
||||
if (
|
||||
text.includes('connection refused')
|
||||
|| text.includes('econnrefused')
|
||||
|| text.includes('getaddrinfo')
|
||||
|| text.includes('enotfound')
|
||||
|| text.includes('eai_again')
|
||||
|| (text.includes('websocket') && text.includes('dns'))
|
||||
) {
|
||||
return 'websocket_connection_refused_or_dns';
|
||||
}
|
||||
|
||||
if (
|
||||
text.includes('invalid url')
|
||||
|| text.includes('malformed')
|
||||
|| text.includes('unsupported protocol')
|
||||
|| text.includes('must start with ws')
|
||||
|| text.includes('invalid relay')
|
||||
) {
|
||||
return 'malformed_relay_url_config';
|
||||
}
|
||||
|
||||
return 'unknown_backend_error';
|
||||
}
|
||||
|
||||
async function artifactExistsSince(filePath, sinceMs) {
|
||||
if (!filePath) return false;
|
||||
const resolved = path.isAbsolute(filePath)
|
||||
@@ -520,10 +302,6 @@ async function getSchedulerConfig(cadence, { isInteractive }) {
|
||||
};
|
||||
}
|
||||
|
||||
function getRunDateKey(now = new Date()) {
|
||||
return now.toISOString().slice(0, 10);
|
||||
}
|
||||
|
||||
async function readRunState(cadence) {
|
||||
const statePath = path.resolve(process.cwd(), 'task-logs', cadence, '.scheduler-run-state.json');
|
||||
const fallback = { run_date: getRunDateKey(), lock_deferral: null };
|
||||
@@ -554,122 +332,6 @@ function createIdempotencyKey({ cadence, selectedAgent, runDate }) {
|
||||
return `${cadence}:${selectedAgent}:${runDate}:${randomUUID()}`;
|
||||
}
|
||||
|
||||
async function runLockHealthPreflight({ cadence, platform }) {
|
||||
const preflight = await runCommand('npm', ['run', 'lock:health', '--', '--cadence', cadence], {
|
||||
env: { AGENT_PLATFORM: platform },
|
||||
});
|
||||
const payload = parseJsonFromOutput(`${preflight.stdout}\n${preflight.stderr}`) || {};
|
||||
const relayList = Array.isArray(payload.relays)
|
||||
? payload.relays.filter((relay) => typeof relay === 'string' && relay.trim()).map((relay) => relay.trim())
|
||||
: [];
|
||||
const combinedOutput = `${preflight.stderr}\n${preflight.stdout}`;
|
||||
const failureCategory = payload.failureCategory || classifyLockBackendError(combinedOutput);
|
||||
|
||||
return {
|
||||
code: preflight.code,
|
||||
payload,
|
||||
relayList,
|
||||
failureCategory,
|
||||
stderrExcerpt: excerptText(preflight.stderr),
|
||||
stdoutExcerpt: excerptText(preflight.stdout),
|
||||
};
|
||||
}
|
||||
|
||||
function sleep(ms) {
|
||||
return new Promise((resolve) => setTimeout(resolve, ms));
|
||||
}
|
||||
|
||||
async function acquireLockWithRetry({ selectedAgent, cadence, platform, model, lockRetry, idempotencyKey }) {
|
||||
const lockCommandArgs = ['run', 'lock:lock', '--', '--agent', selectedAgent, '--cadence', cadence];
|
||||
if (model) {
|
||||
lockCommandArgs.push('--model', model);
|
||||
}
|
||||
const backoffScheduleMs = [];
|
||||
let attempts = 0;
|
||||
const correlationId = idempotencyKey || `${cadence}:${selectedAgent}:${getRunDateKey()}`;
|
||||
|
||||
while (true) {
|
||||
attempts += 1;
|
||||
const result = await runCommand('npm', lockCommandArgs, {
|
||||
env: {
|
||||
AGENT_PLATFORM: platform,
|
||||
...(model ? { AGENT_MODEL: model } : {}),
|
||||
...(idempotencyKey ? { SCHEDULER_LOCK_IDEMPOTENCY_KEY: idempotencyKey } : {}),
|
||||
SCHEDULER_LOCK_CORRELATION_ID: correlationId,
|
||||
SCHEDULER_LOCK_ATTEMPT_ID: String(attempts),
|
||||
},
|
||||
});
|
||||
|
||||
if (result.code !== 2) {
|
||||
return { result, attempts, backoffScheduleMs, correlationId };
|
||||
}
|
||||
|
||||
if (attempts > lockRetry.maxRetries) {
|
||||
return {
|
||||
result,
|
||||
attempts,
|
||||
backoffScheduleMs,
|
||||
correlationId,
|
||||
finalBackendCategory: classifyLockBackendError(`${result.stderr}\n${result.stdout}`),
|
||||
};
|
||||
}
|
||||
|
||||
const exponentialBase = lockRetry.backoffMs * (2 ** (attempts - 1));
|
||||
const jitter = lockRetry.jitterMs > 0
|
||||
? Math.floor(Math.random() * (lockRetry.jitterMs + 1))
|
||||
: 0;
|
||||
const delayMs = exponentialBase + jitter;
|
||||
backoffScheduleMs.push(delayMs);
|
||||
|
||||
console.log(JSON.stringify({
|
||||
event: 'scheduler.lock.retry',
|
||||
attempt: attempts,
|
||||
max_retries: lockRetry.maxRetries,
|
||||
delay_ms: delayMs,
|
||||
correlation_id: correlationId,
|
||||
selected_agent: selectedAgent,
|
||||
cadence,
|
||||
}));
|
||||
|
||||
if (delayMs > 0) {
|
||||
await sleep(delayMs);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
function summarizeLockFailureReasons(outputText) {
|
||||
const events = parseJsonEventsFromOutput(outputText);
|
||||
const quorumFailure = events.findLast((entry) => entry.event === 'lock_publish_quorum_failed');
|
||||
if (quorumFailure && quorumFailure.reasonDistribution && typeof quorumFailure.reasonDistribution === 'object') {
|
||||
return {
|
||||
reasonDistribution: quorumFailure.reasonDistribution,
|
||||
attemptId: quorumFailure.attemptId || null,
|
||||
correlationId: quorumFailure.correlationId || null,
|
||||
totalElapsedMs: quorumFailure.totalElapsedMs ?? null,
|
||||
};
|
||||
}
|
||||
|
||||
const distribution = {};
|
||||
for (const entry of events) {
|
||||
if (entry.event !== 'lock_publish_failure') continue;
|
||||
const reason = typeof entry.reason === 'string' && entry.reason.trim()
|
||||
? entry.reason.trim()
|
||||
: 'unknown';
|
||||
distribution[reason] = (distribution[reason] || 0) + 1;
|
||||
}
|
||||
return {
|
||||
reasonDistribution: distribution,
|
||||
attemptId: null,
|
||||
correlationId: null,
|
||||
totalElapsedMs: null,
|
||||
};
|
||||
}
|
||||
|
||||
function toYamlScalar(value) {
|
||||
const str = String(value ?? '');
|
||||
return `'${str.replace(/'/g, "''")}'`;
|
||||
}
|
||||
|
||||
async function writeLog({ cadence, agent, status, reason, detail, platform, metadata = {} }) {
|
||||
const logDir = path.resolve(process.cwd(), 'task-logs', cadence);
|
||||
await fs.mkdir(logDir, { recursive: true });
|
||||
|
||||
@@ -0,0 +1,173 @@
|
||||
import {
|
||||
runCommand,
|
||||
parseJsonFromOutput,
|
||||
parseJsonEventsFromOutput,
|
||||
excerptText,
|
||||
formatDurationMs,
|
||||
getRunDateKey,
|
||||
sleep,
|
||||
} from './scheduler-utils.mjs';
|
||||
|
||||
const LOCK_INCIDENT_LINK = 'docs/agent-handoffs/learnings/2026-02-15-relay-health-preflight-job.md';
|
||||
|
||||
export function classifyLockBackendError(outputText) {
|
||||
const text = String(outputText || '').toLowerCase();
|
||||
if (!text.trim()) return 'unknown_backend_error';
|
||||
|
||||
const errorCategoryMatch = text.match(/error_category=([a-z0-9_]+)/);
|
||||
if (errorCategoryMatch?.[1]) {
|
||||
return errorCategoryMatch[1];
|
||||
}
|
||||
|
||||
if ((text.includes('relay') || text.includes('query')) && text.includes('timeout')) {
|
||||
return 'relay_query_timeout';
|
||||
}
|
||||
|
||||
if (text.includes('publish failed to all relays') || text.includes('failed to publish to any relay')) {
|
||||
return 'relay_publish_quorum_failure';
|
||||
}
|
||||
|
||||
if (
|
||||
text.includes('connection refused')
|
||||
|| text.includes('econnrefused')
|
||||
|| text.includes('getaddrinfo')
|
||||
|| text.includes('enotfound')
|
||||
|| text.includes('eai_again')
|
||||
|| (text.includes('websocket') && text.includes('dns'))
|
||||
) {
|
||||
return 'websocket_connection_refused_or_dns';
|
||||
}
|
||||
|
||||
if (
|
||||
text.includes('invalid url')
|
||||
|| text.includes('malformed')
|
||||
|| text.includes('unsupported protocol')
|
||||
|| text.includes('must start with ws')
|
||||
|| text.includes('invalid relay')
|
||||
) {
|
||||
return 'malformed_relay_url_config';
|
||||
}
|
||||
|
||||
return 'unknown_backend_error';
|
||||
}
|
||||
|
||||
export function buildLockBackendRemediation({ cadence, retryWindowMs, maxDeferrals, incidentSignalId = null }) {
|
||||
const steps = [
|
||||
'Likely backend/relay connectivity issue during lock acquisition.',
|
||||
`Retry window: ${formatDurationMs(retryWindowMs)} (max deferrals: ${maxDeferrals}).`,
|
||||
`Retry command: npm run scheduler:${cadence}`,
|
||||
`Run health check: npm run lock:health -- --cadence ${cadence}`,
|
||||
`Review incident runbook: ${LOCK_INCIDENT_LINK}`,
|
||||
];
|
||||
if (incidentSignalId) {
|
||||
steps.push(`Incident signal: ${incidentSignalId}`);
|
||||
}
|
||||
return `Recommended auto-remediation: ${steps.join(' ')}`;
|
||||
}
|
||||
|
||||
export async function runLockHealthPreflight({ cadence, platform }) {
|
||||
const preflight = await runCommand('npm', ['run', 'lock:health', '--', '--cadence', cadence], {
|
||||
env: { AGENT_PLATFORM: platform },
|
||||
});
|
||||
const payload = parseJsonFromOutput(`${preflight.stdout}\n${preflight.stderr}`) || {};
|
||||
const relayList = Array.isArray(payload.relays)
|
||||
? payload.relays.filter((relay) => typeof relay === 'string' && relay.trim()).map((relay) => relay.trim())
|
||||
: [];
|
||||
const combinedOutput = `${preflight.stderr}\n${preflight.stdout}`;
|
||||
const failureCategory = payload.failureCategory || classifyLockBackendError(combinedOutput);
|
||||
|
||||
return {
|
||||
code: preflight.code,
|
||||
payload,
|
||||
relayList,
|
||||
failureCategory,
|
||||
stderrExcerpt: excerptText(preflight.stderr),
|
||||
stdoutExcerpt: excerptText(preflight.stdout),
|
||||
};
|
||||
}
|
||||
|
||||
export async function acquireLockWithRetry({ selectedAgent, cadence, platform, model, lockRetry, idempotencyKey }) {
|
||||
const lockCommandArgs = ['run', 'lock:lock', '--', '--agent', selectedAgent, '--cadence', cadence];
|
||||
if (model) {
|
||||
lockCommandArgs.push('--model', model);
|
||||
}
|
||||
const backoffScheduleMs = [];
|
||||
let attempts = 0;
|
||||
const correlationId = idempotencyKey || `${cadence}:${selectedAgent}:${getRunDateKey()}`;
|
||||
|
||||
while (true) {
|
||||
attempts += 1;
|
||||
const result = await runCommand('npm', lockCommandArgs, {
|
||||
env: {
|
||||
AGENT_PLATFORM: platform,
|
||||
...(model ? { AGENT_MODEL: model } : {}),
|
||||
...(idempotencyKey ? { SCHEDULER_LOCK_IDEMPOTENCY_KEY: idempotencyKey } : {}),
|
||||
SCHEDULER_LOCK_CORRELATION_ID: correlationId,
|
||||
SCHEDULER_LOCK_ATTEMPT_ID: String(attempts),
|
||||
},
|
||||
});
|
||||
|
||||
if (result.code !== 2) {
|
||||
return { result, attempts, backoffScheduleMs, correlationId };
|
||||
}
|
||||
|
||||
if (attempts > lockRetry.maxRetries) {
|
||||
return {
|
||||
result,
|
||||
attempts,
|
||||
backoffScheduleMs,
|
||||
correlationId,
|
||||
finalBackendCategory: classifyLockBackendError(`${result.stderr}\n${result.stdout}`),
|
||||
};
|
||||
}
|
||||
|
||||
const exponentialBase = lockRetry.backoffMs * (2 ** (attempts - 1));
|
||||
const jitter = lockRetry.jitterMs > 0
|
||||
? Math.floor(Math.random() * (lockRetry.jitterMs + 1))
|
||||
: 0;
|
||||
const delayMs = exponentialBase + jitter;
|
||||
backoffScheduleMs.push(delayMs);
|
||||
|
||||
console.log(JSON.stringify({
|
||||
event: 'scheduler.lock.retry',
|
||||
attempt: attempts,
|
||||
max_retries: lockRetry.maxRetries,
|
||||
delay_ms: delayMs,
|
||||
correlation_id: correlationId,
|
||||
selected_agent: selectedAgent,
|
||||
cadence,
|
||||
}));
|
||||
|
||||
if (delayMs > 0) {
|
||||
await sleep(delayMs);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
export function summarizeLockFailureReasons(outputText) {
|
||||
const events = parseJsonEventsFromOutput(outputText);
|
||||
const quorumFailure = events.findLast((entry) => entry.event === 'lock_publish_quorum_failed');
|
||||
if (quorumFailure && quorumFailure.reasonDistribution && typeof quorumFailure.reasonDistribution === 'object') {
|
||||
return {
|
||||
reasonDistribution: quorumFailure.reasonDistribution,
|
||||
attemptId: quorumFailure.attemptId || null,
|
||||
correlationId: quorumFailure.correlationId || null,
|
||||
totalElapsedMs: quorumFailure.totalElapsedMs ?? null,
|
||||
};
|
||||
}
|
||||
|
||||
const distribution = {};
|
||||
for (const entry of events) {
|
||||
if (entry.event !== 'lock_publish_failure') continue;
|
||||
const reason = typeof entry.reason === 'string' && entry.reason.trim()
|
||||
? entry.reason.trim()
|
||||
: 'unknown';
|
||||
distribution[reason] = (distribution[reason] || 0) + 1;
|
||||
}
|
||||
return {
|
||||
reasonDistribution: distribution,
|
||||
attemptId: null,
|
||||
correlationId: null,
|
||||
totalElapsedMs: null,
|
||||
};
|
||||
}
|
||||
@@ -0,0 +1,195 @@
|
||||
import fs from 'node:fs/promises';
|
||||
import { spawn } from 'node:child_process';
|
||||
|
||||
const ALLOWED_ENV_KEYS = new Set([
|
||||
'PATH', 'Path', // Windows compatibility
|
||||
'HOME', 'USER', 'SHELL', 'TERM', 'LANG', 'LC_ALL', 'LC_CTYPE', 'TZ',
|
||||
'NODE_ENV', 'NODE_OPTIONS', 'NODE_PATH',
|
||||
'APPDATA', 'LOCALAPPDATA', 'PROGRAMDATA', 'SystemRoot', 'windir', 'ComSpec', 'PATHEXT',
|
||||
'TMPDIR', 'TEMP', 'TMP',
|
||||
'EDITOR', 'VISUAL',
|
||||
// CI variables
|
||||
'CI', 'GITHUB_ACTIONS', 'GITHUB_REF', 'GITHUB_HEAD_REF', 'GITHUB_BASE_REF', 'GITHUB_EVENT_NAME', 'GITHUB_SHA',
|
||||
]);
|
||||
|
||||
export function getSafeEnv() {
|
||||
const safeEnv = {};
|
||||
for (const key in process.env) {
|
||||
if (ALLOWED_ENV_KEYS.has(key) ||
|
||||
key.startsWith('npm_') ||
|
||||
key.startsWith('NOSTR_') ||
|
||||
key.startsWith('TORCH_') ||
|
||||
key.startsWith('SCHEDULER_') ||
|
||||
key.startsWith('AGENT_')) {
|
||||
safeEnv[key] = process.env[key];
|
||||
}
|
||||
}
|
||||
return safeEnv;
|
||||
}
|
||||
|
||||
/**
|
||||
* Spawns a child process with a sanitized environment by default.
|
||||
* @param {string} command
|
||||
* @param {string[]} args
|
||||
* @param {object} options
|
||||
* @param {object} [options.env] - Additional environment variables to merge.
|
||||
* @param {boolean} [options.inheritProcessEnv=false] - If true, use full process.env. If false (default), use sanitized env.
|
||||
*/
|
||||
export async function runCommand(command, args = [], options = {}) {
|
||||
const baseEnv = options.inheritProcessEnv ? process.env : getSafeEnv();
|
||||
return new Promise((resolve) => {
|
||||
const child = spawn(command, args, {
|
||||
cwd: process.cwd(),
|
||||
env: { ...baseEnv, ...(options.env || {}) },
|
||||
stdio: ['ignore', 'pipe', 'pipe'],
|
||||
});
|
||||
|
||||
let stdout = '';
|
||||
let stderr = '';
|
||||
|
||||
const OUTPUT_LIMIT = 20000; // 20KB
|
||||
let stdoutWritten = 0;
|
||||
let stderrWritten = 0;
|
||||
let stdoutTruncated = false;
|
||||
let stderrTruncated = false;
|
||||
|
||||
child.stdout.on('data', (chunk) => {
|
||||
const text = chunk.toString();
|
||||
stdout += text;
|
||||
|
||||
if (stdoutWritten < OUTPUT_LIMIT) {
|
||||
const remaining = OUTPUT_LIMIT - stdoutWritten;
|
||||
if (text.length <= remaining) {
|
||||
process.stdout.write(text);
|
||||
stdoutWritten += text.length;
|
||||
} else {
|
||||
process.stdout.write(text.slice(0, remaining));
|
||||
stdoutWritten += remaining;
|
||||
}
|
||||
} else if (!stdoutTruncated) {
|
||||
process.stdout.write('\n...[stdout truncated]...\n');
|
||||
stdoutTruncated = true;
|
||||
}
|
||||
});
|
||||
|
||||
child.stderr.on('data', (chunk) => {
|
||||
const text = chunk.toString();
|
||||
stderr += text;
|
||||
|
||||
if (stderrWritten < OUTPUT_LIMIT) {
|
||||
const remaining = OUTPUT_LIMIT - stderrWritten;
|
||||
if (text.length <= remaining) {
|
||||
process.stderr.write(text);
|
||||
stderrWritten += text.length;
|
||||
} else {
|
||||
process.stderr.write(text.slice(0, remaining));
|
||||
stderrWritten += remaining;
|
||||
}
|
||||
} else if (!stderrTruncated) {
|
||||
process.stderr.write('\n...[stderr truncated]...\n');
|
||||
stderrTruncated = true;
|
||||
}
|
||||
});
|
||||
|
||||
child.on('close', (code) => {
|
||||
resolve({ code: code ?? 1, stdout, stderr });
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
export function parseJsonFromOutput(text) {
|
||||
const trimmed = String(text || '').trim();
|
||||
if (!trimmed) return null;
|
||||
const lines = trimmed.split(/\r?\n/).map((line) => line.trim()).filter(Boolean);
|
||||
for (let i = lines.length - 1; i >= 0; i -= 1) {
|
||||
const line = lines[i];
|
||||
if (!line.startsWith('{') && !line.startsWith('[')) continue;
|
||||
try {
|
||||
return JSON.parse(line);
|
||||
} catch {
|
||||
// Keep scanning from the end.
|
||||
}
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
export function parseJsonEventsFromOutput(text) {
|
||||
const events = [];
|
||||
for (const line of String(text || '').split(/\r?\n/)) {
|
||||
const trimmed = line.trim();
|
||||
if (!trimmed.startsWith('{')) continue;
|
||||
try {
|
||||
const parsed = JSON.parse(trimmed);
|
||||
if (parsed && typeof parsed === 'object' && !Array.isArray(parsed)) {
|
||||
events.push(parsed);
|
||||
}
|
||||
} catch {
|
||||
// Ignore non-JSON lines.
|
||||
}
|
||||
}
|
||||
return events;
|
||||
}
|
||||
|
||||
export async function readJson(filePath, fallback) {
|
||||
try {
|
||||
const raw = await fs.readFile(filePath, 'utf8');
|
||||
return JSON.parse(raw);
|
||||
} catch {
|
||||
return fallback;
|
||||
}
|
||||
}
|
||||
|
||||
export function normalizeStringList(value, fallback = []) {
|
||||
if (!Array.isArray(value)) return fallback;
|
||||
return value.filter((item) => typeof item === 'string' && item.trim()).map((item) => item.trim());
|
||||
}
|
||||
|
||||
export function parseNonNegativeInt(value, fallback) {
|
||||
const parsed = Number.parseInt(String(value ?? ''), 10);
|
||||
if (Number.isFinite(parsed) && parsed >= 0) {
|
||||
return parsed;
|
||||
}
|
||||
return fallback;
|
||||
}
|
||||
|
||||
export function parseBooleanFlag(value, fallback = false) {
|
||||
if (value === undefined || value === null || value === '') return fallback;
|
||||
const normalized = String(value).trim().toLowerCase();
|
||||
if (['1', 'true', 'yes', 'on'].includes(normalized)) return true;
|
||||
if (['0', 'false', 'no', 'off'].includes(normalized)) return false;
|
||||
return fallback;
|
||||
}
|
||||
|
||||
export function redactSensitive(text) {
|
||||
if (!text) return '';
|
||||
return String(text)
|
||||
.replace(/\b(BEARER\s+)[A-Za-z0-9._~+/=-]+/gi, '$1[REDACTED]')
|
||||
.replace(/\b(token|api[_-]?key|secret(?:[_-]?key)?|password|passwd|authorization)\s*[:=]\s*[^\s,;]+/gi, '$1=[REDACTED]')
|
||||
.replace(/\b(sk|pk|ghp|xoxb|xoxp)_[A-Za-z0-9_-]+\b/g, '[REDACTED]');
|
||||
}
|
||||
|
||||
export function excerptText(text, maxChars = 600) {
|
||||
const clean = redactSensitive(String(text || '').trim());
|
||||
if (!clean) return '';
|
||||
if (clean.length <= maxChars) return clean;
|
||||
return `${clean.slice(0, maxChars)}…`;
|
||||
}
|
||||
|
||||
export function formatDurationMs(ms) {
|
||||
if (!Number.isFinite(ms) || ms < 0) return 'unknown';
|
||||
const minutes = Math.round(ms / 60000);
|
||||
return `${minutes} minute(s)`;
|
||||
}
|
||||
|
||||
export function getRunDateKey(now = new Date()) {
|
||||
return now.toISOString().slice(0, 10);
|
||||
}
|
||||
|
||||
export function toYamlScalar(value) {
|
||||
const str = String(value ?? '');
|
||||
return `'${str.replace(/'/g, "''")}'`;
|
||||
}
|
||||
|
||||
export function sleep(ms) {
|
||||
return new Promise((resolve) => setTimeout(resolve, ms));
|
||||
}
|
||||
Reference in new issue
Block a user