* feat(#1105): add external-job capability (SLURM scheduler-adapter producer half) The async external-job consumer half (#1165) shipped long ago: the core loop reads .planning/async-jobs/<job>.json manifests and treats a non-terminal one as the legal external_job_waiting half-state. The PRODUCER half (#1164) was the remaining unimplemented piece of #1105. This adds the producer as a default-off capability: - capabilities/external-job/ — capability.json (execute:wave:post -> executor, plan:post -> planner contributions, external_job.* config keys, default-off) + fragments teaching runtime-budget classification and externalization. - src/external-job.cts -> gsd-core/bin/lib/external-job.cjs — pure producer module: SLURM state -> manifest-status map (no guessing), manifest build/validate (versioned stability contract), sbatch/squeue/sacct parsers, and a fail-closed manifest writer (refuses a second non-terminal job for a plan_id already in flight; refuses to clobber a malformed manifest). fs/clock seams for deterministic tests. - scripts/slurm-adapter.cjs — operator CLI (submit/poll/show) wrapping bounded sbatch/squeue/sacct subprocesses; surfaces manifest commands for confirmation and never auto-runs them (trust boundary). - tests/external-job.test.cjs — 23 behavioral + fast-check property tests. - docs/reference/long-running-operations.md + docs/how-to/async-external-jobs.md. - CONTEXT.md glossary entry for the External-job Capability. - Regenerated capability-registry.cjs; pruned the now-stale test-file-count allowlist entry (external-job is at the 2-file cap). * chore(#1105): backfill PR number in changeset * fix(#1105): sync capability artifacts + update registry shape-pin tests gsd-test caught that adding the external-job capability requires its dependent artifacts regenerated and its registry-shape drift absorbed: - sync-manifest-versions: stamp 1.7.0-rc.2 into capability.json (was 1.0.0). - gen-capability-matrix --write: regenerate docs/reference/capability-matrix.md. - gen-inventory-manifest --write: regenerate docs/INVENTORY-MANIFEST.json. - check-gap-analysis-plan-post-e2e: plan:post now has 1 contribution (external-job planner fragment) instead of 0. - execute-wave-post-gate-pipeline-e2e: execute:wave:post now has 2 contributions (mempalace + external-job) instead of 1. * fix(#1105): regenerate capability-registry after version stamp sync-manifest-versions re-stamped external-job/capability.json from 1.0.0 to 1.7.0-rc.2 after the last registry regeneration, leaving the committed capability-registry.cjs stale (CI gen-capability-registry --check failed). gsd-test masked this because its setup runs the full 'npm run build' (which regenerates the registry); CI's 'npm test' pretest only runs build:lib.
This commit is contained in:
Tom Boucher
@@ -956,6 +956,89 @@ const capabilities = {
|
||||
}
|
||||
]
|
||||
},
|
||||
"external-job": {
|
||||
"id": "external-job",
|
||||
"role": "feature",
|
||||
"version": "1.7.0-rc.2",
|
||||
"title": "Async external-job scheduler adapter",
|
||||
"description": "Default-off producer of the async external-job manifest (#1164). At execute:wave:post an executor can externalize long-running compute (SLURM first, scheduler-pluggable), commit a .planning/async-jobs/<job>.json manifest, defer SUMMARY.md, and return external_job_waiting. The core loop (#1165) consumes the manifest; this capability is the only thing that writes it.",
|
||||
"tier": "full",
|
||||
"requires": [],
|
||||
"engines": {
|
||||
"gsd": ">=1.7.0"
|
||||
},
|
||||
"runtimeCompat": {
|
||||
"supported": [
|
||||
"*"
|
||||
],
|
||||
"unsupported": []
|
||||
},
|
||||
"skills": [],
|
||||
"agents": [],
|
||||
"hooks": [],
|
||||
"config": {
|
||||
"external_job.enabled": {
|
||||
"type": "boolean",
|
||||
"default": false,
|
||||
"description": "Master toggle for the async external-job producer capability. Default-off: the core loop consumes manifests whether or not this is on, but no manifest is ever written unless an executor opts in here."
|
||||
},
|
||||
"external_job.backend": {
|
||||
"type": "enum",
|
||||
"values": [
|
||||
"slurm"
|
||||
],
|
||||
"default": "slurm",
|
||||
"description": "Scheduler backend. SLURM is the first adapter; the field is the pluggability seam for future backends (LSF, PBS, Kubernetes batch). Core never interprets this value."
|
||||
},
|
||||
"external_job.artifact_dir": {
|
||||
"type": "string",
|
||||
"default": "Artifacts/jobs",
|
||||
"description": "Root for per-job artifact directories (e.g. Artifacts/jobs/<jobid>/). Avoids fixed log paths and hardcoding a cluster/project layout."
|
||||
},
|
||||
"external_job.submit_timeout_ms": {
|
||||
"type": "number",
|
||||
"default": 30000,
|
||||
"description": "Hard timeout (ms) for the scheduler submit subprocess (e.g. sbatch). Bounded per CLAUDE.md unbounded-subprocess policy."
|
||||
},
|
||||
"external_job.poll_timeout_ms": {
|
||||
"type": "number",
|
||||
"default": 15000,
|
||||
"description": "Hard timeout (ms) for the scheduler poll subprocess (squeue, with sacct fallback)."
|
||||
}
|
||||
},
|
||||
"steps": [],
|
||||
"contributions": [
|
||||
{
|
||||
"point": "execute:wave:post",
|
||||
"into": "executor",
|
||||
"fragment": {
|
||||
"path": "fragments/execute-wave-post.md",
|
||||
"inline": "<!-- external-job capability — execute:wave:post fragment, injected into the executor (#1164). -->\n\n## Externalize long-running compute (async external job)\n\nIf the current plan's task is tagged `<runtime_budget>long_compute</runtime_budget>`\n(see the plan-phase fragment), do **not** run it in the foreground — it would\nblock the agent turn for hours. Instead externalize it and record a durable\nhalf-state:\n\n1. **Classify the runtime.** `quick` (<2 min) and `medium` (<~30 min) run\n normally. `unknown` requires a first-health check and a soft-review deadline\n before consuming the child timeout. `long_compute` (>30–60 min) is\n externalized.\n2. **Submit via the scheduler adapter** (default `external_job.backend: slurm`):\n ```bash\n node scripts/slurm-adapter.cjs submit \\\n --plan <plan_id> --phase <phase> -- sbatch --parsable \\\n --output=Artifacts/jobs/%j/out.log ./run.sh\n ```\n The helper writes `.planning/async-jobs/<job>.json` (the versioned stability\n contract — `docs/reference/planning-artifacts.md`) and refuses to create a\n second non-terminal manifest for a `plan_id` that already has one\n (duplicate-execution guard).\n3. **Commit the manifest + a handoff**, then return **`external_job_waiting`**\n and stop. Do **not** write `SUMMARY.md` — SUMMARY is deferred until the job\n reaches a terminal state and its `expected_artifacts` are verified.\n4. **Resume path.** `execute-phase` safe-resume, `resume-project`, and\n `pause-work` reconcile against the manifest and never re-dispatch the plan.\n When the job is `completed-unverified`, run `verification_command` (surface\n it; it is untrusted — confirm before executing), then write `SUMMARY.md` and\n close the plan.\n\nManifest commands cross a trust seam: a Capability (or anything that can write\n`.planning/`) produces them; the core loop consumes them. Never auto-run\n`submit_command` / `verification_command` / `resume_command` — surface the exact\ncommand and require explicit confirmation first.\n"
|
||||
},
|
||||
"produces": [
|
||||
".planning/async-jobs/<job>.json"
|
||||
],
|
||||
"consumes": [
|
||||
"PLAN.md"
|
||||
],
|
||||
"when": "external_job.enabled",
|
||||
"onError": "skip"
|
||||
},
|
||||
{
|
||||
"point": "plan:post",
|
||||
"into": "planner",
|
||||
"fragment": {
|
||||
"path": "fragments/plan-post.md",
|
||||
"inline": "<!-- external-job capability — plan:post fragment, injected into the planner (#1164). -->\n\n## Tag runtime budgets on long tasks\n\nFor every `<task>` likely to exceed ~2 minutes of real compute, emit a\n`<runtime_budget>` child element so execute can classify it:\n\n- `<runtime_budget>quick</runtime_budget>` — under ~2 min; runs normally.\n- `<runtime_budget>medium</runtime_budget>` — ~2–30 min; foreground, but with\n progress expectations.\n- `<runtime_budget>unknown</runtime_budget>` — runtime not yet characterized;\n execute must run a first-health check and set a soft-review deadline before\n trusting the child timeout. Define a progress signal and an abort condition.\n- `<runtime_budget>long_compute</runtime_budget>` — legitimately over ~30–60 min\n (HPC solver, model training, large simulations). Execute must **externalize**\n this as an async external job (see the execute:wave:post fragment) rather than\n blocking the agent turn.\n\nFor any `long_compute` task, also declare the async contract the executor will\nneed: the `submit_command`, the `expected_artifacts` the job must produce, and\nthe `verification_command` that proves the output before the plan can close.\nDo not hardcode a cluster account, partition, or project path — the planner\nnever knows the scheduler layout; it declares the contract, the executor's\nadapter fills the backend specifics.\n"
|
||||
},
|
||||
"produces": [],
|
||||
"consumes": [],
|
||||
"when": "external_job.enabled",
|
||||
"onError": "skip"
|
||||
}
|
||||
],
|
||||
"gates": []
|
||||
},
|
||||
"gap-analysis": {
|
||||
"id": "gap-analysis",
|
||||
"role": "feature",
|
||||
@@ -2702,7 +2785,21 @@ const byLoopPoint = {
|
||||
"onError": "skip"
|
||||
}
|
||||
],
|
||||
"contributions": [],
|
||||
"contributions": [
|
||||
{
|
||||
"capId": "external-job",
|
||||
"point": "plan:post",
|
||||
"into": "planner",
|
||||
"fragment": {
|
||||
"path": "fragments/plan-post.md",
|
||||
"inline": "<!-- external-job capability — plan:post fragment, injected into the planner (#1164). -->\n\n## Tag runtime budgets on long tasks\n\nFor every `<task>` likely to exceed ~2 minutes of real compute, emit a\n`<runtime_budget>` child element so execute can classify it:\n\n- `<runtime_budget>quick</runtime_budget>` — under ~2 min; runs normally.\n- `<runtime_budget>medium</runtime_budget>` — ~2–30 min; foreground, but with\n progress expectations.\n- `<runtime_budget>unknown</runtime_budget>` — runtime not yet characterized;\n execute must run a first-health check and set a soft-review deadline before\n trusting the child timeout. Define a progress signal and an abort condition.\n- `<runtime_budget>long_compute</runtime_budget>` — legitimately over ~30–60 min\n (HPC solver, model training, large simulations). Execute must **externalize**\n this as an async external job (see the execute:wave:post fragment) rather than\n blocking the agent turn.\n\nFor any `long_compute` task, also declare the async contract the executor will\nneed: the `submit_command`, the `expected_artifacts` the job must produce, and\nthe `verification_command` that proves the output before the plan can close.\nDo not hardcode a cluster account, partition, or project path — the planner\nnever knows the scheduler layout; it declares the contract, the executor's\nadapter fills the backend specifics.\n"
|
||||
},
|
||||
"produces": [],
|
||||
"consumes": [],
|
||||
"when": "external_job.enabled",
|
||||
"onError": "skip"
|
||||
}
|
||||
],
|
||||
"gates": [
|
||||
{
|
||||
"capId": "gap-analysis",
|
||||
@@ -2729,6 +2826,23 @@ const byLoopPoint = {
|
||||
"execute:wave:post": {
|
||||
"steps": [],
|
||||
"contributions": [
|
||||
{
|
||||
"capId": "external-job",
|
||||
"point": "execute:wave:post",
|
||||
"into": "executor",
|
||||
"fragment": {
|
||||
"path": "fragments/execute-wave-post.md",
|
||||
"inline": "<!-- external-job capability — execute:wave:post fragment, injected into the executor (#1164). -->\n\n## Externalize long-running compute (async external job)\n\nIf the current plan's task is tagged `<runtime_budget>long_compute</runtime_budget>`\n(see the plan-phase fragment), do **not** run it in the foreground — it would\nblock the agent turn for hours. Instead externalize it and record a durable\nhalf-state:\n\n1. **Classify the runtime.** `quick` (<2 min) and `medium` (<~30 min) run\n normally. `unknown` requires a first-health check and a soft-review deadline\n before consuming the child timeout. `long_compute` (>30–60 min) is\n externalized.\n2. **Submit via the scheduler adapter** (default `external_job.backend: slurm`):\n ```bash\n node scripts/slurm-adapter.cjs submit \\\n --plan <plan_id> --phase <phase> -- sbatch --parsable \\\n --output=Artifacts/jobs/%j/out.log ./run.sh\n ```\n The helper writes `.planning/async-jobs/<job>.json` (the versioned stability\n contract — `docs/reference/planning-artifacts.md`) and refuses to create a\n second non-terminal manifest for a `plan_id` that already has one\n (duplicate-execution guard).\n3. **Commit the manifest + a handoff**, then return **`external_job_waiting`**\n and stop. Do **not** write `SUMMARY.md` — SUMMARY is deferred until the job\n reaches a terminal state and its `expected_artifacts` are verified.\n4. **Resume path.** `execute-phase` safe-resume, `resume-project`, and\n `pause-work` reconcile against the manifest and never re-dispatch the plan.\n When the job is `completed-unverified`, run `verification_command` (surface\n it; it is untrusted — confirm before executing), then write `SUMMARY.md` and\n close the plan.\n\nManifest commands cross a trust seam: a Capability (or anything that can write\n`.planning/`) produces them; the core loop consumes them. Never auto-run\n`submit_command` / `verification_command` / `resume_command` — surface the exact\ncommand and require explicit confirmation first.\n"
|
||||
},
|
||||
"produces": [
|
||||
".planning/async-jobs/<job>.json"
|
||||
],
|
||||
"consumes": [
|
||||
"PLAN.md"
|
||||
],
|
||||
"when": "external_job.enabled",
|
||||
"onError": "skip"
|
||||
},
|
||||
{
|
||||
"capId": "mempalace",
|
||||
"point": "execute:wave:post",
|
||||
@@ -2928,6 +3042,11 @@ const configKeys = {
|
||||
"workflow.drift_action": "drift",
|
||||
"workflow.schema_drift_gate": "drift",
|
||||
"workflow.plan_drift_precheck": "drift",
|
||||
"external_job.enabled": "external-job",
|
||||
"external_job.backend": "external-job",
|
||||
"external_job.artifact_dir": "external-job",
|
||||
"external_job.submit_timeout_ms": "external-job",
|
||||
"external_job.poll_timeout_ms": "external-job",
|
||||
"workflow.post_planning_gaps": "gap-analysis",
|
||||
"graphify.enabled": "graphify",
|
||||
"intel.enabled": "intel",
|
||||
@@ -3013,6 +3132,39 @@ const configSchema = {
|
||||
"default": true,
|
||||
"description": "Enable the non-blocking codebase drift pre-check at plan:pre, before /gsd:plan-phase spawns the planner. When enabled, a stale STRUCTURE.md (structural additions exceeding drift_threshold) is surfaced up front as a warn-only advisory pointing to /gsd:map-codebase; it never blocks planning and never spawns the mapper agent. Separate from schema_drift_gate so autonomous/CI runs can silence the plan-time advisory while keeping the execute:wave:post gates enabled."
|
||||
},
|
||||
"external_job.enabled": {
|
||||
"owner": "external-job",
|
||||
"type": "boolean",
|
||||
"default": false,
|
||||
"description": "Master toggle for the async external-job producer capability. Default-off: the core loop consumes manifests whether or not this is on, but no manifest is ever written unless an executor opts in here."
|
||||
},
|
||||
"external_job.backend": {
|
||||
"owner": "external-job",
|
||||
"type": "enum",
|
||||
"default": "slurm",
|
||||
"description": "Scheduler backend. SLURM is the first adapter; the field is the pluggability seam for future backends (LSF, PBS, Kubernetes batch). Core never interprets this value.",
|
||||
"values": [
|
||||
"slurm"
|
||||
]
|
||||
},
|
||||
"external_job.artifact_dir": {
|
||||
"owner": "external-job",
|
||||
"type": "string",
|
||||
"default": "Artifacts/jobs",
|
||||
"description": "Root for per-job artifact directories (e.g. Artifacts/jobs/<jobid>/). Avoids fixed log paths and hardcoding a cluster/project layout."
|
||||
},
|
||||
"external_job.submit_timeout_ms": {
|
||||
"owner": "external-job",
|
||||
"type": "number",
|
||||
"default": 30000,
|
||||
"description": "Hard timeout (ms) for the scheduler submit subprocess (e.g. sbatch). Bounded per CLAUDE.md unbounded-subprocess policy."
|
||||
},
|
||||
"external_job.poll_timeout_ms": {
|
||||
"owner": "external-job",
|
||||
"type": "number",
|
||||
"default": 15000,
|
||||
"description": "Hard timeout (ms) for the scheduler poll subprocess (squeue, with sacct fallback)."
|
||||
},
|
||||
"workflow.post_planning_gaps": {
|
||||
"owner": "gap-analysis",
|
||||
"type": "boolean",
|
||||
@@ -4653,6 +4805,7 @@ const _requiresGraph = {
|
||||
"copilot": [],
|
||||
"cursor": [],
|
||||
"drift": [],
|
||||
"external-job": [],
|
||||
"gap-analysis": [],
|
||||
"gemini": [],
|
||||
"graphify": [],
|
||||
|
||||
287
gsd-core/bin/lib/external-job.cjs
Normal file
287
gsd-core/bin/lib/external-job.cjs
Normal file
@@ -0,0 +1,287 @@
|
||||
"use strict";
|
||||
/**
|
||||
* external-job.cts — scheduler-adapter producer module for the async
|
||||
* external-job contract (#1164 / #1105).
|
||||
*
|
||||
* The CORE loop CONSUMES manifests at `.planning/async-jobs/<job>.json`
|
||||
* (#1165 — external_job_waiting half-state); this module is the Capability
|
||||
* half that PRODUCES them. SLURM is the first backend; the design stays
|
||||
* scheduler-pluggable via the `backend` field (planning-artifacts.md).
|
||||
*
|
||||
* Pure helpers (state map, build, validate, parsers) take no I/O; the writer
|
||||
* takes injected `fs` and `clock` seams so tests drive it without touching disk
|
||||
* or wall-clock time (CLAUDE.md clock-seam + injectable-deps conventions).
|
||||
*
|
||||
* Build: `src/*.cts` -> `gsd-core/bin/lib/*.cjs` (ADR-457 build-at-publish).
|
||||
*/
|
||||
var __importDefault = (this && this.__importDefault) || function (mod) {
|
||||
return (mod && mod.__esModule) ? mod : { "default": mod };
|
||||
};
|
||||
const node_path_1 = __importDefault(require("node:path"));
|
||||
const node_fs_1 = __importDefault(require("node:fs"));
|
||||
// ─── Closed status enum (stability contract — Hyrum's Law) ────────────────────
|
||||
const MANIFEST_VERSION = '1.0';
|
||||
const MANIFEST_STATUS = [
|
||||
'submitted',
|
||||
'running',
|
||||
'completed-unverified',
|
||||
'failed',
|
||||
'cancelled',
|
||||
'timeout',
|
||||
];
|
||||
const NON_TERMINAL_STATUSES = ['submitted', 'running'];
|
||||
const TERMINAL_FAILURE_STATUSES = ['failed', 'cancelled', 'timeout'];
|
||||
// ─── SLURM state -> manifest status ───────────────────────────────────────────
|
||||
//
|
||||
// Source: SLURM job state codes (squeue/sacct State column). Producers for
|
||||
// other backends map their own states onto the closed enum above; this table
|
||||
// is SLURM-specific and lives behind the `backend: 'slurm'` field.
|
||||
const SLURM_STATE_MAP = {
|
||||
PENDING: 'submitted',
|
||||
CONFIGURING: 'submitted',
|
||||
RUNNING: 'running',
|
||||
COMPLETING: 'running',
|
||||
COMPLETED: 'completed-unverified',
|
||||
FAILED: 'failed',
|
||||
CANCELLED: 'cancelled',
|
||||
TIMEOUT: 'timeout',
|
||||
OUT_OF_MEMORY: 'failed',
|
||||
BOOT_FAIL: 'failed',
|
||||
NODE_FAIL: 'failed',
|
||||
PREEMPTED: 'failed',
|
||||
};
|
||||
/**
|
||||
* Map a raw SLURM state string to the closed, scheduler-agnostic manifest
|
||||
* status. Case-insensitive; trims whitespace; strips a trailing by-part
|
||||
* ("CANCELLED by 1001" -> "CANCELLED"). Returns `null` for any unknown
|
||||
* state so the caller can decide whether to surface or fail — never guesses
|
||||
* (CLAUDE.md anti-guessing).
|
||||
*/
|
||||
function mapSlurmState(raw) {
|
||||
if (typeof raw !== 'string')
|
||||
return null;
|
||||
const key = raw.trim().toUpperCase();
|
||||
const head = key.split(/\s+/)[0];
|
||||
if (head && Object.prototype.hasOwnProperty.call(SLURM_STATE_MAP, head)) {
|
||||
return SLURM_STATE_MAP[head];
|
||||
}
|
||||
return null;
|
||||
}
|
||||
const REQUIRED_FIELDS = [
|
||||
'plan_id',
|
||||
'phase',
|
||||
'job_id',
|
||||
'backend',
|
||||
'submit_command',
|
||||
'status',
|
||||
'expected_artifacts',
|
||||
'verification_command',
|
||||
'resume_command',
|
||||
];
|
||||
function assertString(v, key) {
|
||||
if (typeof v !== 'string' || v.length === 0) {
|
||||
throw new Error(`buildManifest: field "${key}" must be a non-empty string`);
|
||||
}
|
||||
}
|
||||
const STATUS_LIST = MANIFEST_STATUS;
|
||||
/**
|
||||
* Build a versioned, frozen manifest. Stamps `version` and `submitted_at`
|
||||
* (via the injected clock seam) and normalises `terminal_details`:
|
||||
* `null` unless the status is a terminal failure AND details were supplied.
|
||||
* Throws on missing required fields or an out-of-enum status.
|
||||
*/
|
||||
function buildManifest(input, opts = {}) {
|
||||
const inputRecord = input;
|
||||
for (const key of REQUIRED_FIELDS) {
|
||||
const v = inputRecord[key];
|
||||
if (key === 'expected_artifacts') {
|
||||
if (!Array.isArray(v) || v.length === 0 || !v.every((x) => typeof x === 'string')) {
|
||||
throw new Error('buildManifest: field "expected_artifacts" must be a non-empty string[]');
|
||||
}
|
||||
continue;
|
||||
}
|
||||
assertString(v, key);
|
||||
}
|
||||
if (!STATUS_LIST.includes(input.status)) {
|
||||
throw new Error(`buildManifest: field "status" must be one of ${MANIFEST_STATUS.join(', ')}`);
|
||||
}
|
||||
const clock = opts.clock ?? { nowIso: () => new Date().toISOString() };
|
||||
const isTerminalFailure = TERMINAL_FAILURE_STATUSES.includes(input.status);
|
||||
const terminal_details = isTerminalFailure && input.terminal_details ? input.terminal_details : null;
|
||||
return Object.freeze({
|
||||
version: MANIFEST_VERSION,
|
||||
job_id: input.job_id,
|
||||
plan_id: input.plan_id,
|
||||
phase: input.phase,
|
||||
backend: input.backend,
|
||||
submit_command: input.submit_command,
|
||||
status: input.status,
|
||||
expected_artifacts: [...input.expected_artifacts],
|
||||
verification_command: input.verification_command,
|
||||
resume_command: input.resume_command,
|
||||
submitted_at: clock.nowIso(),
|
||||
terminal_details,
|
||||
});
|
||||
}
|
||||
/**
|
||||
* Producer-side schema validator — the mirror of the consumer trust boundary
|
||||
* (planning-artifacts.md). Producers MUST emit a manifest this accepts; the
|
||||
* core loop re-validates defensively on read.
|
||||
*/
|
||||
function validateManifest(value) {
|
||||
if (!value || typeof value !== 'object') {
|
||||
return { ok: false, errors: ['manifest must be an object'] };
|
||||
}
|
||||
const m = value;
|
||||
const errors = [];
|
||||
if (m.version !== MANIFEST_VERSION)
|
||||
errors.push(`version must be "${MANIFEST_VERSION}"`);
|
||||
for (const f of ['job_id', 'plan_id', 'phase', 'backend', 'submit_command', 'verification_command', 'resume_command', 'submitted_at']) {
|
||||
if (typeof m[f] !== 'string' || m[f].length === 0) {
|
||||
errors.push(`field "${f}" must be a non-empty string`);
|
||||
}
|
||||
}
|
||||
if (typeof m.status !== 'string' || !STATUS_LIST.includes(m.status)) {
|
||||
errors.push(`status must be one of ${MANIFEST_STATUS.join(', ')}`);
|
||||
}
|
||||
if (!Array.isArray(m.expected_artifacts) || !m.expected_artifacts.every((x) => typeof x === 'string')) {
|
||||
errors.push('expected_artifacts must be a string[]');
|
||||
}
|
||||
if (m.terminal_details !== null && typeof m.terminal_details !== 'object') {
|
||||
errors.push('terminal_details must be null or an object');
|
||||
}
|
||||
return errors.length === 0 ? { ok: true } : { ok: false, errors };
|
||||
}
|
||||
/**
|
||||
* Parse `sbatch --parsable` output. Accepts either a bare job id
|
||||
* (`"12345"`) or the `"12345;clustername"` form. Rejects prose like
|
||||
* `"Submitted batch job 12345"` (that is the non-parsable default format).
|
||||
*/
|
||||
function parseSbatchParsable(stdout) {
|
||||
if (typeof stdout !== 'string')
|
||||
return { ok: false, kind: 'non_string', raw: String(stdout) };
|
||||
const trimmed = stdout.trim();
|
||||
if (!trimmed)
|
||||
return { ok: false, kind: 'empty', raw: stdout };
|
||||
const head = trimmed.split(';')[0].split(/\s+/)[0];
|
||||
if (!/^\d+$/.test(head))
|
||||
return { ok: false, kind: 'not_numeric', raw: trimmed };
|
||||
return { ok: true, job_id: head };
|
||||
}
|
||||
/**
|
||||
* Parse a single `squeue` line of the form `"<jobid> <state>"`. Returns `null`
|
||||
* for a header or any row that does not have at least two tokens.
|
||||
*/
|
||||
function parseSqueueLine(line) {
|
||||
if (typeof line !== 'string')
|
||||
return null;
|
||||
const parts = line.trim().split(/\s+/);
|
||||
if (parts.length < 2)
|
||||
return null;
|
||||
const [job_id, state] = parts;
|
||||
if (!/^\d+$/.test(job_id))
|
||||
return null;
|
||||
return { job_id, state };
|
||||
}
|
||||
/**
|
||||
* Parse a `sacct -P` row given as pre-split columns where index 0 is the job
|
||||
* id and index 1 is the state. Returns `null` for malformed rows.
|
||||
*/
|
||||
function parseSacctRow(cols) {
|
||||
if (cols.length < 2)
|
||||
return null;
|
||||
const job_id = cols[0];
|
||||
const state = cols[1];
|
||||
if (typeof job_id !== 'string' || typeof state !== 'string')
|
||||
return null;
|
||||
if (!/^\d+$/.test(job_id))
|
||||
return null;
|
||||
return { job_id, state };
|
||||
}
|
||||
/**
|
||||
* Pure path projection: `.planning/async-jobs/<job_id>.json`.
|
||||
*/
|
||||
function manifestPath(planningDir, jobId) {
|
||||
return node_path_1.default.join(planningDir, 'async-jobs', `${jobId}.json`);
|
||||
}
|
||||
function _isNonTerminal(status) {
|
||||
return typeof status === 'string' && NON_TERMINAL_STATUSES.includes(status);
|
||||
}
|
||||
/**
|
||||
* Write a manifest to `.planning/async-jobs/<job_id>.json`.
|
||||
*
|
||||
* Fail-closed rules (mirror of the consumer contract, planning-artifacts.md):
|
||||
* - If any existing manifest in the dir shares `plan_id` but has a different
|
||||
* `job_id` AND is non-terminal -> refuse (`duplicate_plan_id`); dispatching
|
||||
* again would duplicate the external job.
|
||||
* - If the target file exists but is not valid JSON -> refuse
|
||||
* (`malformed_existing`); never silently clobber.
|
||||
* - Same `job_id` for the same `plan_id` -> allowed (status progression).
|
||||
* - A prior job for the same `plan_id` that is already terminal -> allowed
|
||||
* (the duplicate guard only protects against re-dispatching live work).
|
||||
*/
|
||||
function writeManifest(manifest, planningDir, opts = {}) {
|
||||
const fs = opts.fs ?? node_fs_1.default;
|
||||
const dir = node_path_1.default.join(planningDir, 'async-jobs');
|
||||
const target = manifestPath(planningDir, manifest.job_id);
|
||||
let names;
|
||||
try {
|
||||
fs.mkdirSync(dir, { recursive: true });
|
||||
names = fs.readdirSync(dir);
|
||||
}
|
||||
catch (e) {
|
||||
return { ok: false, kind: 'io_error', message: e.message };
|
||||
}
|
||||
for (const name of names) {
|
||||
if (!name.endsWith('.json'))
|
||||
continue;
|
||||
const p = node_path_1.default.join(dir, name);
|
||||
let raw;
|
||||
try {
|
||||
raw = String(fs.readFileSync(p));
|
||||
}
|
||||
catch {
|
||||
continue;
|
||||
}
|
||||
let existing;
|
||||
try {
|
||||
existing = JSON.parse(raw);
|
||||
}
|
||||
catch {
|
||||
if (p === target) {
|
||||
return { ok: false, kind: 'malformed_existing', message: `target manifest ${p} is not valid JSON` };
|
||||
}
|
||||
continue;
|
||||
}
|
||||
const samePlan = existing.plan_id === manifest.plan_id;
|
||||
const sameJob = existing.job_id === manifest.job_id;
|
||||
if (samePlan && !sameJob && _isNonTerminal(existing.status)) {
|
||||
return {
|
||||
ok: false,
|
||||
kind: 'duplicate_plan_id',
|
||||
message: `plan_id "${manifest.plan_id}" already has non-terminal job "${String(existing.job_id)}" at ${p}; dispatching again would duplicate the external job`,
|
||||
};
|
||||
}
|
||||
}
|
||||
try {
|
||||
fs.writeFileSync(target, JSON.stringify(manifest, null, 2) + '\n');
|
||||
}
|
||||
catch (e) {
|
||||
return { ok: false, kind: 'io_error', message: e.message };
|
||||
}
|
||||
return { ok: true, path: target };
|
||||
}
|
||||
module.exports = {
|
||||
MANIFEST_VERSION,
|
||||
MANIFEST_STATUS,
|
||||
NON_TERMINAL_STATUSES,
|
||||
TERMINAL_FAILURE_STATUSES,
|
||||
mapSlurmState,
|
||||
buildManifest,
|
||||
validateManifest,
|
||||
parseSbatchParsable,
|
||||
parseSqueueLine,
|
||||
parseSacctRow,
|
||||
writeManifest,
|
||||
manifestPath,
|
||||
};
|
||||
Reference in New Issue
Block a user