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..a943c2e 100644 --- a/package.json +++ b/package.json @@ -48,6 +48,7 @@ "patches/", "recipes/", "schemas/", + "skills/", "scripts/", "src/", "THIRD_PARTY_NOTICES.md", @@ -64,7 +65,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 +74,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/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. 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); +});