From d93f8e4da9879b42486afe4ad3da7564a68fcca0 Mon Sep 17 00:00:00 2001 From: Jason McCartney Date: Sat, 17 Oct 2026 10:16:37 -0700 Subject: [PATCH 1/2] schedule: built-in hardware reservations and an agent job log Agents shared the Spark pair through a sideband mkdir lock (pair.sh) that suspended LLooM and rerouted aliases in separate steps, and the only way to learn how a job was going was to ask the agent. The schedule-owning gateway (normally the head) now does both. - Reservations are expiring leases on configured resources. Taking one runs the resource's actions (route aliases away, then drain and suspend models), locally or on another cluster node through its admin API. Release undoes them in reverse and reports success only after the resumed models pass their health check, so traffic never goes back to a suspended or unhealthy model. A failed restore, or a restart mid-action, leaves the lease `stuck` and the resource held until a release succeeds. - Each step records what it found: a route already on its target and a model already suspended are left as they were, and a route with no active profile is refused. Resources that control the same model or route are config errors. - Queues with renewal, TTLs with warnings and auto-restore on expiry, leases extended by their holder's job heartbeats, and indefinite leases where a resource allows them. Expiry and promotion pause during a deployment fence. - Jobs: notes, progress, metrics, receipts and heartbeats, with stalled/abandoned detection, an event feed (`/gateway/schedule?since=`) for boards such as Tessera, and a rotated NDJSON history. - CLI: `lloom schedule`, `reserve`, `renew`, `release`, `job`. They target a remote head with LLOOM_HEAD_URL/LLOOM_HEAD_API_KEY and never send the local key there. `reserve --wait` exits 0 only once active, rides out head blips, and gives its place back on Ctrl-C. - Schedule refusals reach the CLI as redacted, bounded messages, including redaction of the key just sent. An unreadable schedule.json disables only the schedule routes, never inference. Co-authored-by: Claude Opus 5.5 --- AGENTS.md | 1 + CHANGELOG.md | 2 + README.md | 16 + bin/lloom.mjs | 32 +- docs/reservations.md | 191 ++++++ package.json | 4 +- src/deployment-fence.mjs | 2 +- src/gateway-client.mjs | 31 +- src/schedule-cli.mjs | 388 ++++++++++++ src/schedule.mjs | 976 +++++++++++++++++++++++++++++ src/server.mjs | 124 +++- test/schedule-integration.test.mjs | 355 +++++++++++ test/schedule.test.mjs | 375 +++++++++++ 13 files changed, 2487 insertions(+), 10 deletions(-) create mode 100644 docs/reservations.md create mode 100644 src/schedule-cli.mjs create mode 100644 src/schedule.mjs create mode 100644 test/schedule-integration.test.mjs create mode 100644 test/schedule.test.mjs diff --git a/AGENTS.md b/AGENTS.md index 226d563..6c4b68f 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -164,6 +164,7 @@ CLI surface evolves; check `lloom help` before documenting a missing command or - `spark-node-01` is the common Spark host alias, but re-check SSH, config, active workload, container/image, and memory state. - Use Tailscale/private admin reachability where configured. Keep inference/admin authorization distinct. - Before changing Jinx's substrate, gracefully drain her Runtime presence or otherwise coordinate the live workload. +- For shared hardware, take a lease on the owning head: `lloom reserve --apply --yes`, report work with `lloom job …`, and `lloom release --apply --yes` when done; check `lloom schedule` before using hardware a lease covers. See `docs/reservations.md`. - For temporary hardware takeover, use `lloom suspend --apply --yes` and `lloom resume --apply --yes` through the owner gateway; see `docs/model-maintenance.md`. Model maintenance gates all aliases and bare model calls, drains active work, verifies unloading, and restores routing only after health. Alias-member suspension alone does not unload or block direct calls. Older deployed gateways without these commands still require the legacy takeover procedure. - Sync LLooM credentials from the actual gateway environment; a 401 after a model change may be stale auth rather than model failure. - Do not call an alternate lane as a “light role model” if doing so causes disruptive swapping. Co-residency and quality must be demonstrated. diff --git a/CHANGELOG.md b/CHANGELOG.md index 48d554a..77f87ff 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,8 @@ All notable changes to LLooM will be documented in this file. The format follows ### Added +- Built-in reservations and a job log (`lloom schedule`, `reserve`, `renew`, `release`, `job`; `/gateway/schedule`). Expiring leases on configured resources run route and suspend actions, possibly on other cluster nodes, and undo them in reverse with health-gated restore. There are queues, renewal and heartbeat-extended leases, stalled-job detection, and an event feed for boards such as Tessera. Schedule refusals now reach the CLI as redacted messages. + - Atlas SparkQwen RC16 two-Spark recipe (`linux-nvidia-dgx-spark-2x-qwen38-flash-next-atlas`, profile `8x262k`) with a fail-closed pin gate and a digest-only image installer, and explicit per-node `${fabricHca}` settings for distributed recipes. Both Atlas recipes pin the published images with exact rail selection; missing settings and ambiguous discovery fail clearly. - LLooM Hear CPU audio analysis, with structured estimates, optional dashboard images, and opt-in upstream interpretation. File and URL inputs require operator configuration. diff --git a/README.md b/README.md index e32c43b..fc1ec12 100644 --- a/README.md +++ b/README.md @@ -587,6 +587,22 @@ candidates. Resume restores eligibility after guarded loading and health checks. Omit `--apply --yes` to preview. See [model maintenance](docs/model-maintenance.md) for alias setup, distributed ownership, timeouts, and recovery. +### Share hardware between agents + +```sh +lloom schedule # holders, queues, jobs +lloom reserve pair --purpose "decode A/B" --wait --apply --yes +job=$(lloom job start "decode A/B" --on pair) +lloom job progress $job 2/6 --unit arms +lloom job done $job "flat vs baseline" +lloom release pair --apply --yes +``` + +A reservation is an expiring lease on a configured resource. Taking it routes +aliases away and suspends that resource's models, and releasing it restores them +after a health check. Jobs report progress to one board instead of each agent's +window. See [reservations and jobs](docs/reservations.md). + For authentication, transport failures, and machine-readable CLI errors, see [Gateway diagnostics](docs/cli-diagnostics.md). diff --git a/bin/lloom.mjs b/bin/lloom.mjs index b599222..06b8b49 100755 --- a/bin/lloom.mjs +++ b/bin/lloom.mjs @@ -93,6 +93,7 @@ import { import { createLloomServer } from '../src/server.mjs'; import { deploymentFailureReport, formatDeploymentReport, runDeploymentCli } from '../src/cluster-deployment-cli.mjs'; import { runNodeAgentCli } from '../src/node-agent-cli.mjs'; +import { runScheduleCli, SCHEDULE_COMMANDS, scheduleUsage } from '../src/schedule-cli.mjs'; import { applySetup, createSetupPlan, syncClusterSetupMembers } from '../src/setup.mjs'; import { createSetupStatus } from '../src/setup-status.mjs'; import { @@ -117,6 +118,11 @@ const COMMAND_REGISTRY = [ { name: 'suspend', aliases: [], tier: 'primary', needsInstalledConfig: true }, { name: 'resume', aliases: [], tier: 'primary', needsInstalledConfig: true }, { name: 'route', aliases: ['routing'], tier: 'primary', needsInstalledConfig: true }, + { name: 'schedule', aliases: [], tier: 'primary', needsInstalledConfig: false }, + { name: 'reserve', aliases: [], tier: 'primary', needsInstalledConfig: false }, + { name: 'renew', aliases: [], tier: 'primary', needsInstalledConfig: false }, + { name: 'release', aliases: [], tier: 'primary', needsInstalledConfig: false }, + { name: 'job', aliases: ['jobs'], tier: 'primary', needsInstalledConfig: false }, { name: 'fleet', aliases: ['profiles'], tier: 'primary', needsInstalledConfig: true }, { name: 'integrate', aliases: [], tier: 'primary', needsInstalledConfig: true }, { name: 'integrations', aliases: [], tier: 'primary', needsInstalledConfig: true }, @@ -194,6 +200,9 @@ Primary commands: lloom models List gateway model IDs lloom suspend --apply --yes Drain, unload, and suspend a model lloom resume --apply --yes Load, verify health, and restore routing + lloom schedule Who holds shared hardware, queues, and agent jobs + lloom reserve Lease shared hardware (preview; add --apply --yes) + lloom job start Report a long-running job; returns its id lloom integrate <client|all> Write client configs (omp, opencode, codex, …) lloom add-model <ref> Import an ad hoc model (dry-run; add --apply --yes) lloom remove-model <model-id> Remove a model and its dedicated config resources @@ -205,7 +214,9 @@ Also useful: lloom help advanced Full command catalog Environment: - LLOOM_CONFIG Path to config JSON + LLOOM_CONFIG Path to config JSON + LLOOM_HEAD_URL Gateway that owns the schedule (key in LLOOM_HEAD_API_KEY) + LLOOM_OWNER Your name on reservations and jobs `; } @@ -256,6 +267,8 @@ Backends and runtimes: lloom runtime-start|runtime-warmup|runtime-stop <runtime-id> lloom keep-warm +${scheduleUsage()} + Models and clients: lloom route [alias [profile]] [--suspend-member model|--resume-member model] [--apply --yes] lloom add-model <hf-url|repo-id|local-path|ollama:tag|lmstudio:id|openai:url#model> [--backend id] [--api-key-env NAME] [--input TYPE] [--capability ID] [--tag ID] [--keep-warm] [--default] [--go|--apply --yes] @@ -399,7 +412,8 @@ const INSTALLED_CONFIG_COMMANDS = new Set([ 'setup', 'setup-status', 'status', - 'up' + 'up', + ...SCHEDULE_COMMANDS ]); const OPERATIONAL_CONFIG_COMMANDS = new Set([ @@ -3074,6 +3088,11 @@ async function main() { handlers['pack-export'] = handlers['recipe-export']; handlers['recipe-pack'] = handlers['recipe-import']; handlers['pack-submit'] = handlers['recipe-submit']; + for (const name of SCHEDULE_COMMANDS) { + handlers[name] = async ({ args, config, command: scheduleCommand }) => { + process.exitCode = (await runScheduleCli(scheduleCommand, args, { config })) || process.exitCode; + }; + } handlers['profiles'] = handlers['fleet']; handlers['routing'] = handlers['route']; handlers['runtime-status'] = handlers['runtimes']; @@ -3117,10 +3136,15 @@ async function main() { // Redact any configured credential value from arbitrary text (including stacks) // before it can reach a log, error, or receipt. +// Keys the CLI may have sent: an explicitly selected one, and a remote head's. +function sentKeyValues() { + return [cliAdminKeyEnvName ? process.env[cliAdminKeyEnvName] : null, process.env.LLOOM_HEAD_API_KEY]; +} + function safeErrorText(text, config = null) { return redactGatewaySecrets(text, { config: config ?? currentCliConfig ?? undefined, - extraValues: cliAdminKeyEnvName ? [process.env[cliAdminKeyEnvName]] : [] + extraValues: sentKeyValues() }); } @@ -3136,7 +3160,7 @@ function formatCliError(error, config = currentCliConfig) { function formatCliErrorJson(error) { const options = { config: currentCliConfig, - extraValues: cliAdminKeyEnvName ? [process.env[cliAdminKeyEnvName]] : [] + extraValues: sentKeyValues() }; // Receipts describe partial service mutations; retain them while redacting // every string value before serializing the envelope. diff --git a/docs/reservations.md b/docs/reservations.md new file mode 100644 index 0000000..667ef2e --- /dev/null +++ b/docs/reservations.md @@ -0,0 +1,191 @@ +# Reservations and jobs + +Several agents (or people) sharing a few machines need two things from +LLooM. The first is a way to take hardware without colliding with each other or +with production serving. The second is a single place that answers "how is it +going?" without anyone having to ask each agent. + +The schedule-owning gateway provides both. This is normally the cluster head. + +- A **reservation** is a lease on a named resource. Taking it runs that + resource's actions: route aliases away, then drain and suspend models. + Releasing it undoes them in reverse. A release only reports success once the + resumed models pass their health check, and only then is traffic routed back. + Leases expire unless they are renewed, so an agent that vanishes cannot hold a + machine forever. +- A **job** is a status record that an agent posts and keeps current: notes, + progress, metrics, a receipt path and an implicit heartbeat. A job that stops + reporting is marked stalled. A job attached to a reservation keeps that lease + alive while it reports. + +Both appear on one board: `lloom schedule`, `GET /gateway/schedule`, and +Tessera's Fleet band. + +## Configure resources + +Resources live in the owning gateway's config: + +```json +{ + "reservations": { + "resources": { + "pair": { + "description": "two-node Atlas pair", + "hosts": ["gpu-box-1", "gpu-box-2"], + "routes": [{ "alias": "omp", "profile": "cloud", "node": "gpu-box-1" }], + "suspend": [{ "model": "pair-model", "node": "gpu-box-1" }] + }, + "builds": { "description": "build slots on gpu-box-3", "capacity": 2 }, + "drafter": { "hosts": ["gpu-box-3", "gpu-box-4"], "allowIndefinite": true } + } + } +} +``` + +| Field | Meaning | +|-------|---------| +| `description`, `hosts` | Shown on the board. | +| `capacity` | Number of concurrent holders. The default is 1. Resources with actions must have capacity 1. | +| `routes` | Route-profile switches applied on take, in order (`lloom route <alias> <profile>`). The profile active at take time is recorded and restored on release. | +| `suspend` | Models to drain and unload on take (`lloom suspend`) and resume on release (`lloom resume`). | +| `node` | The cluster node whose gateway owns the route or model. Omit it for the local gateway. Remote steps use that node's admin API with its `cluster.nodes.<id>` credential, so the node needs `security.allowRemoteAdmin`. | +| `ttlMs`, `maxTtlMs` | Default lease length (30 min) and longest requestable lease (8 h). | +| `allowIndefinite` | Permit `--ttl none`, for standing reservations such as "these hosts belong to the drafter project". | +| `onExpiry` | `release` (the default) restores on expiry. `keep` only flags the lease as overdue. | + +Schedule-wide settings sit beside `resources`: `warnBeforeMs` (5 min), +`queueTtlMs` (30 min), `staleAfterMs` (15 min), `drainTimeoutMs` (5 min), +`retentionMs` (7 days) and `eventLimit` (500). State is kept in +`schedule.json` beside the config, and the durable history is appended to +`schedule-events.ndjson`. Override these locations with `statePath` and +`eventsPath`. + +Routes run before suspends on take, so traffic has already moved when a model +drains. Each step records what it found, so release restores that and nothing +more: + +- A route already on the target profile is left alone. +- A model that was already suspended stays suspended. +- A route with no active profile has nothing exact to restore, so the take + refuses it. Select a profile first with `lloom route <alias> <profile>`. +- Two resources that suspend the same model, or move the same route, would undo + each other's leases. Both are reported as configuration errors until one of + them is changed. A step that fails rolls back the steps already done, and the reservation +ends as `failed`. A restore that fails leaves the reservation `stuck`: the +resource stays held and traffic stays away from the unhealthy model until a +`release` succeeds. A gateway restart during an action also leaves the +reservation `stuck`. Releasing it re-runs the undo, which is idempotent. + +## Use it from an agent + +Point the CLI at the head once. The key is the head's admin key; it is never +taken from argv, and the local gateway's key is never sent to a different +host: + +```sh +export LLOOM_HEAD_URL=http://head.tailnet.example:8100 # private network, or https +export LLOOM_HEAD_API_KEY=... # admin key for the head +export LLOOM_OWNER=buildopt # who you are on the board +export LLOOM_SESSION=claude:<session-id> # optional; lets Tessera open your conversation +``` + +Then: + +```sh +lloom schedule # who holds what, queues, jobs +lloom reserve pair --purpose "decode A/B" # preview the actions +lloom reserve pair --purpose "decode A/B" --wait --apply --yes # queue if busy +job=$(lloom job start "decode A/B" --on pair --note "building") +lloom job progress $job 2/6 --unit arms "arm 2 of 6" +lloom job metric $job tok_s=58.2 image=f399caa +lloom job done $job "flat vs baseline" --receipt receipts/decode-ab.json +lloom release pair --apply --yes # restore and free +``` + +- `reserve` without `--wait` fails straight away when the resource is busy, and + names the holder. +- With `--wait` it queues and polls until the lease is active. It renews its + queue slot while it waits, and an abandoned wait drops out after `queueTtlMs`. +- Taking a resource you already hold renews the lease. +- `renew` and `release` accept the reservation id or the resource name. + `release --force` frees another owner's lease. +- `--ttl 2h` sets the lease length, and `--ttl none` asks for an indefinite one. +- Every job update (`note`, `progress`, `metric`, `beat`, `done`) is a heartbeat. + Updates carry `LLOOM_OWNER` when it is set, and the head refuses updates to + another owner's job. +- A job attaches only to its owner's lease. +- Two agents under the same owner are different holders when both name a session. +- `lloom job start` prints only the id, so scripts can capture it. Add `--json` + to any command for the full record. + +## HTTP API + +These are `/gateway` admin routes, so the same rules apply as for other admin +endpoints: once `security.adminApiKeys` is set they need an admin key, and a +non-loopback bind needs `security.allowRemoteAdmin` for writes. They are served +only by a primary gateway with a file-backed config. An unreadable +`schedule.json` makes only these routes answer 503 `schedule_unavailable`; the +gateway still serves inference, and the file is left untouched for inspection. + +| Route | Purpose | +|-------|---------| +| `GET /gateway/schedule?since=SEQ` | Board: `resources` (with `holders` and `queue`), `reservations`, `jobs` (with derived `health` and `heartbeatAgeMs`) and `events` after `SEQ`. | +| `POST /gateway/schedule/reservations` | `{resource, owner, purpose?, session?, ttlMs?, wait?}`. Returns a preview unless `apply` and `yes` are both true. | +| `GET /gateway/schedule/reservations/:id` | One reservation, including `position` when queued and `remainingMs` when active. | +| `POST /gateway/schedule/reservations/:id/renew` | `{owner, ttlMs?}` | +| `POST /gateway/schedule/reservations/:id/release` | `{owner, force?, apply, yes}`. Without `apply`, returns the undo plan. | +| `POST /gateway/schedule/jobs` | `{title, owner, session?, resource? or reservationId?, note?, staleAfterMs?}` | +| `GET /gateway/schedule/jobs/:id` | One job. | +| `POST /gateway/schedule/jobs/:id` | `{note?, progress?: {done, total?, unit?}, metrics?, receipt?, state?: done or failed or cancelled}` | + +Refusals carry `error.type: "schedule_error"` and a code. For example, +`schedule_busy` (409) includes `details.holders`. + +### Events + +Event kinds: + +- `reservation.acquiring`, `.queued`, `.active`, `.expiring`, `.overdue`, `.expired`, `.released`, `.withdrawn`, `.failed`, `.stuck` +- `job.started`, `.stalled`, `.recovered`, `.done`, `.failed`, `.cancelled` + +Each event has a `severity`: + +- `info` +- `attention`: expiring or overdue leases, stalled jobs, finished jobs +- `error`: failed jobs, failed takes, stuck restores + +Boards poll with `since` to receive only new events. Job notes are written to +the durable log but kept out of the event ring. + +## Limits and recovery + +- At most 500 jobs run at once, and at most 50 agents wait per resource. +- A running job with no report for 24 hours is failed as abandoned. +- The event log rotates to `schedule-events.ndjson.1` at 16 MB. +- While a deployment fence holds, leases neither expire nor start, and they get + that time back afterwards. +- When the gateway shuts down, it waits a few seconds for actions in flight. + Anything still unfinished comes back as `stuck` on the next start. +- `reserve --wait` exits 0 only once the lease is active. It keeps waiting + through a briefly unreachable head, and on Ctrl-C it gives its place back + (exit 130). + +## Security + +Agents on other machines need the head's admin key, and the head needs +`allowRemoteAdmin`. That key opens every admin route, not just the schedule. +There is no narrower schedule-only credential yet. Reach the head over a +private network such as Tailscale, or use HTTPS: never send the key in +cleartext across a network you do not control. + +## Scope + +LLooM enforces reservations through the actions it performs, by suspending a +model and routing traffic away from it. It does not stop processes it did not +start. A build or container launched outside LLooM honours a reservation only +when its agent checks `lloom schedule` or holds the lease, so standing +reservations such as `drafter` work by being visible. + +The schedule belongs to the gateway that keeps it. Moving the head to another +machine (`lloom cluster promote-head`) does not move `schedule.json`. Release +leases first, or copy the file across while both gateways are stopped. diff --git a/package.json b/package.json index d72357d..c33f761 100644 --- a/package.json +++ b/package.json @@ -64,7 +64,7 @@ "check": "npm run check:cluster && npm run check:js && npm run check:entity-stagger && npm run check:python && npm run check:shell && npm run check:spark-deploy", "check:cluster": "node --check src/cluster.mjs && node --check test/cluster.test.mjs && node --check test/cluster-availability.test.mjs && node --check test/cluster-readiness.test.mjs && node --check src/cluster-deployment.mjs && node --check src/cluster-deployment-cli.mjs && node --check src/ssh-deployment-transport.mjs && node --check test/cluster-deployment.test.mjs && node --check test/cluster-deployment-cli.test.mjs && node --check test/cluster-deployment-e2e.test.mjs", "check:entity-stagger": "node --check scripts/entity-stagger-benchmark.mjs && node --check test/entity-stagger-benchmark.test.mjs", - "check:js": "node --check bin/lloom.mjs && node --check bin/lloom-host.mjs && node --check src/backend-catalog.mjs && node --check src/benchmarks.mjs && node --check src/bootstrap.mjs && node --check src/client-integrations.mjs && node --check src/community-client.mjs && node --check src/community-site.mjs && node --check src/config.mjs && node --check src/dashboard.mjs && node --check src/doctor.mjs && node --check src/host-memory.mjs && node --check src/host-server.mjs && node --check src/init.mjs && node --check src/installer.mjs && node --check src/interchange.mjs && node --check src/machine-profile.mjs && node --check src/managed-environment.mjs && node --check src/model-acquisition.mjs && node --check src/model-intake.mjs && node --check src/model-removal.mjs && node --check src/onboarding.mjs && node --check src/recipe-index.mjs && node --check src/recipe-pack-export.mjs && node --check src/recipe-pack.mjs && node --check src/registry.mjs && node --check src/route-control.mjs && node --check src/recipes.mjs && node --check src/resource-fit.mjs && node --check src/request-scheduler.mjs && node --check src/rate-limit.mjs && node --check src/runtime-manager.mjs && node --check src/runtime-policy.mjs && node --check src/server.mjs && node --check src/deployment-fence.mjs && node --check test/deployment-fence.test.mjs && node --check src/setup.mjs && node --check src/setup-status.mjs && node --check src/process-control.mjs && node --check src/security.mjs && node --check src/standby.mjs && node --check src/tts-catalog.mjs && node --check src/voice-profiles.mjs && node --check src/protocol/text.mjs && node --check src/protocol/reasoning-normalize.mjs && node --check src/protocol/reasoning-effort.mjs && node --check src/protocol/responses.mjs && node --check src/protocol/anthropic.mjs && node --check src/protocol/sse.mjs && node --check src/protocol/stream-anthropic.mjs && node --check src/protocol/stream-responses.mjs && node --check src/protocol/index.mjs && node --check scripts/apply-spark-route-catalog.mjs && node --check scripts/capture-live-metrics.mjs && node --check scripts/merge-live-metrics.mjs && node --check scripts/run-qualitative-chat-benchmark.mjs && node --check scripts/run-local-decision-benchmark.mjs && node --check scripts/check-community-deploy.mjs && node --check scripts/generate-client-configs.mjs && node --check scripts/check-interchange.mjs && node --check scripts/check-package.mjs && node --check scripts/release-lib.mjs && node --check scripts/build-release.mjs && node --check scripts/deploy-spark.mjs && node --check test/audio-generations.test.mjs && node --check test/community-host.test.mjs && node --check test/dashboard-status.test.mjs && node --check test/ds4fv-recipe.test.mjs && node --check test/dspark-hotfixes.test.mjs && node --check test/glm53-exl3-recipe.test.mjs && node --check test/host-memory.test.mjs && node --check test/metrics-persistence.test.mjs && node --check test/model-acquisition.test.mjs && node --check test/model-failover.test.mjs && node --check test/qwen38-vllm-recipe.test.mjs && node --check test/reasoning-normalize.test.mjs && node --check test/reasoning-effort.test.mjs && node --check test/resource-fit.test.mjs && node --check test/route-control.test.mjs && node --check test/spark-route-catalog.test.mjs && node --check test/runtime-watchdog.test.mjs && node --check test/tts-catalog.test.mjs && node --check test/voice-profiles.test.mjs && node --check test/chatterbox-recipe.test.mjs && node --check test/hear-recipe.test.mjs && node --check test/smoke.mjs && node --check src/runtime-supervisor.mjs && node --check test/runtime-supervisor.test.mjs && node --check test/comfyui-media.test.mjs && node --check test/spark-audio-recipes.test.mjs && node --check src/runtime-residency.mjs && node --check test/ltx-pipelines.test.mjs && node --check test/ltx-pipelines-recipe.test.mjs && node --check src/video-contracts.mjs && node --check src/dashboard-scene.mjs && node --check src/dashboard-presence.mjs && node --check src/dashboard-presence-client.mjs && node --check src/first-run.mjs && node --check src/first-run-page.mjs && node --check src/browser-setup.mjs && node --check src/runtime-preferences.mjs && node --check src/installation-jobs.mjs && node --check src/model-installation.mjs && node --check src/dashboard-installation.mjs && node --check src/runtime-memory-safety.mjs && node --check src/dashboard-memory.mjs && node --check src/runtime-memory-usage.mjs && node --check src/config-profiles.mjs && node --check test/config-profiles.test.mjs && node --check src/runtime-policy-config.mjs && node --check test/runtime-policy-config.test.mjs && node --check src/audio-providers.mjs && node --check src/protocol/openrouter-provider.mjs && node --check test/audio-providers.test.mjs && node --check test/openrouter-provider.test.mjs && node --check backends/atlas-sparkglm/verify-pins.mjs && node --check test/atlas-sparkglm-recipe.test.mjs && node --check backends/atlas-sparkqwen/verify-pins.mjs && node --check test/atlas-sparkqwen-recipe.test.mjs && node --check src/head-preparation.mjs && node --check src/head-transfer.mjs && node --check src/service-control.mjs && node --check src/gateway-relay.mjs && node --check src/client-retarget.mjs && node --check src/head-promotion.mjs && node --check src/head-promotion-transfer.mjs && node --check src/gateway-client.mjs && node --check test/cli-diagnostics.test.mjs && node --check src/service-install.mjs && node --check test/service-install.test.mjs && node --check src/cluster-deployment.mjs && node --check src/cluster-deployment-cli.mjs && node --check src/ssh-deployment-transport.mjs && node --check src/node-release-agent.mjs && node --check src/node-gateway-adapter.mjs && node --check src/node-agent-cli.mjs && node --check test/cluster-deployment.test.mjs && node --check test/cluster-deployment-cli.test.mjs && node --check test/cluster-deployment-e2e.test.mjs && node --check test/node-release-agent.test.mjs && node --check test/node-gateway-adapter.test.mjs && node --check test/node-gateway-adapter-e2e.test.mjs && node --check test/deployment-bundle.test.mjs && node --check test/node-agent-cli.test.mjs && node --check test/ssh-deployment-transport.test.mjs && node --check scripts/build-deployment-bundle.mjs", + "check:js": "node --check bin/lloom.mjs && node --check bin/lloom-host.mjs && node --check src/backend-catalog.mjs && node --check src/benchmarks.mjs && node --check src/bootstrap.mjs && node --check src/client-integrations.mjs && node --check src/community-client.mjs && node --check src/community-site.mjs && node --check src/config.mjs && node --check src/dashboard.mjs && node --check src/doctor.mjs && node --check src/host-memory.mjs && node --check src/host-server.mjs && node --check src/init.mjs && node --check src/installer.mjs && node --check src/interchange.mjs && node --check src/machine-profile.mjs && node --check src/managed-environment.mjs && node --check src/model-acquisition.mjs && node --check src/model-intake.mjs && node --check src/model-removal.mjs && node --check src/onboarding.mjs && node --check src/recipe-index.mjs && node --check src/recipe-pack-export.mjs && node --check src/recipe-pack.mjs && node --check src/registry.mjs && node --check src/route-control.mjs && node --check src/recipes.mjs && node --check src/resource-fit.mjs && node --check src/request-scheduler.mjs && node --check src/rate-limit.mjs && node --check src/runtime-manager.mjs && node --check src/runtime-policy.mjs && node --check src/server.mjs && node --check src/deployment-fence.mjs && node --check test/deployment-fence.test.mjs && node --check src/schedule.mjs && node --check src/schedule-cli.mjs && node --check test/schedule.test.mjs && node --check test/schedule-integration.test.mjs && node --check src/setup.mjs && node --check src/setup-status.mjs && node --check src/process-control.mjs && node --check src/security.mjs && node --check src/standby.mjs && node --check src/tts-catalog.mjs && node --check src/voice-profiles.mjs && node --check src/protocol/text.mjs && node --check src/protocol/reasoning-normalize.mjs && node --check src/protocol/reasoning-effort.mjs && node --check src/protocol/responses.mjs && node --check src/protocol/anthropic.mjs && node --check src/protocol/sse.mjs && node --check src/protocol/stream-anthropic.mjs && node --check src/protocol/stream-responses.mjs && node --check src/protocol/index.mjs && node --check scripts/apply-spark-route-catalog.mjs && node --check scripts/capture-live-metrics.mjs && node --check scripts/merge-live-metrics.mjs && node --check scripts/run-qualitative-chat-benchmark.mjs && node --check scripts/run-local-decision-benchmark.mjs && node --check scripts/check-community-deploy.mjs && node --check scripts/generate-client-configs.mjs && node --check scripts/check-interchange.mjs && node --check scripts/check-package.mjs && node --check scripts/release-lib.mjs && node --check scripts/build-release.mjs && node --check scripts/deploy-spark.mjs && node --check test/audio-generations.test.mjs && node --check test/community-host.test.mjs && node --check test/dashboard-status.test.mjs && node --check test/ds4fv-recipe.test.mjs && node --check test/dspark-hotfixes.test.mjs && node --check test/glm53-exl3-recipe.test.mjs && node --check test/host-memory.test.mjs && node --check test/metrics-persistence.test.mjs && node --check test/model-acquisition.test.mjs && node --check test/model-failover.test.mjs && node --check test/qwen38-vllm-recipe.test.mjs && node --check test/reasoning-normalize.test.mjs && node --check test/reasoning-effort.test.mjs && node --check test/resource-fit.test.mjs && node --check test/route-control.test.mjs && node --check test/spark-route-catalog.test.mjs && node --check test/runtime-watchdog.test.mjs && node --check test/tts-catalog.test.mjs && node --check test/voice-profiles.test.mjs && node --check test/chatterbox-recipe.test.mjs && node --check test/hear-recipe.test.mjs && node --check test/smoke.mjs && node --check src/runtime-supervisor.mjs && node --check test/runtime-supervisor.test.mjs && node --check test/comfyui-media.test.mjs && node --check test/spark-audio-recipes.test.mjs && node --check src/runtime-residency.mjs && node --check test/ltx-pipelines.test.mjs && node --check test/ltx-pipelines-recipe.test.mjs && node --check src/video-contracts.mjs && node --check src/dashboard-scene.mjs && node --check src/dashboard-presence.mjs && node --check src/dashboard-presence-client.mjs && node --check src/first-run.mjs && node --check src/first-run-page.mjs && node --check src/browser-setup.mjs && node --check src/runtime-preferences.mjs && node --check src/installation-jobs.mjs && node --check src/model-installation.mjs && node --check src/dashboard-installation.mjs && node --check src/runtime-memory-safety.mjs && node --check src/dashboard-memory.mjs && node --check src/runtime-memory-usage.mjs && node --check src/config-profiles.mjs && node --check test/config-profiles.test.mjs && node --check src/runtime-policy-config.mjs && node --check test/runtime-policy-config.test.mjs && node --check src/audio-providers.mjs && node --check src/protocol/openrouter-provider.mjs && node --check test/audio-providers.test.mjs && node --check test/openrouter-provider.test.mjs && node --check backends/atlas-sparkglm/verify-pins.mjs && node --check test/atlas-sparkglm-recipe.test.mjs && node --check backends/atlas-sparkqwen/verify-pins.mjs && node --check test/atlas-sparkqwen-recipe.test.mjs && node --check src/head-preparation.mjs && node --check src/head-transfer.mjs && node --check src/service-control.mjs && node --check src/gateway-relay.mjs && node --check src/client-retarget.mjs && node --check src/head-promotion.mjs && node --check src/head-promotion-transfer.mjs && node --check src/gateway-client.mjs && node --check test/cli-diagnostics.test.mjs && node --check src/service-install.mjs && node --check test/service-install.test.mjs && node --check src/cluster-deployment.mjs && node --check src/cluster-deployment-cli.mjs && node --check src/ssh-deployment-transport.mjs && node --check src/node-release-agent.mjs && node --check src/node-gateway-adapter.mjs && node --check src/node-agent-cli.mjs && node --check test/cluster-deployment.test.mjs && node --check test/cluster-deployment-cli.test.mjs && node --check test/cluster-deployment-e2e.test.mjs && node --check test/node-release-agent.test.mjs && node --check test/node-gateway-adapter.test.mjs && node --check test/node-gateway-adapter-e2e.test.mjs && node --check test/deployment-bundle.test.mjs && node --check test/node-agent-cli.test.mjs && node --check test/ssh-deployment-transport.test.mjs && node --check scripts/build-deployment-bundle.mjs", "check:python": "python3 -m py_compile backends/dspark-vllm/apply-patch-pack.py backends/dspark-vllm/packs/miaai-dsv4flash-d1b76251-defaults/patches/*.py backends/dspark-vllm/packs/miaai-ds4fv-f5665e8/patches/*.py backends/dspark-vllm/packs/miaai-ds4fv-f5665e8/patches/vision_exp/*.py backends/qwen38-vllm/apply-ple-fp8-patch.py backends/qwen38-sglang/*.py backends/glm53-exl3/*.py backends/mlx-audio/lloom_audio_server.py backends/hear/lloom_hear_server.py backends/chatterbox/lloom_chatterbox_server.py patches/apply_mtplx_longctx_fix.py backends/comfyui-media/bridge/*.py backends/comfyui-media/graphs/*.py backends/comfyui-media/build/*.py backends/comfyui-media/install.py backends/comfyui-media/derive_h3_turbo.py backends/qwen-image-diffusers/*.py backends/ace-step-diffusers/*.py backends/ace-step-diffusers/tests/*.py backends/hear/test/test_hear_security.py backends/spark-audio/install.py backends/spark-audio/lloom_audio_cuda_server.py backends/spark-audio/test_lloom_audio_cuda_server.py backends/ltx-pipelines/*.py src/darwin-memory-usage.py", "check:shell": "bash -n backends/dspark-vllm/entrypoint.sh backends/dspark-vllm/runtime/build-hardened-image.sh backends/dspark-vllm/packs/miaai-dsv4flash-d1b76251-defaults/patches/*.sh backends/dspark-vllm/packs/miaai-ds4fv-f5665e8/apply-runtime.sh backends/dspark-vllm/packs/miaai-ds4fv-f5665e8/patches/*.sh backends/qwen38-vllm/entrypoint.sh backends/qwen38-sglang/entrypoint.sh backends/glm53-exl3/entrypoint.sh backends/mlx-audio/install.sh backends/hear/install.sh backends/chatterbox/install.sh clients/examples/lloom-claude clients/examples/lloom-codex clients/examples/lloom-hermes clients/examples/lloom-opencode clients/examples/lloom-zero scripts/remote-install-spark.sh scripts/remote-stage-spark-worker.sh backends/atlas-sparkglm/install.sh backends/atlas-sparkqwen/install.sh", "check:spark-deploy": "node --check test/spark-deploy.test.mjs && node test/spark-deploy.test.mjs", @@ -73,7 +73,7 @@ "community:check": "node scripts/check-community-deploy.mjs", "package:check": "node scripts/check-package.mjs", "test:adoption": "node test/resource-fit.test.mjs && node test/model-acquisition.test.mjs && node test/dashboard-status.test.mjs", - "test:unit": "node --test test/deployment-fence.test.mjs test/cluster-deployment.test.mjs test/cluster-deployment-cli.test.mjs test/cluster-deployment-e2e.test.mjs test/node-release-agent.test.mjs test/node-gateway-adapter.test.mjs test/node-gateway-adapter-e2e.test.mjs test/deployment-bundle.test.mjs test/node-agent-cli.test.mjs test/ssh-deployment-transport.test.mjs && node test/runtime-health-ownership.test.mjs && node test/long-prefill-transport.test.mjs && node test/sparkglm-managed.test.mjs && node test/protocol.test.mjs && node test/protocol-stream.test.mjs && node test/stream-metrics.test.mjs && node test/responses-codex.test.mjs && node test/security.test.mjs && node test/community-host.test.mjs && node test/ds4fv-recipe.test.mjs && node test/dspark-hotfixes.test.mjs && node test/glm53-exl3-recipe.test.mjs && node test/host-memory.test.mjs && node test/config-profiles.test.mjs && node test/runtime-policy.test.mjs && node --test test/runtime-policy-config.test.mjs && node --test test/request-scheduler.test.mjs test/adaptive-priority.test.mjs test/rate-limit.test.mjs test/runtime-queue.test.mjs test/foreground-priority.test.mjs test/federated-priority.test.mjs test/runtime-supervisor.test.mjs && node test/interchange.test.mjs && node test/server-resilience.test.mjs && node --test test/standby.test.mjs && node test/metrics-persistence.test.mjs && node test/model-failover.test.mjs && node test/qwen38-vllm-recipe.test.mjs && node test/reasoning-normalize.test.mjs && node test/reasoning-effort.test.mjs && node test/route-control.test.mjs && node test/spark-route-catalog.test.mjs && node test/chat-lane-canary.test.mjs && node test/runtime-watchdog.test.mjs && node test/tts-catalog.test.mjs && node test/voice-profiles.test.mjs && node test/chatterbox-recipe.test.mjs && node test/hear-recipe.test.mjs && npm run test:adoption && node --test test/performance-sampler.test.mjs && node --test test/audio-generations.test.mjs && node test/video-providers.test.mjs && node --test test/web-functions.test.mjs && node --test test/config-mutation.test.mjs && node --test test/model-maintenance.test.mjs && node --test test/model-maintenance-control.test.mjs && node --test test/route-maintenance-concurrency.test.mjs && node --test test/model-maintenance-integration.test.mjs && node test/comfyui-media.test.mjs && node test/qwen-image-diffusers.test.mjs && node test/ace-step-diffusers.test.mjs && node test/runtime-residency.test.mjs && node test/spark-audio-recipes.test.mjs && node --test test/video-contracts.test.mjs && node test/ltx-pipelines-recipe.test.mjs && npm run test:ux && npm run test:memory-safety && node --test test/dashboard-memory.test.mjs test/runtime-memory-usage.test.mjs && node --test test/speech-stream.test.mjs && node --test test/audio-providers.test.mjs && node test/openrouter-provider.test.mjs && node test/atlas-sparkglm-recipe.test.mjs && node test/atlas-sparkqwen-recipe.test.mjs && node --test test/head-preparation.test.mjs test/head-transfer.test.mjs test/service-control.test.mjs && node --test test/gateway-relay.test.mjs test/client-retarget.test.mjs && node --test test/head-profile-acceptance.test.mjs test/head-promotion.test.mjs test/head-promotion-transfer.test.mjs test/head-transfer-acceptance.test.mjs && node --test test/cli-diagnostics.test.mjs && node --test test/service-install.test.mjs", + "test:unit": "node --test test/deployment-fence.test.mjs test/cluster-deployment.test.mjs test/cluster-deployment-cli.test.mjs test/cluster-deployment-e2e.test.mjs test/node-release-agent.test.mjs test/node-gateway-adapter.test.mjs test/node-gateway-adapter-e2e.test.mjs test/deployment-bundle.test.mjs test/node-agent-cli.test.mjs test/ssh-deployment-transport.test.mjs && node test/runtime-health-ownership.test.mjs && node test/long-prefill-transport.test.mjs && node test/sparkglm-managed.test.mjs && node test/protocol.test.mjs && node test/protocol-stream.test.mjs && node test/stream-metrics.test.mjs && node test/responses-codex.test.mjs && node test/security.test.mjs && node test/community-host.test.mjs && node test/ds4fv-recipe.test.mjs && node test/dspark-hotfixes.test.mjs && node test/glm53-exl3-recipe.test.mjs && node test/host-memory.test.mjs && node test/config-profiles.test.mjs && node test/runtime-policy.test.mjs && node --test test/runtime-policy-config.test.mjs && node --test test/request-scheduler.test.mjs test/adaptive-priority.test.mjs test/rate-limit.test.mjs test/runtime-queue.test.mjs test/foreground-priority.test.mjs test/federated-priority.test.mjs test/runtime-supervisor.test.mjs && node test/interchange.test.mjs && node test/server-resilience.test.mjs && node --test test/standby.test.mjs && node test/metrics-persistence.test.mjs && node test/model-failover.test.mjs && node test/qwen38-vllm-recipe.test.mjs && node test/reasoning-normalize.test.mjs && node test/reasoning-effort.test.mjs && node test/route-control.test.mjs && node test/spark-route-catalog.test.mjs && node test/chat-lane-canary.test.mjs && node test/runtime-watchdog.test.mjs && node test/tts-catalog.test.mjs && node test/voice-profiles.test.mjs && node test/chatterbox-recipe.test.mjs && node test/hear-recipe.test.mjs && npm run test:adoption && node --test test/performance-sampler.test.mjs && node --test test/audio-generations.test.mjs && node test/video-providers.test.mjs && node --test test/web-functions.test.mjs && node --test test/config-mutation.test.mjs && node --test test/model-maintenance.test.mjs && node --test test/model-maintenance-control.test.mjs && node --test test/route-maintenance-concurrency.test.mjs && node --test test/model-maintenance-integration.test.mjs && node --test test/schedule.test.mjs test/schedule-integration.test.mjs && node test/comfyui-media.test.mjs && node test/qwen-image-diffusers.test.mjs && node test/ace-step-diffusers.test.mjs && node test/runtime-residency.test.mjs && node test/spark-audio-recipes.test.mjs && node --test test/video-contracts.test.mjs && node test/ltx-pipelines-recipe.test.mjs && npm run test:ux && npm run test:memory-safety && node --test test/dashboard-memory.test.mjs test/runtime-memory-usage.test.mjs && node --test test/speech-stream.test.mjs && node --test test/audio-providers.test.mjs && node test/openrouter-provider.test.mjs && node test/atlas-sparkglm-recipe.test.mjs && node test/atlas-sparkqwen-recipe.test.mjs && node --test test/head-preparation.test.mjs test/head-transfer.test.mjs test/service-control.test.mjs && node --test test/gateway-relay.test.mjs test/client-retarget.test.mjs && node --test test/head-profile-acceptance.test.mjs test/head-promotion.test.mjs test/head-promotion-transfer.test.mjs test/head-transfer-acceptance.test.mjs && node --test test/cli-diagnostics.test.mjs && node --test test/service-install.test.mjs", "test:cluster": "node test/cluster.test.mjs && node --test test/cluster-availability.test.mjs test/cluster-readiness.test.mjs", "test": "npm run test:cluster && npm run test:unit && npm run test:entity-stagger && npm run smoke && npm run community:check", "test:entity-stagger": "node test/entity-stagger-benchmark.test.mjs", diff --git a/src/deployment-fence.mjs b/src/deployment-fence.mjs index cf7eb82..0cb265c 100644 --- a/src/deployment-fence.mjs +++ b/src/deployment-fence.mjs @@ -295,7 +295,7 @@ function fenceRecordShape(record) { return record; } -async function writeAtomicJson(filePath, value) { +export async function writeAtomicJson(filePath, value) { const directory = path.dirname(filePath); await fs.mkdir(directory, { recursive: true, mode: 0o700 }); const temporary = `${filePath}.${process.pid}.${randomUUID()}.tmp`; diff --git a/src/gateway-client.mjs b/src/gateway-client.mjs index eb6be0a..c27ef4a 100644 --- a/src/gateway-client.mjs +++ b/src/gateway-client.mjs @@ -114,6 +114,7 @@ export class LLooMGatewayError extends Error { if (Number.isInteger(options.status)) this.status = options.status; if (options.sslOnly !== undefined) this.sslOnly = options.sslOnly; if (options.upstreamCode !== undefined) this.upstreamCode = options.upstreamCode; + if (options.schedule) this.schedule = options.schedule; if (options.cause !== undefined) this.cause = options.cause; } } @@ -275,7 +276,26 @@ export async function gatewayRequest( 'RUNTIME_ADMISSION_BUSY' ]); const upstreamCode = allowedCodes.has(detail?.error?.code) ? detail.error.code : undefined; - throw new LLooMGatewayError(gatewayErrorMessage({ kind, status }), { kind, code, status, upstreamCode }); + // Schedule refusals ("pair is held by main") are the answer an agent + // needs, so they pass through: redacted (including the key just sent), + // without control characters, and bounded. + const schedule = + detail?.error?.type === 'schedule_error' && + typeof detail.error.message === 'string' && + /^schedule_[a-z_]{1,40}$/.test(String(detail.error.code ?? '')) + ? redactDiagnosticValue( + { + code: detail.error.code, + message: printable(detail.error.message, 500), + ...(plainDetails(detail.error.details) ? { details: detail.error.details } : {}) + }, + { config, env, extraValues: [adminKey] } + ) + : null; + throw new LLooMGatewayError( + schedule ? `${schedule.message} (HTTP ${status})` : gatewayErrorMessage({ kind, status }), + { kind, code, status, upstreamCode: upstreamCode ?? schedule?.code, schedule } + ); } return await response.json(); } catch (error) { @@ -296,6 +316,15 @@ export async function gatewayRequest( } } +function printable(text, max) { + const clean = text.replace(/[\u0000-\u001f\u007f-\u009f]/g, ' '); + return clean.length > max ? `${clean.slice(0, max - 1)}…` : clean; +} + +function plainDetails(details) { + return details && typeof details === 'object' && JSON.stringify(details).length <= 16384; +} + /** Gateway base URL resolved exactly the way CLI diagnostics expect. */ export function gatewayBaseUrl(config) { const configuredHost = config?.server?.host ?? '127.0.0.1'; diff --git a/src/schedule-cli.mjs b/src/schedule-cli.mjs new file mode 100644 index 0000000..9a25170 --- /dev/null +++ b/src/schedule-cli.mjs @@ -0,0 +1,388 @@ +// Agent-facing schedule commands: `lloom reserve | renew | release | schedule | job`. +// They talk to the gateway that owns the schedule — the local one by default, +// or the cluster head given by --head / LLOOM_HEAD_URL. +import { setTimeout as delay } from 'node:timers/promises'; +import { gatewayRequest, LLooMGatewayError } from './gateway-client.mjs'; + +const VALUE_FLAGS = new Set([ + '--owner', + '--purpose', + '--ttl', + '--session', + '--head', + '--admin-api-key-env', + '--on', + '--reservation', + '--note', + '--stale', + '--receipt', + '--unit', + '--poll', + '--config', + '--home' +]); +const RESERVATION_ID = /^r-[0-9a-f]{8}$/; +const ACTION_TIMEOUT_MS = 5 * 60 * 60 * 1000; + +export const SCHEDULE_COMMANDS = ['reserve', 'renew', 'release', 'schedule', 'job']; + +export function scheduleUsage() { + return `Reservations and jobs (on the schedule-owning gateway; --head URL or LLOOM_HEAD_URL selects a remote head): + lloom schedule [--all] [--json] Resources, holders, queues, and recent jobs + lloom reserve <resource> --owner NAME [--purpose TEXT] [--ttl 30m|none] [--wait] [--apply --yes] + lloom renew <reservation|resource> [--ttl 30m] Extend your lease (job reports also extend it) + lloom release <reservation|resource> [--force] [--apply --yes] Restore and free the resource + lloom job start <title> [--on resource] [--note TEXT] [--stale 15m] Prints the job id + lloom job note|progress|metric|beat|done|fail|cancel|show <job> ... + lloom job list [--all] + Owner and session default to LLOOM_OWNER and LLOOM_SESSION.`; +} + +/** Parse 90s / 30m / 2h / 1d (or "none" → null) to milliseconds. */ +export function parseDuration(value, flag) { + if (value === 'none') return null; + const match = /^(\d+(?:\.\d+)?)(ms|s|m|h|d)$/.exec(String(value ?? '').trim()); + if (!match) throw new Error(`${flag} needs a duration such as 90s, 30m, 2h or none`); + const scale = { ms: 1, s: 1000, m: 60_000, h: 3_600_000, d: 86_400_000 }[match[2]]; + return Math.round(Number(match[1]) * scale); +} + +function parseArgs(args) { + const words = []; + const flags = new Map(); + for (let index = 0; index < args.length; index += 1) { + const arg = args[index]; + if (!arg.startsWith('--')) words.push(arg); + else if (VALUE_FLAGS.has(arg)) flags.set(arg, args[++index]); + else flags.set(arg, true); + } + return { words, flag: (name) => flags.get(name), has: (name) => flags.get(name) === true }; +} + +export function ago(ms) { + const seconds = Math.max(0, Math.round(ms / 1000)); + if (seconds < 60) return `${seconds}s`; + if (seconds < 3600) return `${Math.round(seconds / 60)}m`; + if (seconds < 86400) return `${Math.round(seconds / 360) / 10}h`; + return `${Math.round(seconds / 8640) / 10}d`; +} + +function progressText(progress) { + if (!progress) return ''; + const unit = progress.unit ? ` ${progress.unit}` : ''; + return progress.total == null ? `[${progress.done}${unit}]` : `[${progress.done}/${progress.total}${unit}]`; +} + +function metricsText(metrics) { + return Object.entries(metrics ?? {}) + .map(([key, value]) => `${key}=${value}`) + .join(' '); +} + +export function formatReservation(entry, now) { + const left = + entry.state === 'active' ? (entry.expiresAt == null ? 'no expiry' : `${ago(entry.expiresAt - now)} left`) : null; + const where = entry.state === 'queued' ? `queued #${entry.position}` : entry.state; + const detail = [where, left].filter(Boolean).join(', '); + const why = entry.purpose ? ` — ${entry.purpose}` : ''; + const error = entry.error ? `\n ${entry.error}` : ''; + return `${entry.id} ${entry.resource} ${entry.owner} (${detail})${why}${error}`; +} + +export function formatJob(job) { + const when = job.state === 'running' ? `${ago(job.heartbeatAgeMs)} ago` : job.state; + const parts = [ + job.id, + job.health.padEnd(9), + when.padEnd(8), + job.owner, + `${job.title}${job.note ? ` — ${job.note}` : ''}`, + progressText(job.progress), + metricsText(job.metrics) + ]; + return parts.filter(Boolean).join(' '); +} + +export function formatBoard(board, { all = false } = {}) { + const byId = new Map(board.reservations.map((entry) => [entry.id, entry])); + const lines = ['Resources']; + for (const resource of board.resources) { + const hosts = resource.hosts?.length ? ` (${resource.hosts.join(', ')})` : ''; + const label = `${resource.id}${resource.description ? ` — ${resource.description}` : ''}${hosts}`; + if (resource.error) { + lines.push(` ${label}: config error: ${resource.error}`); + continue; + } + const holders = resource.holders.map((id) => byId.get(id)).filter(Boolean); + const status = holders.length + ? holders.map((entry) => formatReservation(entry, board.now)).join('\n ') + : `free${resource.capacity > 1 ? ` (0/${resource.capacity})` : ''}`; + lines.push(` ${label}\n ${status}`); + for (const id of resource.queue) lines.push(` next: ${formatReservation(byId.get(id), board.now)}`); + } + const recent = board.jobs.filter((job) => all || job.state === 'running' || board.now - job.endedAt < 86_400_000); + lines.push(recent.length ? 'Jobs' : 'Jobs: none'); + for (const job of recent) lines.push(` ${formatJob(job)}`); + return lines.join('\n'); +} + +/** Run one schedule command. Returns a process exit code. */ +export async function runScheduleCli( + command, + args, + { config, env = process.env, out = console.log, err = console.error, sleep = delay } +) { + const { words, flag, has } = parseArgs(args); + const json = has('--json'); + const head = flag('--head') ?? env.LLOOM_HEAD_URL; + const keyEnv = flag('--admin-api-key-env') ?? (head ? 'LLOOM_HEAD_API_KEY' : undefined); + // Never send the local gateway's admin key to a different host. + if (head && !env[keyEnv]) + throw new Error(`a remote head needs its admin key in ${keyEnv} (or --admin-api-key-env NAME)`); + const request = (pathname, { method = 'GET', body, timeoutMs = 30_000 } = {}) => + gatewayRequest(config, pathname, { + method, + body, + timeoutMs, + throwOnError: true, + env, + ...(keyEnv ? { explicitEnvName: keyEnv } : {}), + ...(head ? { urlFor: () => String(head).replace(/\/+$/, '') } : {}) + }); + const owner = () => { + const value = flag('--owner') ?? env.LLOOM_OWNER; + if (!value) throw new Error('--owner NAME (or LLOOM_OWNER) is required'); + return value; + }; + const session = flag('--session') ?? env.LLOOM_SESSION; + const print = (value, text) => out(json ? JSON.stringify(value, null, 2) : text); + const confirmed = has('--apply') && has('--yes'); + if (has('--apply') && !has('--yes')) throw new Error(`${command} requires --apply --yes`); + + async function resolveReservation(ref) { + if (!ref) throw new Error(`${command} needs a reservation id or resource name`); + if (RESERVATION_ID.test(ref)) return ref; + const board = await request('/gateway/schedule'); + const mine = board.reservations.find( + (entry) => entry.resource === ref && entry.owner === owner() && !['released', 'failed'].includes(entry.state) + ); + if (!mine) throw new Error(`${owner()} holds no reservation on ${ref}`); + return mine.id; + } + + if (command === 'schedule' || (command === 'reserve' && words.length < 2)) { + const board = await request('/gateway/schedule'); + print(board, formatBoard(board, { all: has('--all') })); + return 0; + } + + if (command === 'reserve') { + const body = { + resource: words[1], + owner: owner(), + purpose: flag('--purpose') ?? (words.slice(2).join(' ') || undefined), + session, + wait: has('--wait'), + apply: confirmed, + yes: confirmed + }; + if (flag('--ttl') !== undefined) body.ttlMs = parseDuration(flag('--ttl'), '--ttl'); + let entry = await request('/gateway/schedule/reservations', { method: 'POST', body, timeoutMs: ACTION_TIMEOUT_MS }); + if (entry.applied === false) { + const steps = entry.actions.map((action) => + action.kind === 'route' + ? `route ${action.alias}${action.node ? `@${action.node}` : ''} -> ${action.profile}` + : `suspend ${action.model}${action.node ? `@${action.node}` : ''}` + ); + const holders = entry.holders.map((holder) => formatReservation(holder, Date.now())); + print( + entry, + [ + `Would take ${entry.resource}${entry.description ? ` (${entry.description})` : ''} for ${entry.ttlMs == null ? 'no fixed time' : ago(entry.ttlMs)}`, + ` on take: ${steps.join('; ') || 'nothing'} (release undoes these in reverse)`, + holders.length ? ` held: ${holders.join('; ')}` : ' free now', + entry.queue.length ? ` queue: ${entry.queue.length}` : null, + ' re-run with --apply --yes' + ] + .filter(Boolean) + .join('\n') + ); + return 0; + } + if (body.wait && entry.state === 'queued') entry = await waitForTurn(entry, body.owner); + print(entry, formatReservation(entry, Date.now())); + return entry.state === 'active' ? 0 : entry.state === 'released' && entry.endReason === 'interrupted' ? 130 : 1; + } + + // Polls until the queued lease starts (or fails), keeping the queue slot + // renewed and riding out a head that is briefly unreachable. Interrupted, it + // gives the slot back so a later promotion never acts for nobody. + async function waitForTurn(start, holder) { + const poll = Math.max(1000, flag('--poll') === undefined ? 10_000 : parseDuration(flag('--poll'), '--poll')); + let entry = start; + let interrupted = false; + const interrupt = () => { + interrupted = true; + }; + process.on('SIGINT', interrupt); + process.on('SIGTERM', interrupt); + try { + let renewedAt = Date.now(); + let position = null; + while (['queued', 'acquiring'].includes(entry.state) && !interrupted) { + if (entry.state === 'queued' && entry.position !== position) { + position = entry.position; + err(`queued for ${entry.resource} (#${position})`); + } + await sleep(poll); + if (interrupted) break; + const slot = entry.queuedUntil != null && entry.updatedAt != null ? entry.queuedUntil - entry.updatedAt : 0; + const renewEvery = Math.max(poll, slot > 0 ? slot / 2 : 5 * 60_000); + try { + if (entry.state === 'queued' && Date.now() - renewedAt >= renewEvery) { + entry = await request(`/gateway/schedule/reservations/${entry.id}/renew`, { + method: 'POST', + body: { owner: holder } + }); + renewedAt = Date.now(); + } else entry = await request(`/gateway/schedule/reservations/${entry.id}`); + } catch (error) { + const transient = + error instanceof LLooMGatewayError && !['auth', 'authorization', 'http'].includes(error.kind); + if (!transient) throw error; + err(`still waiting; the head did not answer (${error.message})`); + } + } + if (interrupted) { + const given = await request(`/gateway/schedule/reservations/${entry.id}/release`, { + method: 'POST', + body: { owner: holder, apply: true, yes: true }, + timeoutMs: ACTION_TIMEOUT_MS + }).catch((error) => { + err(`could not give back ${entry.id}: ${error.message}`); + return entry; + }); + return { ...given, endReason: given.state === 'released' ? 'interrupted' : given.endReason }; + } + return entry; + } finally { + process.off('SIGINT', interrupt); + process.off('SIGTERM', interrupt); + } + } + + if (command === 'renew') { + const id = await resolveReservation(words[1]); + const body = { owner: owner() }; + if (flag('--ttl') !== undefined) body.ttlMs = parseDuration(flag('--ttl'), '--ttl'); + const entry = await request(`/gateway/schedule/reservations/${id}/renew`, { method: 'POST', body }); + print(entry, formatReservation(entry, Date.now())); + return 0; + } + + if (command === 'release') { + const id = await resolveReservation(words[1]); + const body = { owner: owner(), force: has('--force'), apply: confirmed, yes: confirmed }; + const entry = await request(`/gateway/schedule/reservations/${id}/release`, { + method: 'POST', + body, + timeoutMs: ACTION_TIMEOUT_MS + }); + if (entry.applied === false) { + const undo = entry.undo.map((step) => + step.kind === 'route' ? `route ${step.alias} -> ${step.previous ?? '(unchanged)'}` : `resume ${step.model}` + ); + print( + entry, + `Would release ${formatReservation(entry.reservation, Date.now())}\n restore: ${undo.join('; ') || 'nothing'}\n re-run with --apply --yes` + ); + return 0; + } + print(entry, formatReservation(entry, Date.now())); + return entry.state === 'released' ? 0 : 1; + } + + // job + const [, action, id, ...rest] = words; + const restText = rest.join(' ') || undefined; + // The owner, when known, lets the head refuse updates to someone else's job. + const update = (body) => + request(`/gateway/schedule/jobs/${encodeURIComponent(id ?? '')}`, { + method: 'POST', + body: { owner: flag('--owner') ?? env.LLOOM_OWNER, ...body } + }); + const requireId = () => { + if (!id) throw new Error(`lloom job ${action} needs a job id`); + }; + let job; + switch (action) { + case 'start': { + const title = words.slice(2).join(' '); + if (!title) throw new Error('lloom job start needs a title'); + const body = { title, owner: owner(), session, note: flag('--note') }; + const on = flag('--reservation') ?? flag('--on'); + if (on) body[RESERVATION_ID.test(on) ? 'reservationId' : 'resource'] = on; + if (flag('--stale') !== undefined) body.staleAfterMs = parseDuration(flag('--stale'), '--stale'); + job = await request('/gateway/schedule/jobs', { method: 'POST', body }); + // Plain output is just the id so scripts can capture it. + print(job, job.id); + return 0; + } + case 'list': { + const board = await request('/gateway/schedule'); + const jobs = board.jobs.filter((entry) => has('--all') || entry.state === 'running'); + print(jobs, jobs.length ? jobs.map(formatJob).join('\n') : 'no running jobs'); + return 0; + } + case 'show': + requireId(); + job = await request(`/gateway/schedule/jobs/${encodeURIComponent(id)}`); + print( + job, + [formatJob(job), ...job.notes.map((note) => ` ${new Date(note.at).toISOString()} ${note.text}`)].join('\n') + ); + return 0; + case 'note': + case 'beat': + requireId(); + job = await update({ note: action === 'note' ? (restText ?? flag('--note')) : flag('--note') }); + break; + case 'progress': { + requireId(); + const match = /^(\d+(?:\.\d+)?)(?:\/(\d+(?:\.\d+)?))?$/.exec(rest[0] ?? ''); + if (!match) throw new Error('lloom job progress <job> DONE[/TOTAL] [--unit NAME]'); + job = await update({ + progress: { done: Number(match[1]), total: match[2] == null ? null : Number(match[2]), unit: flag('--unit') }, + note: rest.slice(1).join(' ') || flag('--note') + }); + break; + } + case 'metric': { + requireId(); + if (!rest.length) throw new Error('lloom job metric <job> NAME=VALUE ...'); + const metrics = {}; + for (const pair of rest) { + const split = pair.indexOf('='); + if (split < 1) throw new Error(`metric ${pair} must be NAME=VALUE`); + const value = pair.slice(split + 1); + metrics[pair.slice(0, split)] = value !== '' && Number.isFinite(Number(value)) ? Number(value) : value; + } + job = await update({ metrics }); + break; + } + case 'done': + case 'fail': + case 'cancel': + requireId(); + job = await update({ + state: { done: 'done', fail: 'failed', cancel: 'cancelled' }[action], + note: restText ?? flag('--note'), + receipt: flag('--receipt') + }); + break; + default: + throw new Error(`unknown job command ${action ?? ''}\n${scheduleUsage()}`); + } + print(job, formatJob(job)); + return 0; +} diff --git a/src/schedule.mjs b/src/schedule.mjs new file mode 100644 index 0000000..11453f7 --- /dev/null +++ b/src/schedule.mjs @@ -0,0 +1,976 @@ +// Built-in hardware reservations and agent job log. +// +// One gateway (normally the cluster head) owns the schedule. Agents lease +// named resources from `config.reservations.resources`; a lease runs the +// resource's take actions (route aliases away, suspend models) and its release +// undoes them in reverse, finishing only after resumed models pass their +// health check. Leases expire unless renewed, so a vanished agent cannot hold +// hardware forever. Jobs are agent-posted status records (notes, progress, +// metrics, heartbeat) that a board such as Tessera reads instead of asking each +// agent how it is going. +import { randomBytes } from 'node:crypto'; +import fsSync from 'node:fs'; +import fs from 'node:fs/promises'; +import path from 'node:path'; +import { writeAtomicJson } from './deployment-fence.mjs'; + +export const SCHEDULE_SCHEMA_VERSION = 1; +const MINUTE = 60_000; +export const SCHEDULE_DEFAULTS = Object.freeze({ + ttlMs: 30 * MINUTE, + maxTtlMs: 8 * 60 * MINUTE, + queueTtlMs: 30 * MINUTE, + warnBeforeMs: 5 * MINUTE, + staleAfterMs: 15 * MINUTE, + retentionMs: 7 * 24 * 60 * MINUTE, + drainTimeoutMs: 5 * MINUTE, + sweepMs: 15_000, + eventLimit: 500 +}); + +// Held states keep the resource unavailable to others. `stuck` means a +// restore failed or the gateway restarted mid-action: the hardware is not +// known to be back, so it stays held until a release succeeds. +const HELD = new Set(['acquiring', 'active', 'releasing', 'stuck']); +const OPEN = new Set([...HELD, 'queued']); +const JOB_STATES = new Set(['running', 'done', 'failed', 'cancelled']); +const NAME = /^[A-Za-z0-9][A-Za-z0-9._:@/+-]{0,127}$/; +const NOTE_HISTORY = 20; +const METRIC_LIMIT = 32; +const RUNNING_JOB_LIMIT = 500; +const QUEUE_LIMIT = 50; +// A running job silent this long is abandoned rather than kept stalled forever. +const ABANDON_AFTER_MS = 24 * 60 * MINUTE; +const EVENT_LOG_ROTATE_BYTES = 16 * 1024 * 1024; + +export class ScheduleError extends Error { + constructor(message, { code = 'schedule_error', statusCode = 400, details } = {}) { + super(message); + this.name = 'ScheduleError'; + this.code = code; + this.type = 'schedule_error'; + this.statusCode = statusCode; + if (details !== undefined) this.details = details; + } +} + +function plainObject(value) { + return value && typeof value === 'object' && !Array.isArray(value) ? value : null; +} + +function name(value, field) { + const text = String(value ?? '').trim(); + if (!NAME.test(text)) { + throw new ScheduleError(`${field} is required and must be a short identifier (letters, digits, ._:@/+-)`, { + code: 'schedule_invalid_request' + }); + } + return text; +} + +function text(value, field, max, { required = false } = {}) { + if (value == null || value === '') { + if (required) throw new ScheduleError(`${field} is required`, { code: 'schedule_invalid_request' }); + return null; + } + const result = String(value).trim(); + if (result.length > max) { + throw new ScheduleError(`${field} must be at most ${max} characters`, { code: 'schedule_invalid_request' }); + } + return result || null; +} + +function duration(value, field, { min = 1000, max = Infinity, allowNull = false } = {}) { + if (value === null && allowNull) return null; + const ms = Number(value); + if (!Number.isInteger(ms) || ms < min || ms > max) { + throw new ScheduleError(`${field} must be an integer number of milliseconds from ${min} to ${max}`, { + code: 'schedule_invalid_request' + }); + } + return ms; +} + +function newId(prefix, taken) { + let id; + do id = `${prefix}-${randomBytes(4).toString('hex')}`; + while (Object.hasOwn(taken, id)); + return id; +} + +function settings(config) { + const raw = plainObject(config?.reservations) ?? {}; + const out = {}; + for (const [key, fallback] of Object.entries(SCHEDULE_DEFAULTS)) { + const value = Number(raw[key]); + out[key] = Number.isInteger(value) && value > 0 ? value : fallback; + } + // Model maintenance refuses longer drains. + out.drainTimeoutMs = Math.min(out.drainTimeoutMs, 2 * 60 * 60 * 1000); + return out; +} + +function nodeOf(entry) { + return entry.node == null ? null : name(entry.node, 'node'); +} + +/** Normalize one configured resource. Throws a ScheduleError for an invalid definition. */ +export function normalizeResource(id, raw, config = {}) { + const resource = plainObject(raw); + if (!resource) throw new ScheduleError(`reservation resource ${id} must be an object`, { code: 'schedule_config' }); + const defaults = settings(config); + const capacity = resource.capacity == null ? 1 : Number(resource.capacity); + if (!Number.isInteger(capacity) || capacity < 1) { + throw new ScheduleError(`reservation resource ${id}: capacity must be a positive integer`, { + code: 'schedule_config' + }); + } + const suspend = (resource.suspend ?? []).map((entry) => { + const value = typeof entry === 'string' ? { model: entry } : plainObject(entry); + if (!value) throw new ScheduleError(`reservation resource ${id}: bad suspend entry`, { code: 'schedule_config' }); + return { model: name(value.model, 'suspend.model'), node: nodeOf(value) }; + }); + const routes = (resource.routes ?? []).map((entry) => { + const value = plainObject(entry); + if (!value) throw new ScheduleError(`reservation resource ${id}: bad route entry`, { code: 'schedule_config' }); + return { + alias: name(value.alias, 'routes.alias'), + profile: name(value.profile, 'routes.profile'), + node: nodeOf(value) + }; + }); + // Take actions change shared serving state, which only one holder can own. + if (capacity > 1 && (suspend.length || routes.length)) { + throw new ScheduleError(`reservation resource ${id}: suspend/routes require capacity 1`, { + code: 'schedule_config' + }); + } + const onExpiry = resource.onExpiry ?? 'release'; + if (!['release', 'keep'].includes(onExpiry)) { + throw new ScheduleError(`reservation resource ${id}: onExpiry must be release or keep`, { + code: 'schedule_config' + }); + } + return { + id, + description: text(resource.description, 'description', 500), + hosts: Array.isArray(resource.hosts) ? resource.hosts.map((host) => name(host, 'hosts')) : [], + capacity, + ttlMs: resource.ttlMs == null ? defaults.ttlMs : duration(resource.ttlMs, 'ttlMs'), + maxTtlMs: resource.maxTtlMs == null ? defaults.maxTtlMs : duration(resource.maxTtlMs, 'maxTtlMs'), + allowIndefinite: resource.allowIndefinite === true, + onExpiry, + // Routes move traffic before suspend drains the model; release reverses both. + actions: [ + ...routes.map((route) => ({ kind: 'route', ...route })), + ...suspend.map((entry) => ({ kind: 'suspend', ...entry })) + ] + }; +} + +function emptyState() { + return { + schemaVersion: SCHEDULE_SCHEMA_VERSION, + seq: 0, + reservations: Object.create(null), + jobs: Object.create(null), + events: [] + }; +} + +function readState(statePath, now) { + let raw; + try { + raw = JSON.parse(fsSync.readFileSync(statePath, 'utf8')); + } catch (error) { + if (error?.code === 'ENOENT') return emptyState(); + throw error; + } + if (raw?.schemaVersion !== SCHEDULE_SCHEMA_VERSION) { + throw new Error(`unsupported schedule state schema ${raw?.schemaVersion} in ${statePath}`); + } + const state = { ...emptyState(), ...raw }; + state.reservations = Object.assign(Object.create(null), raw.reservations); + state.jobs = Object.assign(Object.create(null), raw.jobs); + // An action cannot survive a restart. Its outcome is unknown, so hold the + // resource until someone releases it (release re-runs the idempotent undo). + for (const reservation of Object.values(state.reservations)) { + if (reservation.state === 'acquiring' || reservation.state === 'releasing') { + reservation.error = `gateway restarted while ${reservation.state}; release to restore`; + reservation.state = 'stuck'; + reservation.updatedAt = now; + } + } + return state; +} + +/** + * Create the schedule. `actions` performs take/undo steps: + * `activeRoute({alias,node})`, `setRoute({alias,profile,node})`, + * `suspend({model,node,drainTimeoutMs})`, `resume({model,node})`. + */ +export function createSchedule({ + getConfig, + statePath, + eventsPath = null, + actions, + now = Date.now, + logger = console, + isPaused = () => false +}) { + if (!statePath) throw new Error('schedule requires a state path'); + const state = readState(statePath, now()); + let queue = Promise.resolve(); + let timer = null; + let closed = false; + let pausedSince = null; + let logWrites = Promise.resolve(); + let logBytes = null; + const running = new Set(); + + function open() { + if (closed) throw new ScheduleError('the gateway is closing', { code: 'schedule_closed', statusCode: 503 }); + } + + function serialize(fn) { + const result = queue.then(fn); + queue = result.catch(() => {}); + return result; + } + + async function persist() { + await writeAtomicJson(statePath, state); + } + + function emit(kind, { severity = 'info', message, reservation, job, ring = true } = {}) { + const event = { + seq: ++state.seq, + at: now(), + kind, + severity, + message, + ...(reservation + ? { reservationId: reservation.id, resource: reservation.resource, owner: reservation.owner } + : {}), + ...(job ? { jobId: job.id, owner: job.owner, ...(job.resource ? { resource: job.resource } : {}) } : {}) + }; + if (ring) { + state.events.push(event); + const limit = settings(getConfig()).eventLimit; + if (state.events.length > limit) state.events.splice(0, state.events.length - limit); + } + if (eventsPath) { + const line = `${JSON.stringify(event)}\n`; + logWrites = logWrites + .then(async () => { + if (logBytes == null) { + await fs.mkdir(path.dirname(eventsPath), { recursive: true, mode: 0o700 }); + logBytes = (await fs.stat(eventsPath).catch(() => null))?.size ?? 0; + } + if (logBytes > EVENT_LOG_ROTATE_BYTES) { + await fs.rename(eventsPath, `${eventsPath}.1`); + logBytes = 0; + } + await fs.appendFile(eventsPath, line, { mode: 0o600 }); + logBytes += Buffer.byteLength(line); + }) + .catch((error) => logger.warn?.(`schedule event log write failed: ${error?.message ?? error}`)); + } + return event; + } + + function resources() { + const config = getConfig(); + const specs = Object.entries(plainObject(config?.reservations?.resources) ?? {}).map(([id, raw]) => { + try { + return normalizeResource(id, raw, config); + } catch (error) { + return { id, error: error.message }; + } + }); + // Two resources that suspend the same model or move the same route would + // undo each other's leases, so neither is usable until one is changed. + const claims = new Map(); + for (const spec of specs) { + for (const action of spec.actions ?? []) { + const key = `${action.kind}:${action.node ?? ''}:${action.model ?? action.alias}`; + claims.set(key, [...(claims.get(key) ?? []), spec]); + } + } + for (const [key, owners] of claims) { + if (owners.length < 2) continue; + for (const spec of owners) { + spec.error ??= `reservation resources ${owners.map((owner) => owner.id).join(', ')} both control ${key.split(':').slice(2).join(':')}`; + } + } + return specs; + } + + function resource(id) { + const found = resources().find((entry) => entry.id === id); + if (!found) { + throw new ScheduleError(`unknown reservation resource ${id}`, { + code: 'schedule_unknown_resource', + statusCode: 404 + }); + } + if (found.error) throw new ScheduleError(found.error, { code: 'schedule_config', statusCode: 500 }); + return found; + } + + function reservationsFor(resourceId, states) { + return Object.values(state.reservations) + .filter((entry) => entry.resource === resourceId && states.has(entry.state)) + .sort((a, b) => a.createdAt - b.createdAt); + } + + function reservation(id) { + const found = Object.hasOwn(state.reservations, id) ? state.reservations[id] : null; + if (!found) + throw new ScheduleError(`unknown reservation ${id}`, { code: 'schedule_unknown_reservation', statusCode: 404 }); + return found; + } + + function job(id) { + const found = Object.hasOwn(state.jobs, id) ? state.jobs[id] : null; + if (!found) throw new ScheduleError(`unknown job ${id}`, { code: 'schedule_unknown_job', statusCode: 404 }); + return found; + } + + function assertOwner(entry, owner, force) { + if (force === true || entry.owner === owner) return; + throw new ScheduleError(`reservation ${entry.id} is held by ${entry.owner}`, { + code: 'schedule_not_owner', + statusCode: 409, + details: { owner: entry.owner } + }); + } + + function ttlFor(spec, requested) { + if (requested === undefined) return spec.ttlMs; + if (requested === null) { + if (!spec.allowIndefinite) { + throw new ScheduleError(`resource ${spec.id} does not allow indefinite reservations`, { + code: 'schedule_invalid_request' + }); + } + return null; + } + return duration(requested, 'ttlMs', { min: 60_000, max: spec.maxTtlMs }); + } + + function view(entry) { + if (!entry) return null; + const time = now(); + const out = { ...entry, steps: entry.steps.map((step) => ({ ...step })) }; + if (entry.state === 'active' && entry.expiresAt != null) out.remainingMs = Math.max(0, entry.expiresAt - time); + if (entry.state === 'queued') { + out.position = reservationsFor(entry.resource, new Set(['queued'])).findIndex((item) => item.id === entry.id) + 1; + } + return out; + } + + function jobView(entry) { + const out = { ...entry, notes: [...entry.notes] }; + out.heartbeatAgeMs = Math.max(0, now() - entry.heartbeatAt); + out.health = entry.state !== 'running' ? entry.state : entry.stalledAt != null ? 'stalled' : 'ok'; + return out; + } + + // ---- take / undo ------------------------------------------------------- + + async function runTake(entry, spec) { + const { drainTimeoutMs } = settings(getConfig()); + try { + for (const action of spec.actions) { + if (action.kind === 'route') { + const previous = (await actions.activeRoute(action)) ?? null; + // Without an active profile there is nothing exact to restore. + if (previous == null) { + throw new Error(`route ${action.alias} has no active profile to restore; select one first`); + } + entry.steps.push({ ...action, previous, status: 'pending' }); + await serialize(persist); + if (previous !== action.profile) await actions.setRoute(action); + } else { + // A model someone already suspended stays suspended after release. + const wasSuspended = (await actions.isSuspended?.(action)) === true; + entry.steps.push({ ...action, wasSuspended, status: 'pending' }); + await serialize(persist); + if (!wasSuspended) await actions.suspend({ ...action, drainTimeoutMs }); + } + entry.steps.at(-1).status = 'done'; + } + } catch (error) { + const message = error?.message ?? String(error); + const undoError = await runUndo(entry); + return serialize(async () => { + entry.updatedAt = now(); + if (!undoError) entry.endedAt = entry.updatedAt; + entry.error = undoError ? `${message}; restore failed: ${undoError}` : message; + entry.state = undoError ? 'stuck' : 'failed'; + emit(undoError ? 'reservation.stuck' : 'reservation.failed', { + severity: 'error', + message: `${spec.id} take failed: ${entry.error}`, + reservation: entry + }); + await persist(); + if (!undoError) promote(spec.id); + return view(entry); + }); + } + return serialize(async () => { + const time = now(); + entry.state = 'active'; + entry.startedAt = entry.updatedAt = time; + entry.expiresAt = entry.ttlMs == null ? null : time + entry.ttlMs; + emit('reservation.active', { + message: `${entry.owner} holds ${spec.id}${entry.purpose ? `: ${entry.purpose}` : ''}`, + reservation: entry + }); + await persist(); + return view(entry); + }); + } + + /** Undo every recorded step in reverse. Returns an error message, or null. */ + async function runUndo(entry) { + for (const step of [...entry.steps].reverse()) { + if (step.status === 'undone') continue; + try { + if (step.kind === 'suspend') { + if (!step.wasSuspended) await actions.resume(step); + } else if (step.previous && step.previous !== step.profile) { + await actions.setRoute({ alias: step.alias, profile: step.previous, node: step.node }); + } + step.status = 'undone'; + } catch (error) { + step.error = error?.message ?? String(error); + await serialize(persist).catch(() => {}); + return `${step.kind} ${step.model ?? step.alias}: ${step.error}`; + } + } + return null; + } + + function track(promise) { + running.add(promise); + promise.finally(() => running.delete(promise)).catch(() => {}); + return promise; + } + + // Start queued reservations while capacity allows. Called inside serialize. + function promote(resourceId) { + let spec; + try { + spec = resource(resourceId); + } catch { + return; + } + if (isPaused()) return; + const free = spec.capacity - reservationsFor(resourceId, HELD).length; + for (const next of reservationsFor(resourceId, new Set(['queued'])).slice(0, Math.max(0, free))) { + next.state = 'acquiring'; + next.updatedAt = now(); + delete next.queuedUntil; + emit('reservation.acquiring', { message: `${next.owner} is next on ${resourceId}`, reservation: next }); + track(runTake(next, spec)); + } + } + + // ---- public operations ------------------------------------------------- + + function preview(body) { + const spec = resource(name(body.resource, 'resource')); + return { + applied: false, + resource: spec.id, + description: spec.description, + hosts: spec.hosts, + ttlMs: ttlFor(spec, body.ttlMs), + actions: spec.actions, + holders: reservationsFor(spec.id, HELD).map(view), + queue: reservationsFor(spec.id, new Set(['queued'])).map(view) + }; + } + + async function take(body = {}) { + open(); + if (body.apply !== true) return preview(body); + if (body.yes !== true) throw new ScheduleError('take requires apply and yes', { code: 'schedule_confirm' }); + const spec = resource(name(body.resource, 'resource')); + const owner = name(body.owner, 'owner'); + const purpose = text(body.purpose, 'purpose', 500); + const session = text(body.session, 'session', 200); + const ttlMs = ttlFor(spec, body.ttlMs); + const started = await serialize(async () => { + // Taking again is idempotent for the same holder (owner, and session when + // both name one): renew instead of queueing twice. + const mine = reservationsFor(spec.id, OPEN).find( + (entry) => entry.owner === owner && (!session || !entry.session || entry.session === session) + ); + if (mine) return { existing: mine }; + const held = reservationsFor(spec.id, HELD); + const queued = reservationsFor(spec.id, new Set(['queued'])); + if (queued.length >= QUEUE_LIMIT) { + throw new ScheduleError(`${spec.id} already has ${QUEUE_LIMIT} waiting`, { + code: 'schedule_queue_full', + statusCode: 429 + }); + } + const free = held.length < spec.capacity && queued.length === 0; + if (!free && body.wait !== true) { + throw new ScheduleError(`${spec.id} is held by ${held.map((entry) => entry.owner).join(', ') || 'the queue'}`, { + code: 'schedule_busy', + statusCode: 409, + details: { holders: held.map(view), queue: queued.map(view) } + }); + } + const time = now(); + const entry = { + id: newId('r', state.reservations), + resource: spec.id, + owner, + purpose, + session, + state: free ? 'acquiring' : 'queued', + ttlMs, + createdAt: time, + updatedAt: time, + startedAt: null, + expiresAt: null, + endedAt: null, + endReason: null, + warnedAt: null, + error: null, + steps: [] + }; + if (!free) entry.queuedUntil = time + settings(getConfig()).queueTtlMs; + state.reservations[entry.id] = entry; + emit(free ? 'reservation.acquiring' : 'reservation.queued', { + message: free ? `${owner} is taking ${spec.id}` : `${owner} queued for ${spec.id}`, + reservation: entry + }); + await persist(); + return { entry, free }; + }); + if (started.existing) { + // Mid-take, mid-release or stuck: report it as it is rather than renew. + if (!['active', 'queued'].includes(started.existing.state)) return view(started.existing); + return renew(started.existing.id, { owner, ttlMs: body.ttlMs }); + } + if (!started.free) return view(started.entry); + return track(runTake(started.entry, spec)); + } + + function renew(id, body = {}) { + open(); + return serialize(async () => { + const entry = reservation(id); + assertOwner(entry, body.owner, body.force); + const time = now(); + if (entry.state === 'queued') { + entry.queuedUntil = time + settings(getConfig()).queueTtlMs; + } else if (entry.state === 'active') { + if (body.ttlMs !== undefined) entry.ttlMs = ttlFor(resource(entry.resource), body.ttlMs); + entry.expiresAt = entry.ttlMs == null ? null : time + entry.ttlMs; + entry.warnedAt = null; + } else { + throw new ScheduleError(`reservation ${id} is ${entry.state}`, { code: 'schedule_not_open', statusCode: 409 }); + } + entry.updatedAt = time; + await persist(); + return view(entry); + }); + } + + async function release(id, body = {}, { reason = 'released', system = false } = {}) { + const entry = reservation(id); + if (!system) { + open(); + if (body.apply !== true) return { applied: false, reservation: view(entry), undo: [...entry.steps].reverse() }; + if (body.yes !== true) throw new ScheduleError('release requires apply and yes', { code: 'schedule_confirm' }); + } + const begin = await serialize(async () => { + if (!system) assertOwner(entry, body.owner, body.force); + if (entry.state === 'queued') { + entry.state = 'released'; + entry.endedAt = entry.updatedAt = now(); + entry.endReason = reason; + emit('reservation.withdrawn', { + message: `${entry.owner} left the ${entry.resource} queue`, + reservation: entry + }); + await persist(); + return { done: view(entry) }; + } + if (entry.state === 'acquiring' || entry.state === 'releasing') { + throw new ScheduleError(`reservation ${id} is ${entry.state}; wait for it to settle`, { + code: 'schedule_busy', + statusCode: 409 + }); + } + if (!HELD.has(entry.state)) return { done: view(entry) }; + // A renewal may land between the sweep and this release. + if (reason === 'expired' && (entry.state !== 'active' || entry.expiresAt == null || entry.expiresAt > now())) { + return { done: view(entry) }; + } + entry.state = 'releasing'; + entry.updatedAt = now(); + entry.endReason = reason; + await persist(); + return { run: true }; + }); + if (begin.done) return begin.done; + const undoError = await runUndo(entry); + return serialize(async () => { + entry.updatedAt = now(); + if (undoError) { + entry.state = 'stuck'; + entry.error = `restore failed: ${undoError}`; + emit('reservation.stuck', { + severity: 'error', + message: `${entry.resource} restore failed (${entry.owner}): ${undoError}`, + reservation: entry + }); + } else { + entry.state = 'released'; + entry.endedAt = entry.updatedAt; + entry.error = null; + emit(reason === 'expired' ? 'reservation.expired' : 'reservation.released', { + severity: reason === 'expired' ? 'attention' : 'info', + message: + reason === 'expired' + ? `${entry.owner}'s ${entry.resource} lease expired; restored` + : `${entry.owner} released ${entry.resource}`, + reservation: entry + }); + promote(entry.resource); + } + await persist(); + return view(entry); + }); + } + + function startJob(body = {}) { + open(); + return serialize(async () => { + const owner = name(body.owner, 'owner'); + if (Object.values(state.jobs).filter((entry) => entry.state === 'running').length >= RUNNING_JOB_LIMIT) { + throw new ScheduleError(`at most ${RUNNING_JOB_LIMIT} jobs run at once; end some first`, { + code: 'schedule_too_many_jobs', + statusCode: 429 + }); + } + let reservationId = body.reservationId == null ? null : name(body.reservationId, 'reservationId'); + let resourceId = body.resource == null ? null : name(body.resource, 'resource'); + if (reservationId) { + const lease = reservation(reservationId); + // Its reports keep the lease alive, so only the holder's jobs attach. + assertOwner(lease, owner, body.force); + resourceId = lease.resource; + } else if (resourceId) { + // Attach to the owner's open lease on that resource when there is one. + reservationId = reservationsFor(resourceId, OPEN).find((entry) => entry.owner === owner)?.id ?? null; + } + const time = now(); + const entry = { + id: newId('j', state.jobs), + title: text(body.title, 'title', 200, { required: true }), + owner, + session: text(body.session, 'session', 200), + reservationId, + resource: resourceId, + state: 'running', + startedAt: time, + heartbeatAt: time, + endedAt: null, + staleAfterMs: + body.staleAfterMs == null + ? settings(getConfig()).staleAfterMs + : duration(body.staleAfterMs, 'staleAfterMs', { min: 60_000, max: 24 * 60 * MINUTE }), + stalledAt: null, + note: null, + notes: [], + progress: null, + metrics: {}, + receipt: null + }; + applyJobUpdate(entry, body, time); + state.jobs[entry.id] = entry; + emit('job.started', { message: `${owner} started: ${entry.title}`, job: entry }); + await persist(); + return jobView(entry); + }); + } + + /** Validates the whole update before changing anything, so a refused update leaves no trace. */ + function applyJobUpdate(entry, body, time) { + const note = text(body.note, 'note', 2000); + let progress; + if (body.progress !== undefined) { + const raw = plainObject(body.progress); + const done = Number(raw?.done); + const total = raw?.total == null ? null : Number(raw.total); + if (!raw || !Number.isFinite(done) || done < 0 || (total != null && (!Number.isFinite(total) || total <= 0))) { + throw new ScheduleError('progress needs a non-negative done and an optional positive total', { + code: 'schedule_invalid_request' + }); + } + progress = { done, total, unit: text(raw.unit, 'progress.unit', 40) }; + } + let metrics; + if (body.metrics !== undefined) { + const raw = plainObject(body.metrics); + if (!raw) throw new ScheduleError('metrics must be an object', { code: 'schedule_invalid_request' }); + metrics = { ...entry.metrics }; + for (const [key, value] of Object.entries(raw)) { + name(key, 'metric name'); + if ( + !(typeof value === 'number' && Number.isFinite(value)) && + !(typeof value === 'string' && value.length <= 200) + ) { + throw new ScheduleError(`metric ${key} must be a finite number or a short string`, { + code: 'schedule_invalid_request' + }); + } + metrics[key] = value; + } + if (Object.keys(metrics).length > METRIC_LIMIT) { + throw new ScheduleError(`a job keeps at most ${METRIC_LIMIT} metrics`, { code: 'schedule_invalid_request' }); + } + } + const receipt = text(body.receipt, 'receipt', 500); + if (note) { + entry.note = note; + entry.notes.push({ at: time, text: note }); + if (entry.notes.length > NOTE_HISTORY) entry.notes.splice(0, entry.notes.length - NOTE_HISTORY); + emit('job.note', { message: note, job: entry, ring: false }); + } + if (progress) entry.progress = progress; + if (metrics) entry.metrics = metrics; + if (receipt) entry.receipt = receipt; + } + + function updateJob(id, body = {}) { + open(); + return serialize(async () => { + const entry = job(id); + if (body.owner != null && body.owner !== entry.owner && body.force !== true) { + throw new ScheduleError(`job ${id} belongs to ${entry.owner}`, { code: 'schedule_not_owner', statusCode: 409 }); + } + if (entry.state !== 'running') { + throw new ScheduleError(`job ${id} already ended (${entry.state})`, { + code: 'schedule_not_open', + statusCode: 409 + }); + } + const nextState = body.state ?? 'running'; + if (typeof nextState !== 'string' || !JOB_STATES.has(nextState)) { + throw new ScheduleError('state must be running, done, failed or cancelled', { + code: 'schedule_invalid_request' + }); + } + const time = now(); + applyJobUpdate(entry, body, time); + entry.heartbeatAt = time; + if (entry.stalledAt != null) { + entry.stalledAt = null; + emit('job.recovered', { message: `${entry.title} is reporting again`, job: entry }); + } + if (nextState !== 'running') { + entry.state = nextState; + entry.endedAt = time; + emit(`job.${nextState}`, { + severity: nextState === 'failed' ? 'error' : nextState === 'done' ? 'attention' : 'info', + message: `${entry.title}${entry.note ? `: ${entry.note}` : ''}`, + job: entry + }); + } + // A heartbeating job proves its holder is alive, so it keeps its lease. + const lease = + entry.reservationId && Object.hasOwn(state.reservations, entry.reservationId) + ? state.reservations[entry.reservationId] + : null; + if (nextState === 'running' && lease?.state === 'active' && lease.ttlMs != null) { + const until = time + lease.ttlMs; + if (until > lease.expiresAt) { + lease.expiresAt = until; + lease.warnedAt = null; + lease.updatedAt = time; + } + } + await persist(); + return jobView(entry); + }); + } + + function board({ since = 0 } = {}) { + const after = Number(since) || 0; + const reservations = Object.values(state.reservations).sort((a, b) => b.createdAt - a.createdAt); + return { + schemaVersion: SCHEDULE_SCHEMA_VERSION, + now: now(), + seq: state.seq, + resources: resources().map((spec) => ({ + ...spec, + holders: spec.error ? [] : reservationsFor(spec.id, HELD).map((entry) => entry.id), + queue: spec.error ? [] : reservationsFor(spec.id, new Set(['queued'])).map((entry) => entry.id) + })), + reservations: reservations.map(view), + jobs: Object.values(state.jobs) + .sort((a, b) => b.startedAt - a.startedAt) + .map(jobView), + events: state.events.filter((event) => event.seq > after) + }; + } + + // ---- sweep ------------------------------------------------------------- + + async function sweep() { + const expired = await serialize(async () => { + const time = now(); + const config = settings(getConfig()); + const due = []; + let changed = false; + // A deployment fence blocks renewals and actions, so leases neither + // expire nor start while it holds; afterwards they get the time back. + if (isPaused()) { + pausedSince ??= time; + return due; + } + if (pausedSince != null) { + for (const entry of Object.values(state.reservations)) { + if (entry.state === 'active' && entry.expiresAt != null) entry.expiresAt += time - pausedSince; + if (entry.state === 'queued') entry.queuedUntil += time - pausedSince; + } + pausedSince = null; + changed = true; + } + for (const entry of Object.values(state.reservations)) { + if (entry.state === 'queued' && entry.queuedUntil <= time) { + entry.state = 'released'; + entry.endedAt = entry.updatedAt = time; + entry.endReason = 'queue-timeout'; + emit('reservation.withdrawn', { + message: `${entry.owner} dropped from the ${entry.resource} queue (not renewed)`, + reservation: entry + }); + changed = true; + } else if (entry.state === 'active' && entry.expiresAt != null) { + let onExpiry = 'release'; + try { + onExpiry = resource(entry.resource).onExpiry; + } catch { + // A resource removed from config still expires and restores. + } + if (entry.expiresAt <= time && onExpiry === 'release') due.push(entry.id); + else if (entry.warnedAt == null && entry.expiresAt - time <= config.warnBeforeMs) { + entry.warnedAt = time; + emit(entry.expiresAt <= time ? 'reservation.overdue' : 'reservation.expiring', { + severity: 'attention', + message: `${entry.owner}'s ${entry.resource} lease ${entry.expiresAt <= time ? 'is overdue' : 'expires soon'}; renew to keep it`, + reservation: entry + }); + changed = true; + } + } + } + for (const entry of Object.values(state.jobs)) { + if (entry.state === 'running' && time - entry.heartbeatAt > ABANDON_AFTER_MS) { + entry.state = 'failed'; + entry.endedAt = time; + entry.note = `abandoned: no report for ${Math.round((time - entry.heartbeatAt) / (60 * MINUTE))} h`; + emit('job.failed', { severity: 'error', message: `${entry.title}: ${entry.note}`, job: entry }); + changed = true; + continue; + } + if (entry.state === 'running' && entry.stalledAt == null && time - entry.heartbeatAt > entry.staleAfterMs) { + entry.stalledAt = time; + emit('job.stalled', { + severity: 'attention', + message: `${entry.title}: no report for ${Math.round((time - entry.heartbeatAt) / MINUTE)} min`, + job: entry + }); + changed = true; + } + } + for (const [id, entry] of Object.entries(state.reservations)) { + if (!OPEN.has(entry.state) && entry.endedAt != null && time - entry.endedAt > config.retentionMs) { + delete state.reservations[id]; + changed = true; + } + } + for (const [id, entry] of Object.entries(state.jobs)) { + if (entry.state !== 'running' && time - entry.endedAt > config.retentionMs) { + delete state.jobs[id]; + changed = true; + } + } + for (const spec of resources()) if (!spec.error) promote(spec.id); + if (changed) await persist(); + return due; + }); + for (const id of expired) { + await release(id, {}, { reason: 'expired', system: true }).catch((error) => { + // Someone else's release got there first. + if (error?.code !== 'schedule_busy') + logger.error?.(`schedule expiry release ${id} failed: ${error?.message ?? error}`); + }); + } + } + + return { + board, + take, + renew, + release: (id, body) => release(id, body), + reservation: (id) => view(reservation(id)), + startJob, + updateJob, + job: (id) => jobView(job(id)), + sweep: () => (closed ? Promise.resolve() : track(sweep())), + start() { + if (timer) return; + timer = setInterval(() => { + track(sweep()).catch((error) => logger.error?.(`schedule sweep failed: ${error?.message ?? error}`)); + }, settings(getConfig()).sweepMs); + timer.unref?.(); + }, + /** Refuses new work and waits a little for actions in flight; any still running are + * recovered as `stuck` on the next start, since every step is recorded before it runs. */ + async stop({ waitMs = 5000 } = {}) { + closed = true; + if (timer) clearInterval(timer); + timer = null; + let timeout; + await Promise.race([ + Promise.allSettled([...running]).then(() => queue), + new Promise((resolve) => { + timeout = setTimeout(resolve, waitMs); + timeout.unref?.(); + }) + ]); + clearTimeout(timeout); + await logWrites; + } + }; +} + +/** Default state and event log locations beside a file-backed config. */ +export function schedulePaths(config) { + const configured = plainObject(config?.reservations) ?? {}; + const root = config?.sourcePath ? path.dirname(config.sourcePath) : null; + const resolve = (value) => (root && !path.isAbsolute(value) ? path.join(root, value) : value); + const statePath = configured.statePath + ? resolve(configured.statePath) + : root + ? path.join(root, 'schedule.json') + : null; + const eventsPath = configured.eventsPath + ? resolve(configured.eventsPath) + : statePath + ? statePath.replace(/\.json$/, '') + '-events.ndjson' + : null; + return { statePath, eventsPath }; +} diff --git a/src/server.mjs b/src/server.mjs index 12ff2f8..882dee5 100644 --- a/src/server.mjs +++ b/src/server.mjs @@ -63,6 +63,7 @@ import { createRegistry, UnknownModelError } from './registry.mjs'; import { routeProfileStatus, writeRouteMemberSuspension, writeRouteProfile } from './route-control.mjs'; import { mutateConfigSource } from './config-mutation.mjs'; import { createModelMaintenanceController } from './model-maintenance-control.mjs'; +import { createSchedule, ScheduleError, schedulePaths } from './schedule.mjs'; import { acquireRateLimitSlot, awaitRateBudget, @@ -486,13 +487,14 @@ function numberOrNull(value) { return Number.isFinite(number) ? number : null; } -function errorBody(message, { type = 'invalid_request_error', code = 'error', model } = {}) { +function errorBody(message, { type = 'invalid_request_error', code = 'error', model, details } = {}) { return { error: { message, type, code, - model + model, + ...(details === undefined ? {} : { details }) } }; } @@ -2610,6 +2612,73 @@ export function createLloomServer( }); const dashboardInstallation = createDashboardInstallation({ getConfig: () => config, reload: reloadConfig }); const fleetProfiles = createFleetProfileController({ getConfig: () => config, reload: reloadConfig }); + let scheduleUnavailable = 'this gateway does not keep a schedule (needs a file-backed primary config)'; + const schedule = createGatewaySchedule(); + + // Reservation take/undo steps. A step names the node whose gateway owns the + // model or route; other nodes are driven through their admin API. + function createGatewaySchedule() { + const { statePath, eventsPath } = schedulePaths(config); + if (!statePath || processRole !== 'primary' || isStandby(config)) return null; + async function maintenance(model, action, node, drainTimeoutMs = 300000, apply = true) { + const body = { apply, yes: apply, drainTimeoutMs }; + const result = clusterCoordinator.isLocalNode(node) + ? await modelMaintenance.run(model, action, { ...body, requestedBy: currentNodeId(config) }) + : await clusterCoordinator.requestNode(node, `/gateway/models/${encodeURIComponent(model)}/${action}`, { + method: 'POST', + body, + timeoutMs: 4 * 60 * 60 * 1000 + }); + if (!apply) return result; + const expected = action === 'suspend' ? 'suspended' : 'ready'; + if (result?.status !== expected) throw new Error(`${action} ${model} returned ${result?.status ?? 'no status'}`); + return result; + } + try { + return createSchedule({ + getConfig: () => config, + statePath, + eventsPath, + logger, + isPaused: () => deploymentFence.isFenced(), + actions: { + // The dry-run plan reports each runtime's maintenance record. + async isSuspended({ model, node }) { + const plan = await maintenance(model, 'suspend', node, undefined, false); + return Object.values(plan?.state ?? {}).some(Boolean); + }, + async activeRoute({ alias, node }) { + const routes = clusterCoordinator.isLocalNode(node) + ? routeProfileStatus(config, alias) + : (await clusterCoordinator.requestNode(node, '/gateway/routing', { timeoutMs: 10000 })).routes; + return routes?.find((route) => route.alias === alias)?.activeRoute ?? null; + }, + async setRoute({ alias, profile, node }) { + if (!clusterCoordinator.isLocalNode(node)) { + return clusterCoordinator.requestNode(node, `/gateway/routes/${encodeURIComponent(alias)}/profile`, { + method: 'POST', + body: { profile }, + timeoutMs: 30000 + }); + } + const result = await writeRouteProfile(config, alias, profile); + if (result.changed) { + reloadConfig(); + await reloadInFlight; + } + return result; + }, + suspend: ({ model, node, drainTimeoutMs }) => maintenance(model, 'suspend', node, drainTimeoutMs), + resume: ({ model, node }) => maintenance(model, 'resume', node) + } + }); + } catch (error) { + // A damaged or newer schedule file must not take inference down with it. + scheduleUnavailable = `schedule unavailable: ${error?.message ?? error}`; + logger.error?.(scheduleUnavailable); + return null; + } + } async function routingStatus() { const cacheMs = Math.max(0, Number(config.cluster?.routingStatusCacheMs ?? 250)); @@ -4692,6 +4761,42 @@ export function createLloomServer( ); } + async function handleSchedule(req, res, url) { + if (!schedule) throw new ScheduleError(scheduleUnavailable, { code: 'schedule_unavailable', statusCode: 503 }); + let parts; + try { + parts = url.pathname.replace(/\/+$/, '').split('/').slice(3).map(decodeURIComponent); + } catch { + throw new ScheduleError('malformed schedule path', { code: 'schedule_invalid_request' }); + } + const [collection, id, verb] = parts; + const body = req.method === 'POST' ? await readJson(req) : {}; + let result; + if (req.method === 'GET' && parts.length === 0) { + result = schedule.board({ since: url.searchParams.get('since') }); + } else if (collection === 'reservations' && req.method === 'POST' && !id) { + result = await schedule.take(body); + } else if (collection === 'reservations' && id && req.method === 'GET' && !verb) { + result = schedule.reservation(id); + } else if (collection === 'reservations' && id && req.method === 'POST' && verb === 'renew') { + result = await schedule.renew(id, body); + } else if (collection === 'reservations' && id && req.method === 'POST' && verb === 'release') { + result = await schedule.release(id, body); + } else if (collection === 'jobs' && req.method === 'POST' && !id) { + result = await schedule.startJob(body); + } else if (collection === 'jobs' && id && !verb && req.method === 'GET') { + result = schedule.job(id); + } else if (collection === 'jobs' && id && !verb && req.method === 'POST') { + result = await schedule.updateJob(id, body); + } else { + throw new ScheduleError(`no schedule route for ${req.method} ${url.pathname}`, { + code: 'not_found', + statusCode: 404 + }); + } + sendJson(res, 200, result); + } + async function handleRequest(req, res) { setCors(res, config); if (req.method === 'OPTIONS') { @@ -4949,6 +5054,11 @@ export function createLloomServer( return; } + if (url.pathname === '/gateway/schedule' || url.pathname.startsWith('/gateway/schedule/')) { + await handleSchedule(req, res, url); + return; + } + const maintenanceMatch = url.pathname.match(/^\/gateway\/models\/([^/]+)\/(suspend|resume)$/); if (maintenanceMatch && req.method === 'POST') { const body = await readJson(req); @@ -5833,6 +5943,14 @@ export function createLloomServer( ); return; } + if (error instanceof ScheduleError && !res.headersSent) { + sendJson( + res, + error.statusCode, + errorBody(error.message, { type: error.type, code: error.code, details: error.details }) + ); + return; + } if (error instanceof DeploymentFenceError) { if (res.headersSent) { endResponseWithError(res, error, { @@ -5940,6 +6058,7 @@ export function createLloomServer( }) .catch((error) => logger.error?.(error)); if (configPath && !isStandby(config)) watchFile(configPath, { interval: 500 }, reloadConfig); + schedule?.start(); resolve(server); }); }); @@ -5956,6 +6075,7 @@ export function createLloomServer( runtimeManager.cancelQueuedRequests?.(closingError); runtimeManager.cancelAdmissionOperations?.(closingError); rateLimitRegistry.close(); + await schedule?.stop(); unsubscribeScheduling(); unsubscribeLimits(); await reloadInFlight; diff --git a/test/schedule-integration.test.mjs b/test/schedule-integration.test.mjs new file mode 100644 index 0000000..2947533 --- /dev/null +++ b/test/schedule-integration.test.mjs @@ -0,0 +1,355 @@ +import test from 'node:test'; +import assert from 'node:assert/strict'; +import http from 'node:http'; +import { execFile } from 'node:child_process'; +import { promisify } from 'node:util'; +import { fileURLToPath } from 'node:url'; +import fs from 'node:fs/promises'; +import os from 'node:os'; +import path from 'node:path'; +import { loadConfig } from '../src/config.mjs'; +import { createLloomServer } from '../src/server.mjs'; +import { RuntimeManager } from '../src/runtime-manager.mjs'; +import { assertMaintenanceStartAllowed } from '../src/model-maintenance.mjs'; + +const quiet = { info() {}, warn() {}, error() {} }; +const cli = fileURLToPath(new URL('../bin/lloom.mjs', import.meta.url)); + +async function listen(server) { + await new Promise((resolve) => server.listen(0, '127.0.0.1', resolve)); + return `http://127.0.0.1:${server.address().port}`; +} + +async function readBody(req) { + let text = ''; + for await (const chunk of req) text += chunk; + return text ? JSON.parse(text) : {}; +} + +// A stand-in owner gateway for a remote node: records admin calls. +async function remoteGateway(t) { + const calls = []; + const state = { route: 'local-atlas', failResume: false, suspended: false }; + const server = http.createServer(async (req, res) => { + const body = await readBody(req); + calls.push({ method: req.method, url: req.url, auth: req.headers.authorization, body }); + res.setHeader('content-type', 'application/json'); + if (req.url === '/gateway/routing') { + res.end(JSON.stringify({ routes: [{ alias: 'omp', activeRoute: state.route }] })); + } else if (req.url === '/gateway/routes/omp/profile') { + state.route = body.profile; + res.end(JSON.stringify({ changed: true, alias: 'omp', activeRoute: body.profile })); + } else if (req.url === '/gateway/models/pair-model/suspend') { + // A dry run reports the runtime's current maintenance record. + res.end( + JSON.stringify( + body.apply + ? { applied: true, status: 'suspended' } + : { applied: false, state: { 'pair-runtime': state.suspended ? { state: 'suspended' } : null } } + ) + ); + } else if (req.url === '/gateway/models/pair-model/resume') { + if (state.failResume) { + res.statusCode = 500; + res.end(JSON.stringify({ error: { message: 'runtime failed health check after resume' } })); + } else res.end(JSON.stringify({ applied: true, status: 'ready' })); + } else { + res.statusCode = 404; + res.end('{}'); + } + }); + const url = await listen(server); + t.after(() => new Promise((resolve) => server.close(resolve))); + return { url, calls, state }; +} + +async function fixture(t) { + const dir = await fs.mkdtemp(path.join(os.tmpdir(), 'lloom-schedule-http-')); + const remote = await remoteGateway(t); + const raw = { + server: { host: '127.0.0.1', port: 0 }, + security: { apiKeys: ['synthetic-inference'], adminApiKeys: ['synthetic-admin'] }, + logging: { metricsPersistence: false }, + runtimePolicy: { enabled: false }, + backends: { test: { type: 'openai', baseUrl: 'http://127.0.0.1:9/v1' } }, + runtimes: { resident: { enabled: true, keepWarm: false } }, + models: [ + { id: 'local', backend: 'test', runtime: 'resident', upstreamModel: 'local' }, + { id: 'cloud', backend: 'test', upstreamModel: 'cloud' } + ], + aliases: { + chat: { + members: ['local', 'cloud'], + activeRoute: 'local', + routeProfiles: { local: { members: ['local', 'cloud'] }, cloud: { members: ['cloud'] } } + } + }, + cluster: { + nodeId: 'head', + nodes: { + head: { endpoint: 'http://127.0.0.1:1' }, + pairhost: { endpoint: remote.url, apiKey: 'remote-admin' } + } + }, + reservations: { + resources: { + gpu: { description: 'local GPU', routes: [{ alias: 'chat', profile: 'cloud' }], suspend: ['local'] }, + pair: { + hosts: ['pairhost'], + routes: [{ alias: 'omp', profile: 'cloud', node: 'pairhost' }], + suspend: [{ model: 'pair-model', node: 'pairhost' }] + } + } + } + }; + const sourcePath = path.join(dir, 'config.json'); + await fs.writeFile(sourcePath, JSON.stringify(raw)); + const config = await loadConfig(sourcePath); + const manager = new RuntimeManager(config, { logger: quiet }); + let healthy = true; + manager.status = async () => ({ runtimes: { resident: { status: healthy ? 'running' : 'stopped', healthy } } }); + manager.isHealthy = async () => healthy; + manager.runtimeAppearsLoaded = async () => healthy; + manager.ensure = async (id) => assertMaintenanceStartAllowed(manager.config, id); + manager.stop = async (id) => { + healthy = false; + manager.setStatus(id, 'stopped'); + }; + manager.admit = async () => { + healthy = true; + }; + const app = createLloomServer(config, { runtimeManager: manager, logger: quiet }); + const url = await listen(app.server); + t.after(async () => { + await app.close({ stopRuntimes: false, httpGraceMs: 100 }); + await fs.rm(dir, { recursive: true, force: true }); + }); + const call = async (method, route, body, key = 'synthetic-admin') => { + const response = await fetch(url + route, { + method, + headers: { 'content-type': 'application/json', ...(key ? { authorization: `Bearer ${key}` } : {}) }, + body: body === undefined ? undefined : JSON.stringify(body) + }); + return { status: response.status, json: await response.json() }; + }; + return { call, sourcePath, remote, dir, url, isHealthy: () => healthy }; +} + +test('HTTP reservations route away, suspend, and restore through the local gateway', { timeout: 15000 }, async (t) => { + const f = await fixture(t); + assert.equal((await f.call('GET', '/gateway/schedule', undefined, null)).status, 401); + assert.equal((await f.call('GET', '/gateway/schedule', undefined, 'synthetic-inference')).status, 401); + + const preview = await f.call('POST', '/gateway/schedule/reservations', { resource: 'gpu', owner: 'main' }); + assert.equal(preview.status, 200); + assert.equal(preview.json.applied, false); + + const taken = await f.call('POST', '/gateway/schedule/reservations', { + resource: 'gpu', + owner: 'main', + purpose: 'kernel A/B', + apply: true, + yes: true + }); + assert.equal(taken.status, 200, JSON.stringify(taken.json)); + assert.equal(taken.json.state, 'active'); + let source = JSON.parse(await fs.readFile(f.sourcePath, 'utf8')); + assert.equal(source.aliases.chat.activeRoute, 'cloud'); + assert.equal(source.runtimes.resident.maintenance.state, 'suspended'); + assert.equal(f.isHealthy(), false); + + const busy = await f.call('POST', '/gateway/schedule/reservations', { + resource: 'gpu', + owner: 'other', + apply: true, + yes: true + }); + assert.equal(busy.status, 409); + assert.equal(busy.json.error.code, 'schedule_busy'); + assert.equal(busy.json.error.details.holders[0].owner, 'main'); + + const job = await f.call('POST', '/gateway/schedule/jobs', { + title: 'kernel A/B', + owner: 'main', + resource: 'gpu', + session: 'claude:abc' + }); + assert.equal(job.json.reservationId, taken.json.id); + const progressed = await f.call('POST', `/gateway/schedule/jobs/${job.json.id}`, { + note: 'arm 1 done', + progress: { done: 1, total: 2 }, + metrics: { tok_s: 51.5 } + }); + assert.equal(progressed.json.progress.total, 2); + const board = await f.call('GET', '/gateway/schedule'); + assert.deepEqual(board.json.resources.find((entry) => entry.id === 'gpu').holders, [taken.json.id]); + assert.equal(board.json.jobs[0].note, 'arm 1 done'); + + const released = await f.call('POST', `/gateway/schedule/reservations/${taken.json.id}/release`, { + owner: 'main', + apply: true, + yes: true + }); + assert.equal(released.status, 200, JSON.stringify(released.json)); + assert.equal(released.json.state, 'released'); + source = JSON.parse(await fs.readFile(f.sourcePath, 'utf8')); + assert.equal(source.aliases.chat.activeRoute, 'local'); + assert.equal(source.runtimes.resident.maintenance, undefined); + assert.equal(f.isHealthy(), true); + assert.equal((await f.call('GET', '/gateway/schedule/reservations/r-missing')).status, 404); + const persisted = JSON.parse(await fs.readFile(path.join(f.dir, 'schedule.json'), 'utf8')); + assert.equal(persisted.reservations[taken.json.id].state, 'released'); +}); + +test('remote steps go through the owner node admin API; a failed resume stays held', { timeout: 15000 }, async (t) => { + const f = await fixture(t); + const body = { resource: 'pair', owner: 'main', apply: true, yes: true }; + const taken = await f.call('POST', '/gateway/schedule/reservations', body); + assert.equal(taken.json.state, 'active', JSON.stringify(taken.json)); + assert.deepEqual( + f.remote.calls.map((entry) => `${entry.method} ${entry.url}`), + [ + 'GET /gateway/routing', + 'POST /gateway/routes/omp/profile', + 'POST /gateway/models/pair-model/suspend', + 'POST /gateway/models/pair-model/suspend' + ] + ); + assert.deepEqual( + f.remote.calls.slice(2).map((entry) => entry.body.apply), + [false, true] + ); + assert.ok(f.remote.calls.every((entry) => entry.auth === 'Bearer remote-admin')); + assert.equal(f.remote.state.route, 'cloud'); + + f.remote.state.failResume = true; + const route = `/gateway/schedule/reservations/${taken.json.id}/release`; + const stuck = await f.call('POST', route, { owner: 'main', apply: true, yes: true }); + assert.equal(stuck.json.state, 'stuck'); + assert.equal(f.remote.state.route, 'cloud'); + f.remote.state.failResume = false; + const released = await f.call('POST', route, { owner: 'main', apply: true, yes: true }); + assert.equal(released.json.state, 'released'); + assert.equal(f.remote.state.route, 'local-atlas'); +}); + +test('agents drive the schedule from the CLI against a remote head', { timeout: 30000 }, async (t) => { + const f = await fixture(t); + const home = await fs.mkdtemp(path.join(os.tmpdir(), 'lloom-schedule-cli-')); + t.after(() => fs.rm(home, { recursive: true, force: true })); + const env = { + ...process.env, + HOME: home, + LLOOM_HEAD_URL: f.url, + LLOOM_OWNER: 'main', + LLOOM_HEAD_API_KEY: 'synthetic-admin' + }; + delete env.LLOOM_CONFIG; + delete env.LLOOM_ADMIN_API_KEY; + const lloom = async (args, extra = {}) => { + try { + const { stdout, stderr } = await promisify(execFile)(process.execPath, [cli, ...args], { + env: { ...env, ...extra } + }); + return { code: 0, stdout, stderr }; + } catch (error) { + return { code: error.code, stdout: error.stdout, stderr: error.stderr }; + } + }; + + const unkeyed = await lloom(['schedule'], { LLOOM_HEAD_API_KEY: '' }); + assert.notEqual(unkeyed.code, 0); + assert.match(unkeyed.stderr, /LLOOM_HEAD_API_KEY/); + + const preview = await lloom(['reserve', 'gpu', '--purpose', 'kernel A/B']); + assert.equal(preview.code, 0, preview.stderr); + assert.match(preview.stdout, /Would take gpu .*\n {2}on take: route chat -> cloud; suspend local/); + + const taken = await lloom(['reserve', 'gpu', '--purpose', 'kernel A/B', '--ttl', '45m', '--apply', '--yes']); + assert.equal(taken.code, 0, taken.stderr); + assert.match(taken.stdout, /^r-[0-9a-f]{8} {2}gpu {2}main \(active, 45m left\) — kernel A\/B/); + + const busy = await lloom(['reserve', 'gpu', '--apply', '--yes'], { LLOOM_OWNER: 'buildopt' }); + assert.notEqual(busy.code, 0); + assert.match(busy.stderr, /gpu is held by main \(HTTP 409\)/); + + const started = await lloom(['job', 'start', 'kernel', 'A/B', '--on', 'gpu', '--note', 'building']); + const jobId = started.stdout.trim(); + assert.match(jobId, /^j-[0-9a-f]{8}$/); + assert.equal((await lloom(['job', 'progress', jobId, '2/5', '--unit', 'arms', 'arm', '2'])).code, 0); + assert.equal((await lloom(['job', 'metric', jobId, 'tok_s=58.2', 'image=f399caa'])).code, 0); + const board = await lloom(['schedule']); + assert.match(board.stdout, /gpu — local GPU\n {4}r-[0-9a-f]{8} {2}gpu {2}main \(active, 45m left\)/); + assert.match(board.stdout, new RegExp(`${jobId} ok .*kernel A/B — arm 2 \\[2/5 arms\\] tok_s=58.2 image=f399caa`)); + const done = await lloom(['job', 'done', jobId, 'NLL', 'flat', '--receipt', 'receipts/ab.json']); + assert.match(done.stdout, /done .*kernel A\/B — NLL flat/); + const json = JSON.parse((await lloom(['job', 'show', jobId, '--json'])).stdout); + assert.equal(json.receipt, 'receipts/ab.json'); + assert.deepEqual(json.metrics, { tok_s: 58.2, image: 'f399caa' }); + + assert.equal((await lloom(['renew', 'gpu', '--ttl', '1h'])).code, 0); + const released = await lloom(['release', 'gpu', '--apply', '--yes']); + assert.equal(released.code, 0, released.stderr); + assert.match(released.stdout, /gpu {2}main \(released\)/); + assert.equal(f.isHealthy(), true); + const nothing = await lloom(['release', 'gpu', '--apply', '--yes']); + assert.match(nothing.stderr, /main holds no reservation on gpu/); +}); + +test('reserve --wait queues, then returns once the holder releases', { timeout: 30000 }, async (t) => { + const f = await fixture(t); + const env = { ...process.env, LLOOM_HEAD_URL: f.url, LLOOM_HEAD_API_KEY: 'synthetic-admin', LLOOM_OWNER: 'buildopt' }; + const held = await f.call('POST', '/gateway/schedule/reservations', { + resource: 'gpu', + owner: 'main', + apply: true, + yes: true + }); + const waiter = promisify(execFile)( + process.execPath, + [cli, 'reserve', 'gpu', '--wait', '--poll', '1s', '--apply', '--yes'], + { env } + ); + let queued = []; + for (let attempt = 0; attempt < 50 && !queued.length; attempt += 1) { + await new Promise((resolve) => setTimeout(resolve, 100)); + queued = (await f.call('GET', '/gateway/schedule')).json.resources.find((entry) => entry.id === 'gpu').queue; + } + assert.equal(queued.length, 1); + await f.call('POST', `/gateway/schedule/reservations/${held.json.id}/release`, { + owner: 'main', + apply: true, + yes: true + }); + const { stdout, stderr } = await waiter; + assert.match(stderr, /queued for gpu \(#1\)/); + assert.match(stdout, /gpu {2}buildopt \(active/); +}); + +test( + 'a head cannot echo the key it was sent, or terminal escapes, back through the CLI', + { timeout: 15000 }, + async (t) => { + const head = http.createServer((req, res) => { + res.writeHead(409, { 'content-type': 'application/json' }); + res.end( + JSON.stringify({ + error: { + type: 'schedule_error', + code: 'schedule_busy', + message: `pair is held by ${req.headers.authorization} \u001b[31mred\u001b[0m` + } + }) + ); + }); + const url = await listen(head); + t.after(() => new Promise((resolve) => head.close(resolve))); + const env = { ...process.env, LLOOM_HEAD_URL: url, LLOOM_HEAD_API_KEY: 'head-secret-key', LLOOM_OWNER: 'main' }; + const failed = await promisify(execFile)(process.execPath, [cli, 'reserve', 'pair', '--apply', '--yes'], { + env + }).catch((error) => error); + assert.notEqual(failed.code, 0); + assert.match(failed.stderr, /pair is held by Bearer \[redacted\]/); + assert.doesNotMatch(failed.stderr, /head-secret-key|\u001b/); + } +); diff --git a/test/schedule.test.mjs b/test/schedule.test.mjs new file mode 100644 index 0000000..a9f1777 --- /dev/null +++ b/test/schedule.test.mjs @@ -0,0 +1,375 @@ +import test from 'node:test'; +import assert from 'node:assert/strict'; +import fs from 'node:fs/promises'; +import os from 'node:os'; +import path from 'node:path'; +import { createSchedule, normalizeResource, ScheduleError } from '../src/schedule.mjs'; + +const MINUTE = 60_000; + +function fakeActions({ failResume = 0, failSuspend = false } = {}) { + const calls = []; + const routes = { omp: 'local-atlas' }; + return { + calls, + routes, + failResume, + async activeRoute({ alias, node }) { + calls.push(['activeRoute', alias, node]); + return routes[alias] ?? null; + }, + async setRoute({ alias, profile, node }) { + calls.push(['setRoute', alias, profile, node]); + routes[alias] = profile; + }, + async suspend({ model, node }) { + calls.push(['suspend', model, node]); + if (failSuspend) throw new Error('drain timed out'); + }, + async resume({ model, node }) { + calls.push(['resume', model, node]); + if (this.failResume > 0) { + this.failResume -= 1; + throw new Error('runtime failed health check after resume'); + } + } + }; +} + +const RESOURCES = { + pair: { + description: 'two-node pair', + hosts: ['node-a', 'node-b'], + routes: [{ alias: 'omp', profile: 'cloud', node: 'node-a' }], + suspend: [{ model: 'pair-model', node: 'node-a' }] + }, + builds: { capacity: 2 }, + drafter: { allowIndefinite: true } +}; + +async function fixture(t, { actions = fakeActions(), resources = RESOURCES, statePath } = {}) { + const dir = statePath ? path.dirname(statePath) : await fs.mkdtemp(path.join(os.tmpdir(), 'lloom-schedule-')); + t.after(() => fs.rm(dir, { recursive: true, force: true })); + const clock = { now: 1_800_000_000_000 }; + const config = { reservations: { resources } }; + const options = { + getConfig: () => config, + statePath: statePath ?? path.join(dir, 'schedule.json'), + eventsPath: path.join(dir, 'schedule-events.ndjson'), + actions, + now: () => clock.now, + logger: { warn() {}, error() {} } + }; + const schedule = createSchedule(options); + t.after(() => schedule.stop()); + return { schedule, actions, clock, config, options, dir }; +} + +const take = (schedule, body) => schedule.take({ apply: true, yes: true, ...body }); +const release = (schedule, id, body) => schedule.release(id, { apply: true, yes: true, ...body }); +const kinds = (schedule) => schedule.board().events.map((event) => event.kind); + +test('take previews by default, then routes away before suspending', async (t) => { + const { schedule, actions } = await fixture(t); + const plan = await schedule.take({ resource: 'pair', owner: 'main' }); + assert.equal(plan.applied, false); + assert.deepEqual( + plan.actions.map((action) => action.kind), + ['route', 'suspend'] + ); + assert.equal(actions.calls.length, 0); + + const held = await take(schedule, { resource: 'pair', owner: 'main', purpose: 'decode A/B' }); + assert.equal(held.state, 'active'); + assert.equal(held.expiresAt - held.startedAt, 30 * MINUTE); + assert.deepEqual(actions.calls, [ + ['activeRoute', 'omp', 'node-a'], + ['setRoute', 'omp', 'cloud', 'node-a'], + ['suspend', 'pair-model', 'node-a'] + ]); + assert.equal(held.steps[0].previous, 'local-atlas'); + assert.ok(kinds(schedule).includes('reservation.active')); +}); + +test('a busy resource refuses, or queues and promotes on release', async (t) => { + const { schedule, actions } = await fixture(t); + const first = await take(schedule, { resource: 'pair', owner: 'main' }); + await assert.rejects(take(schedule, { resource: 'pair', owner: 'buildopt' }), (error) => { + assert.ok(error instanceof ScheduleError); + assert.equal(error.code, 'schedule_busy'); + assert.equal(error.statusCode, 409); + assert.equal(error.details.holders[0].owner, 'main'); + return true; + }); + const waiting = await take(schedule, { resource: 'pair', owner: 'buildopt', wait: true }); + assert.equal(waiting.state, 'queued'); + assert.equal(waiting.position, 1); + + actions.calls.length = 0; + const released = await release(schedule, first.id, { owner: 'main' }); + assert.equal(released.state, 'released'); + // Undo runs in reverse: resume (health-checked) before routing traffic back. + assert.deepEqual(actions.calls.slice(0, 2), [ + ['resume', 'pair-model', 'node-a'], + ['setRoute', 'omp', 'local-atlas', 'node-a'] + ]); + await schedule.stop(); + const promoted = schedule.reservation(waiting.id); + assert.equal(promoted.state, 'active'); + assert.equal(promoted.steps[0].previous, 'local-atlas'); +}); + +test('only the owner releases unless forced; taking twice renews', async (t) => { + const { schedule, clock } = await fixture(t); + const held = await take(schedule, { resource: 'pair', owner: 'main' }); + clock.now += 10 * MINUTE; + const again = await take(schedule, { resource: 'pair', owner: 'main' }); + assert.equal(again.id, held.id); + assert.equal(again.expiresAt, clock.now + 30 * MINUTE); + await assert.rejects(release(schedule, held.id, { owner: 'other' }), { code: 'schedule_not_owner' }); + const preview = await schedule.release(held.id, { owner: 'other' }); + assert.equal(preview.applied, false); + assert.equal((await release(schedule, held.id, { owner: 'other', force: true })).state, 'released'); +}); + +test('a failed restore keeps the resource held until a release succeeds', async (t) => { + const actions = fakeActions({ failResume: 1 }); + const { schedule } = await fixture(t, { actions }); + const held = await take(schedule, { resource: 'pair', owner: 'main' }); + const stuck = await release(schedule, held.id, { owner: 'main' }); + assert.equal(stuck.state, 'stuck'); + assert.match(stuck.error, /health check/); + // Traffic was not routed back to the unhealthy model. + assert.equal(actions.routes.omp, 'cloud'); + await assert.rejects(take(schedule, { resource: 'pair', owner: 'buildopt' }), { code: 'schedule_busy' }); + assert.ok(kinds(schedule).includes('reservation.stuck')); + const retried = await release(schedule, held.id, { owner: 'main' }); + assert.equal(retried.state, 'released'); + assert.equal(actions.routes.omp, 'local-atlas'); +}); + +test('a failed take rolls back what it did and frees the resource', async (t) => { + const actions = fakeActions({ failSuspend: true }); + const { schedule } = await fixture(t, { actions }); + const failed = await take(schedule, { resource: 'pair', owner: 'main' }); + assert.equal(failed.state, 'failed'); + assert.match(failed.error, /drain timed out/); + assert.equal(actions.routes.omp, 'local-atlas'); + actions.calls.length = 0; + actions.failResume = 0; + const failSuspend = actions.suspend; + actions.suspend = async (args) => actions.calls.push(['suspend', args.model, args.node]); + assert.equal((await take(schedule, { resource: 'pair', owner: 'buildopt' })).state, 'active'); + actions.suspend = failSuspend; +}); + +test('leases warn, then expire and restore; renewals and job heartbeats extend them', async (t) => { + const { schedule, clock, actions } = await fixture(t); + const held = await take(schedule, { resource: 'pair', owner: 'main', ttlMs: 20 * MINUTE }); + clock.now += 10 * MINUTE; + const job = await schedule.startJob({ title: 'decode sweep', owner: 'main', resource: 'pair' }); + assert.equal(job.reservationId, held.id); + await schedule.updateJob(job.id, { progress: { done: 1, total: 4, unit: 'arms' } }); + assert.equal(schedule.reservation(held.id).expiresAt, clock.now + 20 * MINUTE); + + clock.now += 16 * MINUTE; + await schedule.sweep(); + assert.ok(kinds(schedule).includes('reservation.expiring')); + await schedule.renew(held.id, { owner: 'main' }); + assert.equal(schedule.reservation(held.id).expiresAt, clock.now + 20 * MINUTE); + + clock.now += 21 * MINUTE; + actions.calls.length = 0; + await schedule.sweep(); + assert.equal(schedule.reservation(held.id).state, 'released'); + assert.equal(schedule.reservation(held.id).endReason, 'expired'); + assert.deepEqual(actions.calls[0], ['resume', 'pair-model', 'node-a']); + assert.ok(kinds(schedule).includes('reservation.expired')); +}); + +test('jobs carry notes, progress and metrics; silence marks them stalled', async (t) => { + const { schedule, clock, dir } = await fixture(t); + const job = await schedule.startJob({ + title: 'FP4 NLL receipts', + owner: 'glm-fp4l', + session: 'claude:abc123', + note: 'building', + staleAfterMs: 10 * MINUTE + }); + assert.equal(job.health, 'ok'); + await schedule.updateJob(job.id, { note: 'arm 2 of 5', metrics: { tok_s: 58.2, image: 'f399caa' } }); + clock.now += 11 * MINUTE; + await schedule.sweep(); + assert.equal(schedule.job(job.id).health, 'stalled'); + await schedule.updateJob(job.id, { note: 'back' }); + assert.equal(schedule.job(job.id).health, 'ok'); + const done = await schedule.updateJob(job.id, { state: 'done', receipt: 'receipts/fp4l.json', note: 'NLL +0.002' }); + assert.equal(done.state, 'done'); + assert.deepEqual( + done.notes.map((note) => note.text), + ['building', 'arm 2 of 5', 'back', 'NLL +0.002'] + ); + assert.equal(done.metrics.tok_s, 58.2); + await assert.rejects(schedule.updateJob(job.id, { note: 'late' }), { code: 'schedule_not_open' }); + const events = schedule.board().events.map((event) => event.kind); + assert.deepEqual(events, ['job.started', 'job.stalled', 'job.recovered', 'job.done']); + // Notes reach the durable log but not the board's event ring. + await new Promise((resolve) => setTimeout(resolve, 20)); + const log = await fs.readFile(path.join(dir, 'schedule-events.ndjson'), 'utf8'); + assert.equal( + log + .trim() + .split('\n') + .filter((line) => JSON.parse(line).kind === 'job.note').length, + 4 + ); +}); + +test('state survives restart and an interrupted action stays held', async (t) => { + const first = await fixture(t); + const held = await take(first.schedule, { resource: 'pair', owner: 'main' }); + await first.schedule.stop(); + const raw = JSON.parse(await fs.readFile(first.options.statePath, 'utf8')); + raw.reservations[held.id].state = 'releasing'; + await fs.writeFile(first.options.statePath, JSON.stringify(raw)); + + const second = createSchedule(first.options); + t.after(() => second.stop()); + const recovered = second.reservation(held.id); + assert.equal(recovered.state, 'stuck'); + assert.match(recovered.error, /restarted/); + assert.equal((await release(second, held.id, { owner: 'main' })).state, 'released'); +}); + +test('queue entries expire unless renewed; board lists events after a cursor', async (t) => { + const { schedule, clock } = await fixture(t); + await take(schedule, { resource: 'pair', owner: 'main', ttlMs: 2 * 60 * MINUTE }); + const waiting = await take(schedule, { resource: 'pair', owner: 'buildopt', wait: true }); + const cursor = schedule.board().seq; + clock.now += 20 * MINUTE; + await schedule.renew(waiting.id, { owner: 'buildopt' }); + clock.now += 20 * MINUTE; + await schedule.sweep(); + assert.equal(schedule.reservation(waiting.id).state, 'queued'); + clock.now += 31 * MINUTE; + await schedule.sweep(); + assert.equal(schedule.reservation(waiting.id).endReason, 'queue-timeout'); + const board = schedule.board({ since: cursor }); + assert.ok(board.events.every((event) => event.seq > cursor)); + assert.deepEqual( + board.events.map((event) => event.kind), + ['reservation.withdrawn'] + ); + const pair = board.resources.find((entry) => entry.id === 'pair'); + assert.equal(pair.holders.length, 1); + assert.equal(pair.queue.length, 0); +}); + +test('capacity, indefinite leases and resource validation', async (t) => { + const { schedule } = await fixture(t); + await take(schedule, { resource: 'builds', owner: 'a' }); + await take(schedule, { resource: 'builds', owner: 'b' }); + await assert.rejects(take(schedule, { resource: 'builds', owner: 'c' }), { code: 'schedule_busy' }); + await assert.rejects(take(schedule, { resource: 'pair', owner: 'a', ttlMs: null }), { + code: 'schedule_invalid_request' + }); + const forever = await take(schedule, { resource: 'drafter', owner: 'drafter', ttlMs: null }); + assert.equal(forever.expiresAt, null); + await assert.rejects(take(schedule, { resource: 'nope', owner: 'a' }), { code: 'schedule_unknown_resource' }); + await assert.rejects(schedule.take({ resource: 'pair', owner: 'a', apply: true }), { code: 'schedule_confirm' }); + assert.throws(() => normalizeResource('x', { capacity: 2, suspend: ['m'] }), /capacity 1/); + assert.throws(() => normalizeResource('x', { onExpiry: 'explode' }), /onExpiry/); +}); + +test('routes and suspensions that were already in place are left as they were', async (t) => { + const actions = fakeActions(); + actions.isSuspended = async ({ model }) => model === 'pair-model'; + const { schedule } = await fixture(t, { actions }); + actions.routes.omp = 'cloud'; + const held = await take(schedule, { resource: 'pair', owner: 'main' }); + assert.equal(held.state, 'active'); + assert.deepEqual(actions.calls, [['activeRoute', 'omp', 'node-a']]); + actions.calls.length = 0; + assert.equal((await release(schedule, held.id, { owner: 'main' })).state, 'released'); + assert.deepEqual(actions.calls, []); + assert.equal(actions.routes.omp, 'cloud'); + + // A route with no active profile has nothing exact to restore, so the take refuses. + delete actions.routes.omp; + const refused = await take(schedule, { resource: 'pair', owner: 'main' }); + assert.equal(refused.state, 'failed'); + assert.match(refused.error, /no active profile/); +}); + +test('resources that control the same model are unusable until fixed', async (t) => { + const { schedule } = await fixture(t, { + resources: { a: { suspend: ['shared'] }, b: { suspend: ['shared'] }, c: {} } + }); + await assert.rejects(take(schedule, { resource: 'a', owner: 'x' }), { code: 'schedule_config' }); + const board = schedule.board(); + assert.match(board.resources.find((entry) => entry.id === 'b').error, /a, b both control shared/); + assert.equal(board.resources.find((entry) => entry.id === 'c').error, undefined); +}); + +test('refused updates leave no trace; leases and jobs keep to their holders', async (t) => { + const { schedule } = await fixture(t); + const lease = await take(schedule, { resource: 'pair', owner: 'main' }); + await assert.rejects(schedule.startJob({ title: 'x', owner: 'other', reservationId: lease.id }), { + code: 'schedule_not_owner' + }); + const job = await schedule.startJob({ title: 'sweep', owner: 'main' }); + const tooMany = Object.fromEntries(Array.from({ length: 40 }, (_, i) => [`m${i}`, i])); + await assert.rejects(schedule.updateJob(job.id, { note: 'kept?', metrics: tooMany }), { + code: 'schedule_invalid_request' + }); + const after = schedule.job(job.id); + assert.deepEqual(after.metrics, {}); + assert.equal(after.notes.length, 0); + await assert.rejects(schedule.updateJob(job.id, { owner: 'other', note: 'mine now' }), { + code: 'schedule_not_owner' + }); + for (const id of ['__proto__', 'constructor', 'toString']) { + assert.throws(() => schedule.reservation(id), { code: 'schedule_unknown_reservation' }); + assert.throws(() => schedule.job(id), { code: 'schedule_unknown_job' }); + } + // A second agent under the same owner name is a different holder when both name a session. + const shared = await take(schedule, { resource: 'drafter', owner: 'team', session: 'claude:a', ttlMs: null }); + await assert.rejects(take(schedule, { resource: 'drafter', owner: 'team', session: 'claude:b' }), { + code: 'schedule_busy' + }); + assert.equal((await take(schedule, { resource: 'drafter', owner: 'team', session: 'claude:a' })).id, shared.id); +}); + +test('a stuck lease is reported, not renewed', async (t) => { + const actions = fakeActions({ failResume: 1 }); + const { schedule } = await fixture(t, { actions }); + const held = await take(schedule, { resource: 'pair', owner: 'main' }); + await release(schedule, held.id, { owner: 'main' }); + await assert.rejects(schedule.renew(held.id, { owner: 'main' }), { code: 'schedule_not_open' }); + assert.equal((await take(schedule, { resource: 'pair', owner: 'main' })).state, 'stuck'); +}); + +test('a deployment fence holds expiry and gives the time back', async (t) => { + const dir = await fs.mkdtemp(path.join(os.tmpdir(), 'lloom-schedule-')); + t.after(() => fs.rm(dir, { recursive: true, force: true })); + const clock = { now: 1_800_000_000_000 }; + let fenced = false; + const schedule = createSchedule({ + getConfig: () => ({ reservations: { resources: RESOURCES } }), + statePath: path.join(dir, 'schedule.json'), + actions: fakeActions(), + now: () => clock.now, + isPaused: () => fenced + }); + t.after(() => schedule.stop()); + const held = await take(schedule, { resource: 'builds', owner: 'ci' }); + fenced = true; + clock.now += 10 * MINUTE; + await schedule.sweep(); + clock.now += 40 * MINUTE; + await schedule.sweep(); + assert.equal(schedule.reservation(held.id).state, 'active'); + fenced = false; + await schedule.sweep(); + assert.equal(schedule.reservation(held.id).remainingMs, 20 * MINUTE); +}); From 518bed1468b16969456b45421654232b6d5166a0 Mon Sep 17 00:00:00 2001 From: Jason McCartney <jmac@theroot.org> Date: Sat, 17 Oct 2026 11:43:47 -0700 Subject: [PATCH 2/2] skills: lloom-schedule, for agents reserving hardware and reporting jobs Ships in the npm package (skills/) beside docs/reservations.md. Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com> --- package.json | 1 + skills/lloom-schedule/SKILL.md | 84 ++++++++++++++++++++++++++++++++++ 2 files changed, 85 insertions(+) create mode 100644 skills/lloom-schedule/SKILL.md diff --git a/package.json b/package.json index c33f761..a943c2e 100644 --- a/package.json +++ b/package.json @@ -48,6 +48,7 @@ "patches/", "recipes/", "schemas/", + "skills/", "scripts/", "src/", "THIRD_PARTY_NOTICES.md", diff --git a/skills/lloom-schedule/SKILL.md b/skills/lloom-schedule/SKILL.md new file mode 100644 index 0000000..31e6c10 --- /dev/null +++ b/skills/lloom-schedule/SKILL.md @@ -0,0 +1,84 @@ +--- +name: lloom-schedule +description: >- + Reserve shared GPU hardware and report long-running jobs through the LLooM + head (`lloom reserve`, `lloom release`, `lloom job`, `lloom schedule`). Use + before running anything on shared machines (the Spark pair, build boxes, + benchmark hosts), when the task says "take/hold/lock the pair", "is the pair + free", "who has the GPUs", "release the pair", "wait for the pair", or when + starting a build, benchmark, training or eval run that takes more than a few + minutes and others should be able to follow. Replaces sideband locks such as + pair.sh / pair-mac.sh. +--- + +# LLooM reservations and jobs + +The LLooM head keeps one schedule. **Reservations** are expiring leases on named +resources. Taking one moves traffic away and suspends that resource's models, and +releasing it restores them once they pass a health check. **Jobs** are status +records you keep current. People and Tessera read both, so nobody has to ask you +how it's going. + +## Setup + +```sh +export LLOOM_HEAD_URL=http://<head>:8100 # the schedule-owning gateway +export LLOOM_HEAD_API_KEY=... # its admin key (never pass keys on the command line) +export LLOOM_OWNER=<short-name> # who you are on the board, e.g. buildopt +export LLOOM_SESSION=claude:<session-id> # optional: lets Tessera open your conversation +``` + +On the head itself, `LLOOM_HEAD_URL` isn't needed. + +## Look before you touch hardware + +```sh +lloom schedule # resources, holders, time left, who waits, running jobs +``` + +If a resource shows a holder or a standing reservation (for example the drafter +boxes), do not run GPU work there. Builds and containers you start yourself +aren't stopped by LLooM; honouring the schedule is on you. + +## Take, keep, give back + +```sh +lloom reserve pair --purpose "decode A/B: indexer v2" # preview the actions +lloom reserve pair --purpose "decode A/B: indexer v2" --wait --apply --yes # take it, queueing if busy +lloom renew pair # extend (job reports also extend it) +lloom release pair --apply --yes # restore and free +``` + +- `reserve` without `--wait` fails straight away if the resource is busy, and + names the holder. With `--wait` it queues and returns once the lease is active + (exit 0). Ctrl-C gives your place back. +- Leases default to 30 minutes and warn 5 minutes before expiry. At expiry LLooM + restores the resource, even in the middle of your run. Renew, or keep a job + reporting. +- Take the lease only for the time you are using the hardware. Release as soon as + you're done, before writing up results. +- `release` exits non-zero if the restore failed. The resource stays held (`stuck`) + and traffic stays away from the unhealthy model. Fix the cause, release again, + and tell the developer. + +## Report a job + +```sh +job=$(lloom job start "decode A/B: indexer v2" --on pair --note "building image") +lloom job progress $job 3/8 --unit arms "arm 3: 58 tok/s" +lloom job metric $job tok_s=58.2 ttft_ms=412 +lloom job note $job "arm 4 OOM at 512K, retrying at 256K" +lloom job done $job "flat vs baseline at 4K, +4% at 64K" --receipt receipts/ab-1017.json +lloom job fail $job "image build broke: missing CUTLASS header" +``` + +- Every update is a heartbeat. After 15 minutes of silence (`--stale` sets + another limit) the job shows as stalled on the board and in Tessera. +- A job started `--on <resource>` keeps your lease alive while it reports. +- Notes should say what a person needs to know, not log lines. Put numbers in + metrics and file paths in `--receipt`. +- Finished, failed and stalled jobs show in Tessera's Needs-you lane. That + replaces "how's it going?". + +`--json` on any command gives the full record. `lloom help` lists everything, and +`docs/reservations.md` in the LLooM repo covers the HTTP API and resource configuration.