From 1eeffbf60230379818599ca40d207cbdd7dd0a4e Mon Sep 17 00:00:00 2001 From: Jeremy Cohn Date: Wed, 14 Oct 2026 10:08:06 -0700 Subject: [PATCH] feat(mocactl): workloads create/get/delete and spawn --workload mocactl workloads create --context FILE creates a workload, streams the context file to Context Service with the one-time grant, and activates the uploaded revision. mocactl spawn --workload ID runs the subagent on that workload's context. Replaces kubectl cp seeding of sandbox fixtures (#517). Assisted-By: Claude (Anthropic AI) Signed-off-by: Jeremy Cohn --- docs/context-service.md | 8 + docs/reference/cli-mocactl.md | 26 +- packages/mocactl/README.md | 42 +- packages/mocactl/src/api/workloads.ts | 258 +++++++++++ packages/mocactl/src/cli.ts | 49 +++ packages/mocactl/src/core/subagent-ledger.ts | 3 + packages/mocactl/src/headless.ts | 5 +- packages/mocactl/src/subagent.ts | 24 +- packages/mocactl/src/transport/leaf-turn.ts | 2 + packages/mocactl/src/workloads.ts | 284 ++++++++++++ packages/mocactl/test/cli.test.ts | 66 +++ packages/mocactl/test/subagent-list.test.ts | 13 + packages/mocactl/test/subagent-spawn.test.ts | 62 +++ packages/mocactl/test/workloads.test.ts | 439 +++++++++++++++++++ 14 files changed, 1275 insertions(+), 6 deletions(-) create mode 100644 packages/mocactl/src/api/workloads.ts create mode 100644 packages/mocactl/src/workloads.ts create mode 100644 packages/mocactl/test/workloads.test.ts diff --git a/docs/context-service.md b/docs/context-service.md index fb86e229..19f27a0b 100644 --- a/docs/context-service.md +++ b/docs/context-service.md @@ -26,6 +26,14 @@ Context Service: the workload's frozen revision available read-only. 5. **Delete:** `DELETE /workloads/{id}` deletes the Context, then the workload. +`mocactl` is the client for this flow +([#495](https://github.com/rossoctl/moca/issues/495), one CLI for delegating subagent tasks; +[#517](https://github.com/rossoctl/moca/issues/517), seed sandbox files before a leaf runs): +`mocactl workloads create --context FILE` does steps 1 to 3, `mocactl spawn --kind prompt --workload ID` +step 4, and `mocactl workloads delete ID` step 5. It does not build the context file: +`contextctl ctx export` does. See "Delegate with context" in +[`packages/mocactl/README.md`](../packages/mocactl/README.md). + The routes are off by default. With them off, every `/workloads` route answers `501 workloads_unavailable`. diff --git a/docs/reference/cli-mocactl.md b/docs/reference/cli-mocactl.md index 6482676e..a38a0b80 100644 --- a/docs/reference/cli-mocactl.md +++ b/docs/reference/cli-mocactl.md @@ -19,7 +19,8 @@ All flags are accepted by one parser. An unknown flag, or a missing flag value, exits 2. Some flags are always rejected (exit 2) outside their command: `--consumer`, `--host` and `--endpoint` outside `credentials add`; `--kind` outside `credentials add` and `spawn`; `--config` outside `run` and `spawn`; `--repo` and `--ref` outside `spawn`; `--tag` outside `spawn` and -`subagents`; `--out` outside `get`; and `--all` outside `logout`. Other flags a command does not use +`subagents`; `--out` outside `get`; `--workload` outside `spawn`; `--context` and `--name` outside +`workloads create`; and `--all` outside `logout`. Other flags a command does not use are not rejected and may be ignored. ## Endpoints @@ -73,6 +74,9 @@ reachable?` instead; `auth token` exits 3 (login required) or 4; `logout` while | `mocactl spawn "prompt" [--config DIGEST] --kind prompt\|solve [--repo URL --ref REF] [--tag T] [--json]` | Creates a session, writes a ledger record, forks a detached worker that holds a `POST /v1/runs`. Prints the sessionId on stdout | | `mocactl subagents [--tag T] [--json]` | Lists spawned subagents from the ledger joined with `GET /runs/status` | | `mocactl get SESSIONID [--out DIR] [--json]` | Fetches a terminal result from `GET /runs/status` and writes `.text` / `.verdict.json` / `.patch` | +| `mocactl workloads create --context FILE [--name NAME] [--json]` | `POST /workloads`, uploads `FILE` to the returned grant, then `POST /workloads/{id}/activate`. Prints the workload id | +| `mocactl workloads get ID [--json]` | Reads a workload (`GET /workloads/{id}`) | +| `mocactl workloads delete ID [--json]` | Deletes a workload and its context (`DELETE /workloads/{id}`) | ### `run` @@ -113,7 +117,8 @@ for behavior; this section covers only CLI-reference details. - `spawn` exit 2: empty prompt, prompt over 64 KiB, bundle digest not matching `sha256:<64 lowercase hex>`, `--kind` other than `prompt`/`solve`, `--kind solve` without both `--repo` and `--ref`, or - `--repo`/`--ref` on `--kind prompt`. + `--repo`/`--ref` on `--kind prompt`, `--workload` without `--kind prompt`, or a workload id that + is not a lowercase Kubernetes name of at most 50 characters. - `spawn` on a sync 503 `saturated`: the worker retries up to 3 times honoring `Retry-After`, bounded by a 5-minute wall budget; after that the ledger row is `failed` with `stateDetail: 'saturated — exhausted retries'`. @@ -121,9 +126,24 @@ hex>`, `--kind` other than `prompt`/`solve`, `--kind solve` without both `--repo `responded`/`done`/`solved`; `queued`/`paused`/`failed` exit 1. - `--json`: - `spawn` → `{"type":"session","sessionId","pid"}`. - - `subagents` → `{"subagents":[{"sessionId","tag","kind","state","detail","bundle","submittedAt"}]}`. + - `subagents` → `{"subagents":[{"sessionId","tag","kind","state","detail","bundle","workload","submittedAt"}]}`. - `get` → `{"sessionId","status","files":[…],"reason"?,"message"?}`. +### `workloads` + +See "Delegate with context" in [`packages/mocactl/README.md`](../../packages/mocactl/README.md) and +[`docs/context-service.md`](../context-service.md). The routes are on the harness and take the API +token. There is no list command: the harness has no list route. + +- `create` exit 2: no `--context`, a missing, empty or non-regular file, or an invalid `--name`; all + before any request. Exit 1 for a refusal, including a file larger than the grant's `maxBytes` or an + upload URL that is neither HTTPS nor loopback HTTP; a workload created before the failure is deleted. +- `get`/`delete` exit 2 for an invalid id, exit 1 for `404 workload_not_found` or another refusal. +- `--json`: + - `create` → `{"workloadId","status","contextType","revision","files","bytes"}`. + - `get` → `{"workloadId","status","contextType","workspace","revision"}` (`null` when unset). + - `delete` → `{"workloadId","status":"deleted"}`. + ### `promote` exit codes | Code | Cause | diff --git a/packages/mocactl/README.md b/packages/mocactl/README.md index 0f229e11..1d42c4ca 100644 --- a/packages/mocactl/README.md +++ b/packages/mocactl/README.md @@ -102,6 +102,13 @@ mocactl spawn "prompt" [--config sha256:…] --kind prompt|solve \ [--repo URL --ref REF] [--tag T] [--json] # fork a detached subagent; prints its sessionId mocactl subagents [--tag T] [--json] # list spawned subagents + current honest state mocactl get SESSIONID [--out DIR] [--json] # write result files + bounded summary +mocactl spawn "prompt" --kind prompt --workload ID [...] # run it on a workload's context +``` + +```bash +mocactl workloads create --context FILE [--name NAME] [--json] # upload a context; prints its id +mocactl workloads get ID [--json] # status, context type, size and frozen revision +mocactl workloads delete ID [--json] # delete the workload and its context ``` `mocactl spawn` creates a session (pinned to the bundle if `--config` is given), writes a record @@ -160,6 +167,36 @@ Exit codes, for every command: `0` it worked, `1` it failed (the turn, or the co refused), `2` a usage or setup problem (bad flags, not logged in, no destination for the turn), `130` cancelled (Ctrl-C, which also stops a listing or a stdin read that is waiting). +### Delegate with context + +A workload gives a subagent files to work from: a context, uploaded once and available read-only +to every run on it ([`docs/context-service.md`](../../docs/context-service.md)). Build the context +file with `contextctl` (`mocactl` never calls it), then: + +```bash +contextctl ctx export ... -o ctx.context +id=$(mocactl workloads create --context ctx.context --name demo) +sid=$(mocactl spawn "summarize the repo in the context" --kind prompt --workload "$id") +mocactl get "$sid" # once `mocactl subagents` says done +mocactl workloads delete "$id" +``` + +`workloads create` creates the workload (`POST /workloads` on the harness, with your API token), +uploads the file straight to Context Service with the one-time grant it returns, then activates the +revision the upload reports, which freezes it. It checks that the file exists before it starts, and +that it is no larger than the grant allows before it uploads; the upload streams from disk. A grant +that has expired, or that the upload refuses with `401`, is replaced once +(`POST /workloads/ID/uploads`). The grant's URL must be HTTPS (plain HTTP only on loopback). If +anything fails after the workload exists, `create` deletes it again, so it does not count against +your quota (5 live workloads by default), and exits `1`. `--name` is a lowercase Kubernetes name of +at most 50 characters; without one the harness generates it. The harness has no route that lists +workloads, so there is no `mocactl workloads` listing: keep the id `create` prints. + +`spawn --workload ID` adds `workloadId` to the run and records it in the ledger row +(`mocactl subagents --json` shows it as `workload`). Only prompt runs can use a workload, so +`--workload` needs `--kind prompt`. A workload that is not `ready`, or not yours, fails the run +(`workload_not_ready`, `workload_not_found`), as `mocactl subagents` then shows. + ### Output stdout carries only the result; every message goes to stderr, including "no sessions" for an @@ -175,8 +212,11 @@ their own shapes. These shapes are stable: fields may be added, but none is rena | `credentials add` | `{"name","status":"stored"}` | | `credentials delete` | `{"name","status":"deleted"}` | | `spawn` | `{"type":"session","sessionId","pid"}` | -| `subagents` | `{"subagents":[{"sessionId","tag","kind","state","detail","bundle","submittedAt"}]}` | +| `subagents` | `{"subagents":[{"sessionId","tag","kind","state","detail","bundle","workload","submittedAt"}]}` | | `get` | `{"sessionId","status","files":["",…],"reason"?,"message"?}` | +| `workloads create` | `{"workloadId","status","contextType","revision","files","bytes"}` | +| `workloads get` | `{"workloadId","status","contextType","workspace","revision"}`, `null` for what is not set | +| `workloads delete` | `{"workloadId","status":"deleted"}` | `createdAt` and `lastTurnAt` are epoch milliseconds as the control plane sends them (`lastTurnAt` is `null` before the first turn); `endpoint` is `null` when the deployment default applies. diff --git a/packages/mocactl/src/api/workloads.ts b/packages/mocactl/src/api/workloads.ts new file mode 100644 index 00000000..c27bb379 --- /dev/null +++ b/packages/mocactl/src/api/workloads.ts @@ -0,0 +1,258 @@ +/** + * Client for the harness's `/workloads` routes (docs/context-service.md) and the one-time upload + * of a context file straight to Context Service. The routes take the user's API token; the upload + * takes only the grant's own token. + */ +import { createReadStream } from 'node:fs'; +import { Readable } from 'node:stream'; +import { ApiError, errorFromResponse, networkError } from './errors.js'; +import type { LeafHarness } from '../transport/leaf-turn.js'; +import { trimTrailingSlashes } from './url.js'; + +/** The media type Context Service accepts on an upload. */ +export const CONTEXT_MEDIA_TYPE = 'application/vnd.rossoctl.context'; +/** A workload name: a lowercase Kubernetes name, as the harness checks it. */ +export const WORKLOAD_ID = /^[a-z0-9](?:[-a-z0-9]*[a-z0-9])?$/; +export const MAX_WORKLOAD_ID_LENGTH = 50; +const REVISION = /^[a-f0-9]{64}$/; + +export function validWorkloadId(id: string): boolean { + return id.length <= MAX_WORKLOAD_ID_LENGTH && WORKLOAD_ID.test(id); +} + +/** A one-time grant to PUT one context file to Context Service. */ +export interface UploadGrant { + uploadUrl: string; + token: string; + expiresAt: string; + method: 'PUT'; + contentType: string; + maxBytes: number; +} + +export interface Workload { + workloadId: string; + status: string; + contextType?: string; + workspace?: { size?: string; storageClass?: string }; + revision?: string; +} + +export interface CreateWorkloadRequest { + name?: string; +} + +export interface UploadResult { + revision: string; + files?: number; + bytes?: number; +} + +/** A refusal from Context Service's upload route; `status` 0 means it was never reached. */ +export class ContextUploadError extends Error { + constructor( + readonly status: number, + readonly code: string, + message?: string, + ) { + super(message ?? code); + this.name = 'ContextUploadError'; + } +} + +/** + * Where an upload may go: HTTPS, or plain HTTP on loopback only (local development), the rule the + * harness applies to `CONTEXT_SERVICE_PUBLIC_URL`. Anything else would send the grant's token in + * the clear. Returns the reason it is refused, or undefined. + */ +export function uploadUrlProblem(raw: string): string | undefined { + let url: URL; + try { + url = new URL(raw); + } catch { + return 'the upload URL is not a URL'; + } + if (url.protocol === 'https:') return undefined; + const loopback = ['localhost', '127.0.0.1', '[::1]'].includes(url.hostname); + if (url.protocol === 'http:' && loopback) return undefined; + return 'the upload URL must use HTTPS (plain HTTP only on loopback)'; +} + +export class WorkloadsClient { + constructor( + private readonly harness: LeafHarness, + /** The user's API token, refreshed if needed; read before every request. */ + private readonly apiToken: () => Promise, + private readonly fetchImpl: typeof fetch = fetch, + ) {} + + /** POST /workloads: the workload, awaiting upload, plus its first grant. */ + async create( + req: CreateWorkloadRequest, + signal?: AbortSignal, + ): Promise { + const body = await this.request('POST', '/workloads', req, signal); + const workload = asWorkload(body); + const upload = asGrant((body as { upload?: unknown }).upload); + return { ...workload, upload }; + } + + /** POST /workloads/{id}/uploads: a fresh grant while the workload awaits upload. */ + async freshUpload(id: string, signal?: AbortSignal): Promise { + return asGrant( + await this.request('POST', `/workloads/${encodeURIComponent(id)}/uploads`, undefined, signal), + ); + } + + /** POST /workloads/{id}/activate: freeze the uploaded revision; the workload becomes ready. */ + async activate(id: string, revision: string, signal?: AbortSignal): Promise { + return asWorkload( + await this.request( + 'POST', + `/workloads/${encodeURIComponent(id)}/activate`, + { revision }, + signal, + ), + ); + } + + async get(id: string, signal?: AbortSignal): Promise { + return asWorkload( + await this.request('GET', `/workloads/${encodeURIComponent(id)}`, undefined, signal), + ); + } + + /** DELETE /workloads/{id}: 204. */ + async delete(id: string, signal?: AbortSignal): Promise { + await this.request('DELETE', `/workloads/${encodeURIComponent(id)}`, undefined, signal); + } + + /** + * PUT the file at `path` (`size` bytes) to the grant's URL, streamed from disk rather than read + * into memory. Throws ContextUploadError for a refusal or an unusable reply. + */ + async upload( + grant: UploadGrant, + path: string, + size: number, + signal?: AbortSignal, + ): Promise { + const problem = uploadUrlProblem(grant.uploadUrl); + if (problem) throw new ContextUploadError(0, 'upload_url_unsafe', problem); + const stream = createReadStream(path); + let res: Response; + try { + res = await this.fetchImpl(grant.uploadUrl, { + method: 'PUT', + headers: { + authorization: `Bearer ${grant.token}`, + 'content-type': CONTEXT_MEDIA_TYPE, + 'content-length': String(size), + }, + body: Readable.toWeb(stream) as ReadableStream, + // Required by Node's fetch for a streamed request body. + duplex: 'half', + signal, + } as RequestInit & { duplex: 'half' }); + } catch (err) { + stream.destroy(); + if (signal?.aborted) throw err; + throw new ContextUploadError( + 0, + 'network_error', + `cannot reach Context Service: ${err instanceof Error ? err.message : String(err)}`, + ); + } + if (!res.ok) { + const e = await errorFromResponse('harness', res); + throw new ContextUploadError(res.status, e.code, e.message); + } + const body = (await res.json().catch(() => null)) as Partial | null; + if (!body || typeof body.revision !== 'string' || !REVISION.test(body.revision)) { + throw new ContextUploadError( + res.status, + 'invalid_body', + 'Context Service returned no revision', + ); + } + return { + revision: body.revision, + ...(typeof body.files === 'number' ? { files: body.files } : {}), + ...(typeof body.bytes === 'number' ? { bytes: body.bytes } : {}), + }; + } + + private async request( + method: string, + path: string, + body: unknown, + signal?: AbortSignal, + ): Promise { + const base = trimTrailingSlashes(await this.harness.baseUrl()); + const token = await this.apiToken(); + let res: Response; + try { + res = await this.fetchImpl(`${base}${path}`, { + method, + headers: { + ...(body !== undefined ? { 'content-type': 'application/json' } : {}), + ...(token ? { authorization: `Bearer ${token}` } : {}), + }, + ...(body !== undefined ? { body: JSON.stringify(body) } : {}), + signal, + }); + } catch (err) { + if (signal?.aborted) throw err; + throw networkError('harness', err); + } + if (!res.ok) throw await errorFromResponse('harness', res); + if (res.status === 204) return undefined; + try { + return await res.json(); + } catch { + throw new ApiError('harness', 0, 'invalid_body', `the harness returned non-JSON for ${path}`); + } + } +} + +function invalid(what: string): ApiError { + return new ApiError('harness', 0, 'invalid_body', `the harness returned an invalid ${what}`); +} + +function asWorkload(body: unknown): Workload { + const b = body as Partial | null; + if (!b || typeof b.workloadId !== 'string' || typeof b.status !== 'string') { + throw invalid('workload'); + } + return { + workloadId: b.workloadId, + status: b.status, + ...(typeof b.contextType === 'string' ? { contextType: b.contextType } : {}), + ...(b.workspace && typeof b.workspace === 'object' ? { workspace: b.workspace } : {}), + ...(typeof b.revision === 'string' ? { revision: b.revision } : {}), + }; +} + +function asGrant(body: unknown): UploadGrant { + const g = body as Partial | null; + if ( + !g || + typeof g.uploadUrl !== 'string' || + typeof g.token !== 'string' || + !g.token || + typeof g.expiresAt !== 'string' || + typeof g.maxBytes !== 'number' || + !Number.isSafeInteger(g.maxBytes) || + g.maxBytes <= 0 + ) { + throw invalid('upload grant'); + } + return { + uploadUrl: g.uploadUrl, + token: g.token, + expiresAt: g.expiresAt, + method: 'PUT', + contentType: typeof g.contentType === 'string' ? g.contentType : CONTEXT_MEDIA_TYPE, + maxBytes: g.maxBytes, + }; +} diff --git a/packages/mocactl/src/cli.ts b/packages/mocactl/src/cli.ts index e80a4970..821b4340 100644 --- a/packages/mocactl/src/cli.ts +++ b/packages/mocactl/src/cli.ts @@ -18,6 +18,7 @@ import { } from './headless.js'; import { buildRuntime, ensureRuntimeAuth, type Runtime } from './runtime.js'; import { cmdGet, cmdSpawn, cmdSubagents, runSpawnWorker } from './subagent.js'; +import { cmdWorkloadCreate, cmdWorkloadDelete, cmdWorkloadGet } from './workloads.js'; import { VERSION } from './version.js'; export const USAGE = `usage: @@ -41,8 +42,15 @@ export const USAGE = `usage: mocactl credentials delete NAME [--json] mocactl spawn "prompt" [--config DIGEST] --kind prompt|solve [--repo URL --ref REF] [--tag T] [--json] fork a detached subagent (solve needs --repo/--ref) + mocactl spawn "prompt" --kind prompt --workload ID [...] + run the subagent on a workload's context mocactl subagents [--tag T] [--json] list spawned subagents + current state mocactl get SESSIONID [--out DIR] [--json] write result files + bounded summary + mocactl workloads create --context FILE [--name NAME] [--json] + upload a context file (contextctl ctx export) to a + new workload; prints its id + mocactl workloads get ID [--json] + mocactl workloads delete ID [--json] delete a workload and its context flags for every command: --control-plane-url URL (the control plane says where the harness is; --harness-url URL overrides that)`; @@ -108,6 +116,9 @@ export async function main( ref: { type: 'string' }, tag: { type: 'string' }, out: { type: 'string' }, + workload: { type: 'string' }, + context: { type: 'string' }, + name: { type: 'string' }, help: { type: 'boolean', short: 'h' }, version: { type: 'boolean', short: 'V' }, }, @@ -165,6 +176,17 @@ export async function main( io.err(`--out only applies to \`mocactl get\`\n${USAGE}`); return 2; } + if (values.workload !== undefined && command !== 'spawn') { + io.err(`--workload only applies to \`mocactl spawn\`\n${USAGE}`); + return 2; + } + const workloadsCreate = command === 'workloads' && rest[0] === 'create'; + if ((values.context !== undefined || values.name !== undefined) && !workloadsCreate) { + io.err( + `--${values.context !== undefined ? 'context' : 'name'} only applies to \`mocactl workloads create\`\n${USAGE}`, + ); + return 2; + } // B14: refresh an expired API token once, up front, so every command's login check -- and the // interactive login screen -- sees the refreshed login rather than asking for a new one. `login`, // `logout` and `auth` manage the login themselves. `__spawn-worker` is the detached child of @@ -292,6 +314,7 @@ export async function main( repoUrl: values.repo, ref: values.ref, tag: values.tag, + workloadId: values.workload, json, }); } @@ -304,6 +327,32 @@ export async function main( if (sid === undefined || extra.length > 0) return usage(io); return cmdGet(rt, io, { sessionId: sid, outDir: values.out, json, signal }); } + case 'workloads': { + const [sub, id, ...extra] = rest; + if (sub === undefined) { + // The harness has no route that lists workloads. + io.err(`mocactl workloads needs create, get or delete (there is no list)\n${USAGE}`); + return 2; + } + if (sub === 'create') { + if (id !== undefined) return usage(io); + if (!values.context) { + io.err(`workloads create needs --context FILE\n${USAGE}`); + return 2; + } + return cmdWorkloadCreate(rt, io, { + contextPath: values.context, + name: values.name, + json, + signal, + }); + } + if (sub !== 'get' && sub !== 'delete') return unknown(io, 'workloads', sub); + if (id === undefined || extra.length > 0) return usage(io); + return sub === 'get' + ? cmdWorkloadGet(rt, io, { id, json, signal }) + : cmdWorkloadDelete(rt, io, { id, json, signal }); + } // Hidden entrypoint: the detached worker forked by `mocactl spawn` re-execs itself with this // subcommand, reads its payload from stdin, and writes the result to the ledger. Not in USAGE // because end users should never run it directly. diff --git a/packages/mocactl/src/core/subagent-ledger.ts b/packages/mocactl/src/core/subagent-ledger.ts index 12d202bd..0c78958c 100644 --- a/packages/mocactl/src/core/subagent-ledger.ts +++ b/packages/mocactl/src/core/subagent-ledger.ts @@ -26,12 +26,15 @@ export interface LedgerEnvelope { repoUrl?: string; ref?: string; configRef?: string; + workloadId?: string; } export interface LedgerRecord { sessionId: string; tag?: string; bundle?: string; + /** The workload the run was sent to (`spawn --workload`). */ + workloadId?: string; envelope: LedgerEnvelope; submittedAt: number; /** The detached worker's pid. Absent until the child is forked. */ diff --git a/packages/mocactl/src/headless.ts b/packages/mocactl/src/headless.ts index a1d83547..6f356689 100644 --- a/packages/mocactl/src/headless.ts +++ b/packages/mocactl/src/headless.ts @@ -155,7 +155,10 @@ export interface RunOptions { } /** False (with the reason on stderr) unless there is a control plane and a valid login for it. */ -function ready(rt: Runtime, io: Io): rt is Runtime & Required> { +export function ready( + rt: Runtime, + io: Io, +): rt is Runtime & Required> { if (missing(rt, io, ['controlPlaneUrl']) || !rt.cp || !rt.harness) return false; if (!apiTokenValid(rt.auth, rt.now())) { io.err( diff --git a/packages/mocactl/src/subagent.ts b/packages/mocactl/src/subagent.ts index 23a71aa2..34b6ea41 100644 --- a/packages/mocactl/src/subagent.ts +++ b/packages/mocactl/src/subagent.ts @@ -3,6 +3,7 @@ import { spawn as spawnChild } from 'node:child_process'; import { writeFileSync, mkdirSync, existsSync, openSync, closeSync } from 'node:fs'; import { join, resolve } from 'node:path'; import { ApiError } from './api/errors.js'; +import { validWorkloadId } from './api/workloads.js'; import type { ControlPlaneApi, CreateSessionRequest } from './api/types.js'; import { describeError } from './core/messages.js'; import { sanitizeRemote } from './core/sanitize.js'; @@ -72,6 +73,8 @@ export interface SpawnOptions { repoUrl?: string; ref?: string; tag?: string; + /** A ready workload whose context the run sees; prompt runs only (the harness refuses others). */ + workloadId?: string; json: boolean; /** Test seams: fork + ledger injection. */ forkFn?: ( @@ -109,6 +112,16 @@ export async function cmdSpawn(rt: Runtime, io: Io, opts: SpawnOptions): Promise io.err('--repo and --ref are only meaningful with --kind solve'); return 2; } + if (opts.workloadId !== undefined) { + if (opts.kind !== 'prompt') { + io.err('--workload needs --kind prompt: the harness runs only prompts on a workload'); + return 2; + } + if (!validWorkloadId(opts.workloadId)) { + io.err(`invalid workload id '${sanitizeRemote(opts.workloadId)}'`); + return 2; + } + } let session: SpawnedSession; try { @@ -131,6 +144,7 @@ export async function cmdSpawn(rt: Runtime, io: Io, opts: SpawnOptions): Promise ...(opts.tag ? { tag: opts.tag } : {}), envelope: envelopeForLedger(opts), ...(opts.configRef ? { bundle: opts.configRef } : {}), + ...(opts.workloadId ? { workloadId: opts.workloadId } : {}), ...(opts.repoUrl ? { repoUrl: opts.repoUrl } : {}), ...(opts.ref ? { ref: opts.ref } : {}), submittedAt: rt.now(), @@ -183,7 +197,13 @@ function buildEnvelope(sessionId: string, opts: SpawnOptions): LeafEnvelope { // `/runs` reads configRef from the body (not the session); the harness builds the solve prompt // itself from `problemStatement`. if (opts.configRef) base.configRef = opts.configRef; - if (opts.kind === 'prompt') return { ...base, prompt: opts.prompt }; + if (opts.kind === 'prompt') { + return { + ...base, + prompt: opts.prompt, + ...(opts.workloadId ? { workloadId: opts.workloadId } : {}), + }; + } return { ...base, problemStatement: opts.prompt, repoUrl: opts.repoUrl!, ref: opts.ref! }; } @@ -194,6 +214,7 @@ function envelopeForLedger(opts: SpawnOptions): LedgerEnvelope { ...(opts.repoUrl ? { repoUrl: opts.repoUrl } : {}), ...(opts.ref ? { ref: opts.ref } : {}), ...(opts.configRef ? { configRef: opts.configRef } : {}), + ...(opts.workloadId ? { workloadId: opts.workloadId } : {}), }; } @@ -440,6 +461,7 @@ export async function cmdSubagents(rt: Runtime, io: Io, opts: SubagentsOptions): state: view.state, detail: view.detail ?? null, bundle: record.bundle ?? null, + workload: record.workloadId ?? null, submittedAt: record.submittedAt, })), }) + '\n', diff --git a/packages/mocactl/src/transport/leaf-turn.ts b/packages/mocactl/src/transport/leaf-turn.ts index c28c70d8..9ffc0582 100644 --- a/packages/mocactl/src/transport/leaf-turn.ts +++ b/packages/mocactl/src/transport/leaf-turn.ts @@ -14,6 +14,8 @@ export interface LeafEnvelope { repoUrl?: string; ref?: string; configRef?: string; + /** A ready workload whose context the run sees (prompt runs only). */ + workloadId?: string; } /** diff --git a/packages/mocactl/src/workloads.ts b/packages/mocactl/src/workloads.ts new file mode 100644 index 00000000..8968f8a0 --- /dev/null +++ b/packages/mocactl/src/workloads.ts @@ -0,0 +1,284 @@ +/** + * `mocactl workloads create | get | delete`: a workload carries a context (built elsewhere, e.g. + * with `contextctl ctx export`) that `mocactl spawn --workload` runs prompts on. The harness has no + * route that lists workloads, so there is no list command. + */ +import { statSync } from 'node:fs'; +import { resolve } from 'node:path'; +import { ApiError, TOKEN_CODES } from './api/errors.js'; +import { + ContextUploadError, + MAX_WORKLOAD_ID_LENGTH, + WorkloadsClient, + validWorkloadId, + type UploadGrant, + type UploadResult, + type Workload, +} from './api/workloads.js'; +import { describeError } from './core/messages.js'; +import { sanitizeRemote } from './core/sanitize.js'; +import { ready, type Io } from './headless.js'; +import { ensureRuntimeAuth, type Runtime } from './runtime.js'; + +/** A grant this close to its expiry is replaced before the upload starts. */ +const GRANT_EXPIRY_MARGIN_MS = 10_000; + +export interface WorkloadOptions { + json: boolean; + signal?: AbortSignal; + /** Test seam; otherwise built from the runtime. */ + client?: WorkloadsClient; +} + +function clientFor(rt: Runtime & Required>): WorkloadsClient { + return new WorkloadsClient( + rt.harness, + async () => { + // A long upload can outlive the 15-minute API token; this is a no-op while it is fresh. + await ensureRuntimeAuth(rt); + return rt.auth?.apiToken; + }, + rt.fetchImpl, + ); +} + +/** The sentence for a failed workload request: the harness's own code, which matters here. */ +export function describeWorkloadError(err: unknown, id?: string): string { + if (err instanceof ContextUploadError) { + return sanitizeRemote( + err.status === 0 + ? err.message + : `Context Service refused the upload: ${err.status} ${err.code}` + + (err.message && err.message !== err.code ? ` (${err.message})` : ''), + ); + } + if (err instanceof ApiError && err.status !== 0 && !TOKEN_CODES.has(err.code)) { + switch (err.code) { + case 'workloads_unavailable': + return 'workloads are not enabled on this harness — its operator turns them on (docs/context-service.md)'; + case 'workloads_misconfigured': + return "the harness's workload settings are incomplete — ask its operator"; + case 'workload_not_found': + return id ? `no workload '${sanitizeRemote(id)}' (or it is not yours)` : 'no such workload'; + case 'workload_quota_exceeded': + return 'you have as many workloads as you may — delete one with `mocactl workloads delete ID`'; + case 'workload_name_taken': + return `you already have a workload named '${sanitizeRemote(id ?? '')}'`; + } + return sanitizeRemote( + `the harness refused: ${err.status} ${err.code}` + + (err.message && err.message !== err.code ? ` (${err.message})` : ''), + ); + } + return describeError(err); +} + +function aborted(signal: AbortSignal | undefined): boolean { + return signal?.aborted === true; +} + +// --- create --- + +export interface CreateOptions extends WorkloadOptions { + /** The context file to upload, as `contextctl ctx export` writes it. */ + contextPath: string; + name?: string; +} + +/** + * POST /workloads, PUT the context file to the grant, then activate the revision the upload + * returned. A grant that has expired, or that the upload rejects with 401, is replaced once via + * POST .../uploads. On any failure after the workload exists, it is deleted again (best effort) so + * it does not hold a slot of the per-user quota. + */ +export async function cmdWorkloadCreate(rt: Runtime, io: Io, opts: CreateOptions): Promise { + if (opts.name !== undefined && !validWorkloadId(opts.name)) { + io.err( + `--name must be a lowercase Kubernetes name of at most ${MAX_WORKLOAD_ID_LENGTH} characters (a-z, 0-9, '-')`, + ); + return 2; + } + const path = resolve(opts.contextPath); + let size: number; + try { + const st = statSync(path); + if (!st.isFile()) { + io.err(`--context ${opts.contextPath} is not a file`); + return 2; + } + size = st.size; + } catch { + io.err( + `--context ${opts.contextPath} does not exist (build one with \`contextctl ctx export\`)`, + ); + return 2; + } + if (size === 0) { + io.err(`--context ${opts.contextPath} is empty`); + return 2; + } + if (!ready(rt, io)) return 2; + const client = opts.client ?? clientFor(rt); + const { signal } = opts; + + let created: Workload & { upload: UploadGrant }; + try { + created = await client.create(opts.name ? { name: opts.name } : {}, signal); + } catch (err) { + if (aborted(signal)) return 130; + io.err(describeWorkloadError(err, opts.name)); + return 1; + } + const id = created.workloadId; + io.err(`created workload ${sanitizeRemote(id)}; uploading ${size} bytes`); + + const fail = async (message: string): Promise => { + io.err(message); + if (aborted(signal)) { + io.err( + `workload ${sanitizeRemote(id)} was left awaiting upload: \`mocactl workloads delete ${sanitizeRemote(id)}\``, + ); + return 130; + } + try { + await client.delete(id); + io.err(`deleted workload ${sanitizeRemote(id)}`); + } catch (err) { + io.err( + `could not delete workload ${sanitizeRemote(id)} (${describeWorkloadError(err, id)}); ` + + `run \`mocactl workloads delete ${sanitizeRemote(id)}\``, + ); + } + return 1; + }; + const tooLarge = (grant: UploadGrant) => + size > grant.maxBytes + ? `--context ${opts.contextPath} is ${size} bytes; Context Service accepts at most ${grant.maxBytes}` + : undefined; + + let grant = created.upload; + let replaced = false; + let uploaded: UploadResult | undefined; + try { + if (Date.parse(grant.expiresAt) - rt.now() < GRANT_EXPIRY_MARGIN_MS) { + grant = await client.freshUpload(id, signal); + replaced = true; + } + for (;;) { + const problem = tooLarge(grant); + if (problem) return await fail(problem); + try { + uploaded = await client.upload(grant, path, size, signal); + break; + } catch (err) { + // The grant expired in transit, or was already spent: one fresh grant, one more try. + if (!replaced && err instanceof ContextUploadError && err.status === 401) { + io.err('the upload grant was refused; retrying with a fresh one'); + grant = await client.freshUpload(id, signal); + replaced = true; + continue; + } + throw err; + } + } + } catch (err) { + return fail(describeWorkloadError(err, id)); + } + + let active: Workload; + try { + active = await client.activate(id, uploaded.revision, signal); + } catch (err) { + return fail(describeWorkloadError(err, id)); + } + io.err( + `workload ${sanitizeRemote(id)} is ${sanitizeRemote(active.status)} (revision ${uploaded.revision.slice(0, 12)}` + + (uploaded.files !== undefined ? `, ${uploaded.files} files` : '') + + ')', + ); + if (opts.json) { + io.out( + JSON.stringify({ + workloadId: id, + status: active.status, + contextType: active.contextType ?? null, + revision: uploaded.revision, + files: uploaded.files ?? null, + bytes: uploaded.bytes ?? null, + }) + '\n', + ); + } else { + io.out(`${id}\n`); + } + return 0; +} + +// --- get / delete --- + +export interface IdOptions extends WorkloadOptions { + id: string; +} + +function checkId(io: Io, id: string): boolean { + if (validWorkloadId(id)) return true; + io.err(`invalid workload id '${sanitizeRemote(id)}'`); + return false; +} + +export async function cmdWorkloadGet(rt: Runtime, io: Io, opts: IdOptions): Promise { + if (!checkId(io, opts.id)) return 2; + if (!ready(rt, io)) return 2; + const client = opts.client ?? clientFor(rt); + let w: Workload; + try { + w = await client.get(opts.id, opts.signal); + } catch (err) { + if (aborted(opts.signal)) return 130; + io.err(describeWorkloadError(err, opts.id)); + return 1; + } + const view = { + workloadId: w.workloadId, + status: w.status, + contextType: w.contextType ?? null, + workspace: w.workspace ?? null, + revision: w.revision ?? null, + }; + if (opts.json) { + io.out(JSON.stringify(view) + '\n'); + return 0; + } + const rows = [ + ['WORKLOAD', 'STATUS', 'TYPE', 'SIZE', 'REVISION'], + [ + view.workloadId, + view.status, + view.contextType ?? '-', + view.workspace?.size ?? '-', + view.revision ?? '-', + ].map(sanitizeRemote), + ]; + const widths = rows[0]!.map((_, i) => Math.max(...rows.map((r) => r[i]!.length))); + io.out( + rows + .map((r) => r.map((c, i) => (i === r.length - 1 ? c : c.padEnd(widths[i]!))).join(' ')) + .join('\n') + '\n', + ); + return 0; +} + +export async function cmdWorkloadDelete(rt: Runtime, io: Io, opts: IdOptions): Promise { + if (!checkId(io, opts.id)) return 2; + if (!ready(rt, io)) return 2; + const client = opts.client ?? clientFor(rt); + try { + await client.delete(opts.id, opts.signal); + } catch (err) { + if (aborted(opts.signal)) return 130; + io.err(describeWorkloadError(err, opts.id)); + return 1; + } + io.err(`deleted workload ${sanitizeRemote(opts.id)}`); + if (opts.json) io.out(JSON.stringify({ workloadId: opts.id, status: 'deleted' }) + '\n'); + return 0; +} diff --git a/packages/mocactl/test/cli.test.ts b/packages/mocactl/test/cli.test.ts index 5f14b955..eadde43c 100644 --- a/packages/mocactl/test/cli.test.ts +++ b/packages/mocactl/test/cli.test.ts @@ -468,3 +468,69 @@ describe('spawn / subagents / get', () => { expect(USAGE).toContain('mocactl get'); }); }); + +/* ---------------- workloads ---------------- */ + +describe('workloads and spawn --workload', () => { + it('spawn routes --workload through to cmdSpawn; --workload is refused elsewhere', async () => { + const mod = await import('../src/subagent.js'); + const stub = vi.fn(async (..._a: unknown[]) => 0); + const spy = vi.spyOn(mod, 'cmdSpawn').mockImplementation(stub); + onTestFinished(() => spy.mockRestore()); + expect( + await main(['spawn', 'p', '--kind', 'prompt', '--workload', 'demo'], {}, io(), { + buildRuntime: fakeBuild, + }), + ).toBe(0); + expect(stub.mock.calls[0][2]).toMatchObject({ kind: 'prompt', workloadId: 'demo' }); + const o = io(); + expect(await main(['run', 'p', '--workload', 'demo'], {}, o, { buildRuntime: fakeBuild })).toBe( + 2, + ); + expect(o.errs.join('\n')).toContain('--workload only applies to `mocactl spawn`'); + }); + + it('routes workloads create, get and delete; --context and --name only on create', async () => { + const mod = await import('../src/workloads.js'); + const create = vi.fn(async (..._a: unknown[]) => 0); + const get = vi.fn(async (..._a: unknown[]) => 0); + const del = vi.fn(async (..._a: unknown[]) => 0); + const spies = [ + vi.spyOn(mod, 'cmdWorkloadCreate').mockImplementation(create), + vi.spyOn(mod, 'cmdWorkloadGet').mockImplementation(get), + vi.spyOn(mod, 'cmdWorkloadDelete').mockImplementation(del), + ]; + onTestFinished(() => spies.forEach((s) => s.mockRestore())); + const run = (argv: string[], o = io()) => main(argv, {}, o, { buildRuntime: fakeBuild }); + + expect( + await run(['workloads', 'create', '--context', 'ctx.context', '--name', 'demo', '--json']), + ).toBe(0); + expect(create.mock.calls[0][2]).toMatchObject({ + contextPath: 'ctx.context', + name: 'demo', + json: true, + }); + expect(await run(['workloads', 'get', 'demo', '--json'])).toBe(0); + expect(get.mock.calls[0][2]).toMatchObject({ id: 'demo', json: true }); + expect(await run(['workloads', 'delete', 'demo'])).toBe(0); + expect(del.mock.calls[0][2]).toMatchObject({ id: 'demo', json: false }); + + const noContext = io(); + expect(await run(['workloads', 'create'], noContext)).toBe(2); + expect(noContext.errs.join('\n')).toContain('needs --context FILE'); + const noList = io(); + expect(await run(['workloads'], noList)).toBe(2); + expect(noList.errs.join('\n')).toContain('there is no list'); + expect(await run(['workloads', 'list'])).toBe(2); + expect(await run(['workloads', 'get'])).toBe(2); + expect(await run(['workloads', 'delete', 'a', 'b'])).toBe(2); + expect(await run(['workloads', 'get', 'demo', '--context', 'x'])).toBe(2); + expect(await run(['sessions', '--name', 'x'])).toBe(2); + expect(create).toHaveBeenCalledTimes(1); + expect(get).toHaveBeenCalledTimes(1); + expect(del).toHaveBeenCalledTimes(1); + expect(USAGE).toContain('mocactl workloads create --context FILE'); + expect(USAGE).toContain('--workload ID'); + }); +}); diff --git a/packages/mocactl/test/subagent-list.test.ts b/packages/mocactl/test/subagent-list.test.ts index acd4dba3..c5dd9b24 100644 --- a/packages/mocactl/test/subagent-list.test.ts +++ b/packages/mocactl/test/subagent-list.test.ts @@ -203,3 +203,16 @@ describe('cmdSubagents — terminal rows are not re-queried (24h TTL safety)', ( expect(parsed.subagents[0].detail).toBe('boom'); }); }); + +describe('cmdSubagents — workload', () => { + it('reports the workload a row ran on, and null for one without', async () => { + const rt = makeRt(scratch); + const ledger = openLedger(rt.paths.subagentsFile, rt.now); + ledger.append(seedRecord({ sessionId: 'w', workloadId: 'demo', lastKnownState: 'done' })); + ledger.append(seedRecord({ sessionId: 'n', submittedAt: 2, lastKnownState: 'done' })); + const { io, out } = makeIo(); + await cmdSubagents(rt, io, base({ json: true, ledger, transport: {} as LeafTransport })); + const rows = JSON.parse(out[0]!.trim()).subagents; + expect(rows.map((r: { workload: string | null }) => r.workload)).toEqual(['demo', null]); + }); +}); diff --git a/packages/mocactl/test/subagent-spawn.test.ts b/packages/mocactl/test/subagent-spawn.test.ts index 7a6517ca..2fe60896 100644 --- a/packages/mocactl/test/subagent-spawn.test.ts +++ b/packages/mocactl/test/subagent-spawn.test.ts @@ -333,3 +333,65 @@ describe('cmdSpawn — failure paths that must not create a ledger orphan', () = // Imported late to keep the file's top imports stable. type LedgerRecord = ReturnType[number]; + +/* --- --workload --- */ + +describe('cmdSpawn — --workload', () => { + it('sends workloadId in the run envelope and records it in the ledger row', async () => { + const { rt } = makeRt(); + const ledger = withLedger(rt); + const { io } = makeIo(); + let seenEnvelope: LeafEnvelope | undefined; + const forkFn: SpawnOptions['forkFn'] = (_sid, _tok, env) => { + seenEnvelope = env; + return 7; + }; + const r = await cmdSpawn(rt, io, baseOpts({ workloadId: 'demo', forkFn, ledger })); + expect(r).toBe(0); + expect(seenEnvelope).toEqual({ + sessionId: 'sess-abc', + kind: 'prompt', + prompt: 'summarize the repo', + workloadId: 'demo', + }); + const row = openLedger(ledgerPath).readAll()[0]!; + expect(row.workloadId).toBe('demo'); + expect(row.envelope.workloadId).toBe('demo'); + }); + + it('refuses --workload with --kind solve, and an invalid workload id, before creating a session', async () => { + for (const over of [ + { kind: 'solve' as const, repoUrl: 'https://x', ref: 'main', workloadId: 'demo' }, + { workloadId: 'Bad/Id' }, + ]) { + const { rt, cp } = makeRt(); + const { io, err } = makeIo(); + const r = await cmdSpawn( + rt, + io, + baseOpts({ ...over, forkFn: () => 1, ledger: withLedger(rt) }), + ); + expect(r).toBe(2); + expect(cp.createSession).not.toHaveBeenCalled(); + expect(err.join('\n')).toMatch(/--workload needs --kind prompt|invalid workload id/); + } + }); + + it('the worker transport puts workloadId in the POST /v1/runs body', async () => { + const { LeafTransport } = await import('../src/transport/leaf-turn.js'); + const bodies: unknown[] = []; + const fetchImpl = (async (_url: string, init: RequestInit) => { + bodies.push(JSON.parse(String(init.body))); + return new Response(JSON.stringify({ status: 'responded', text: 'ok' }), { + status: 200, + headers: { 'content-type': 'application/json' }, + }); + }) as unknown as typeof fetch; + const transport = new LeafTransport({ baseUrl: async () => 'https://harness.test' }, fetchImpl); + await transport.submitLeafTurn( + { sessionId: 's', kind: 'prompt', prompt: 'p', workloadId: 'demo' }, + 'tok', + ); + expect(bodies[0]).toMatchObject({ kind: 'prompt', workloadId: 'demo' }); + }); +}); diff --git a/packages/mocactl/test/workloads.test.ts b/packages/mocactl/test/workloads.test.ts new file mode 100644 index 00000000..0351d6c4 --- /dev/null +++ b/packages/mocactl/test/workloads.test.ts @@ -0,0 +1,439 @@ +import { createServer, type IncomingMessage } from 'node:http'; +import type { AddressInfo } from 'node:net'; +import { mkdtempSync, writeFileSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { describe, expect, it, onTestFinished } from 'vitest'; +import { uploadUrlProblem } from '../src/api/workloads.js'; +import type { Io } from '../src/headless.js'; +import { cmdWorkloadCreate, cmdWorkloadDelete, cmdWorkloadGet } from '../src/workloads.js'; +import { json, scriptedFetch } from './helpers/fake-fetch.js'; +import { fakeHarness } from './helpers/fakes.js'; +import { testRuntime } from './helpers/runtime.js'; + +function io(): Io & { stdout: string; stderr: string[] } { + const o = { + stdout: '', + stderr: [] as string[], + out: (s: string) => void (o.stdout += s), + err: (s: string) => void o.stderr.push(s), + }; + return o; +} + +const REV = 'c'.repeat(64); +// testRuntime's clock is 1e12 ms (2001); these are either side of it. +const FUTURE = '2030-01-01T00:00:00Z'; +const PAST = '2000-01-01T00:00:00Z'; + +function grant(over: Record = {}) { + return { + uploadUrl: 'https://cs.example/v1/uploads/up1', + token: 'grant-1', + expiresAt: FUTURE, + method: 'PUT', + contentType: 'application/vnd.rossoctl.context', + maxBytes: 1024, + ...over, + }; +} + +const workload = (status: string, over: Record = {}) => ({ + workloadId: 'demo', + status, + contextType: 'workspace', + workspace: { size: '1Gi' }, + ...over, +}); + +function contextFile(bytes = 'context-bytes'): string { + const p = join(mkdtempSync(join(tmpdir(), 'mocactl-ctx-')), 'ctx.context'); + writeFileSync(p, bytes); + return p; +} + +describe('workloads create', () => { + it('creates, uploads the file to the grant, activates the revision, and prints the id', async () => { + const { fetch, calls } = scriptedFetch( + json({ ...workload('awaiting_upload'), upload: grant() }, 201), + json({ revision: REV, files: 3, bytes: 13 }, 201), + json(workload('ready', { revision: REV })), + ); + const rt = testRuntime({ fetchImpl: fetch }); + const o = io(); + const code = await cmdWorkloadCreate(rt, o, { + contextPath: contextFile(), + name: 'demo', + json: false, + }); + expect(code).toBe(0); + expect(o.stdout).toBe('demo\n'); + expect(calls.map((c) => `${c.method} ${c.url}`)).toEqual([ + 'POST http://h/workloads', + 'PUT https://cs.example/v1/uploads/up1', + 'POST http://h/workloads/demo/activate', + ]); + // The harness gets the API token; Context Service only the grant's own token. + expect(calls[0]!.headers.authorization).toBe('Bearer a'); + expect(calls[0]!.body).toEqual({ name: 'demo' }); + expect(calls[1]!.headers).toMatchObject({ + authorization: 'Bearer grant-1', + 'content-type': 'application/vnd.rossoctl.context', + 'content-length': '13', + }); + expect(calls[2]!.body).toEqual({ revision: REV }); + expect(o.stderr.join('\n')).toContain('workload demo is ready'); + }); + + it('prints one JSON document with --json', async () => { + const { fetch } = scriptedFetch( + json({ ...workload('awaiting_upload'), upload: grant() }, 201), + json({ revision: REV, files: 3, bytes: 13 }, 201), + json(workload('ready', { revision: REV })), + ); + const o = io(); + const code = await cmdWorkloadCreate(testRuntime({ fetchImpl: fetch }), o, { + contextPath: contextFile(), + json: true, + }); + expect(code).toBe(0); + expect(JSON.parse(o.stdout)).toEqual({ + workloadId: 'demo', + status: 'ready', + contextType: 'workspace', + revision: REV, + files: 3, + bytes: 13, + }); + expect(o.stdout.trimEnd().split('\n')).toHaveLength(1); + }); + + it('replaces an already-expired grant before uploading', async () => { + const { fetch, calls } = scriptedFetch( + json({ ...workload('awaiting_upload'), upload: grant({ expiresAt: PAST }) }, 201), + json(grant({ token: 'grant-2', uploadUrl: 'https://cs.example/v1/uploads/up2' }), 201), + json({ revision: REV }, 201), + json(workload('ready', { revision: REV })), + ); + const code = await cmdWorkloadCreate(testRuntime({ fetchImpl: fetch }), io(), { + contextPath: contextFile(), + json: false, + }); + expect(code).toBe(0); + expect(calls[1]!.url).toBe('http://h/workloads/demo/uploads'); + expect(calls[2]!.url).toBe('https://cs.example/v1/uploads/up2'); + expect(calls[2]!.headers.authorization).toBe('Bearer grant-2'); + }); + + it('retries once with a fresh grant when the upload answers 401', async () => { + const { fetch, calls } = scriptedFetch( + json({ ...workload('awaiting_upload'), upload: grant() }, 201), + json({ error: 'capability_expired' }, 401), + json(grant({ token: 'grant-2' }), 201), + json({ revision: REV }, 201), + json(workload('ready', { revision: REV })), + ); + const o = io(); + const code = await cmdWorkloadCreate(testRuntime({ fetchImpl: fetch }), o, { + contextPath: contextFile(), + json: false, + }); + expect(code).toBe(0); + expect(calls.map((c) => c.method + ' ' + c.url)).toEqual([ + 'POST http://h/workloads', + 'PUT https://cs.example/v1/uploads/up1', + 'POST http://h/workloads/demo/uploads', + 'PUT https://cs.example/v1/uploads/up1', + 'POST http://h/workloads/demo/activate', + ]); + expect(calls[3]!.headers.authorization).toBe('Bearer grant-2'); + expect(o.stderr.join('\n')).toContain('retrying with a fresh one'); + }); + + it('retries only once, then deletes the workload it created', async () => { + const { fetch, calls } = scriptedFetch( + json({ ...workload('awaiting_upload'), upload: grant() }, 201), + json({ error: 'capability_expired' }, 401), + json(grant({ token: 'grant-2' }), 201), + json({ error: 'capability_expired' }, 401), + new Response(null, { status: 204 }), + ); + const o = io(); + const code = await cmdWorkloadCreate(testRuntime({ fetchImpl: fetch }), o, { + contextPath: contextFile(), + json: false, + }); + expect(code).toBe(1); + expect(calls.at(-1)!.method + ' ' + calls.at(-1)!.url).toBe('DELETE http://h/workloads/demo'); + const err = o.stderr.join('\n'); + expect(err).toContain('Context Service refused the upload: 401 capability_expired'); + expect(err).toContain('deleted workload demo'); + expect(o.stdout).toBe(''); + }); + + it("refuses a file larger than the grant's maxBytes without uploading it", async () => { + const { fetch, calls } = scriptedFetch( + json({ ...workload('awaiting_upload'), upload: grant({ maxBytes: 4 }) }, 201), + new Response(null, { status: 204 }), + ); + const o = io(); + const code = await cmdWorkloadCreate(testRuntime({ fetchImpl: fetch }), o, { + contextPath: contextFile('0123456789'), + json: false, + }); + expect(code).toBe(1); + expect(calls.map((c) => c.method)).toEqual(['POST', 'DELETE']); + expect(o.stderr.join('\n')).toContain('is 10 bytes; Context Service accepts at most 4'); + }); + + it('refuses a missing or empty file, and a bad --name, before any request', async () => { + const { fetch, calls } = scriptedFetch(); + const rt = testRuntime({ fetchImpl: fetch }); + const missing = io(); + expect( + await cmdWorkloadCreate(rt, missing, { + contextPath: '/nonexistent/ctx.context', + json: false, + }), + ).toBe(2); + expect(missing.stderr.join('\n')).toContain('does not exist'); + expect(await cmdWorkloadCreate(rt, io(), { contextPath: contextFile(''), json: false })).toBe( + 2, + ); + expect( + await cmdWorkloadCreate(rt, io(), { + contextPath: contextFile(), + name: 'Not_A_Name', + json: false, + }), + ).toBe(2); + expect(calls).toEqual([]); + }); + + it('refuses when not logged in', async () => { + const o = io(); + const code = await cmdWorkloadCreate(testRuntime({ auth: null }), o, { + contextPath: contextFile(), + json: false, + }); + expect(code).toBe(2); + expect(o.stderr.join('\n')).toContain('not logged in'); + }); + + it("surfaces the harness's error codes", async () => { + const cases: Array<[Response, string]> = [ + [json({ error: 'workloads_unavailable' }, 501), 'workloads are not enabled on this harness'], + [json({ error: 'workload_quota_exceeded' }, 429), 'mocactl workloads delete ID'], + [json({ error: 'workload_name_taken' }, 409), "already have a workload named 'demo'"], + [json({ error: 'context_service_error' }, 502), '502 context_service_error'], + ]; + for (const [reply, message] of cases) { + const { fetch, calls } = scriptedFetch(reply); + const o = io(); + const code = await cmdWorkloadCreate(testRuntime({ fetchImpl: fetch }), o, { + contextPath: contextFile(), + name: 'demo', + json: false, + }); + expect(code).toBe(1); + expect(calls).toHaveLength(1); + expect(o.stderr.join('\n')).toContain(message); + } + }); + + it('surfaces an activation refusal and deletes the workload', async () => { + const { fetch, calls } = scriptedFetch( + json({ ...workload('awaiting_upload'), upload: grant() }, 201), + json({ revision: REV }, 201), + json({ error: 'revision_mismatch' }, 409), + new Response(null, { status: 204 }), + ); + const o = io(); + const code = await cmdWorkloadCreate(testRuntime({ fetchImpl: fetch }), o, { + contextPath: contextFile(), + json: false, + }); + expect(code).toBe(1); + expect(calls.at(-1)!.method).toBe('DELETE'); + expect(o.stderr.join('\n')).toContain('409 revision_mismatch'); + }); + + it('says how to clean up when the rollback delete fails too', async () => { + const { fetch } = scriptedFetch( + json({ ...workload('awaiting_upload'), upload: grant() }, 201), + json({ error: 'internal' }, 500), + json({ error: 'context_service_error' }, 502), + ); + const o = io(); + const code = await cmdWorkloadCreate(testRuntime({ fetchImpl: fetch }), o, { + contextPath: contextFile(), + json: false, + }); + expect(code).toBe(1); + expect(o.stderr.join('\n')).toContain('run `mocactl workloads delete demo`'); + }); + + it('never sends the grant over plain HTTP to a non-loopback host', async () => { + const { fetch, calls } = scriptedFetch( + json( + { + ...workload('awaiting_upload'), + upload: grant({ uploadUrl: 'http://cs.example/v1/uploads/up1' }), + }, + 201, + ), + new Response(null, { status: 204 }), + ); + const o = io(); + const code = await cmdWorkloadCreate(testRuntime({ fetchImpl: fetch }), o, { + contextPath: contextFile(), + json: false, + }); + expect(code).toBe(1); + expect(calls.map((c) => c.method)).toEqual(['POST', 'DELETE']); + expect(o.stderr.join('\n')).toContain('must use HTTPS'); + expect(uploadUrlProblem('https://cs.example/x')).toBeUndefined(); + expect(uploadUrlProblem('http://127.0.0.1:8080/x')).toBeUndefined(); + expect(uploadUrlProblem('http://[::1]/x')).toBeUndefined(); + expect(uploadUrlProblem('http://localhost/x')).toBeUndefined(); + expect(uploadUrlProblem('http://10.0.0.1/x')).toBeDefined(); + expect(uploadUrlProblem('ftp://cs.example/x')).toBeDefined(); + }); + + it('streams the file to a real loopback server with its length and type', async () => { + const seen: Array<{ + method?: string; + url?: string; + headers: IncomingMessage['headers']; + body: Buffer; + }> = []; + const server = createServer((req, res) => { + const chunks: Buffer[] = []; + req.on('data', (c: Buffer) => chunks.push(c)); + req.on('end', () => { + seen.push({ + method: req.method, + url: req.url, + headers: req.headers, + body: Buffer.concat(chunks), + }); + const send = (status: number, body?: unknown) => + res + .writeHead(status, { 'content-type': 'application/json' }) + .end(body === undefined ? undefined : JSON.stringify(body)); + if (req.url === '/workloads') { + const { port } = server.address() as AddressInfo; + return send(201, { + ...workload('awaiting_upload'), + upload: grant({ + uploadUrl: `http://127.0.0.1:${port}/v1/uploads/up1`, + maxBytes: 1 << 20, + }), + }); + } + if (req.url === '/v1/uploads/up1') return send(201, { revision: REV, files: 1, bytes: 5 }); + if (req.url === '/workloads/demo/activate') return send(200, workload('ready')); + send(404, { error: 'not_found' }); + }); + }); + await new Promise((r) => server.listen(0, '127.0.0.1', r)); + onTestFinished(() => new Promise((r) => server.close(() => r()))); + const { port } = server.address() as AddressInfo; + const rt = testRuntime({ + fetchImpl: fetch, + harness: fakeHarness([], { baseUrl: async () => `http://127.0.0.1:${port}` }), + }); + const payload = Buffer.alloc(300_000, 7); + const path = join(mkdtempSync(join(tmpdir(), 'mocactl-ctx-')), 'big.context'); + writeFileSync(path, payload); + const o = io(); + const code = await cmdWorkloadCreate(rt, o, { contextPath: path, json: false }); + expect(o.stderr.join('\n')).not.toContain('refused'); + expect(code).toBe(0); + const put = seen.find((s) => s.method === 'PUT')!; + expect(put.headers['content-length']).toBe(String(payload.length)); + expect(put.headers['content-type']).toBe('application/vnd.rossoctl.context'); + expect(put.headers.authorization).toBe('Bearer grant-1'); + expect(put.body.equals(payload)).toBe(true); + }); +}); + +describe('workloads get', () => { + it('prints the workload as JSON', async () => { + const { fetch, calls } = scriptedFetch(json(workload('ready', { revision: REV, extra: 1 }))); + const o = io(); + expect( + await cmdWorkloadGet(testRuntime({ fetchImpl: fetch }), o, { id: 'demo', json: true }), + ).toBe(0); + expect(calls[0]!.method + ' ' + calls[0]!.url).toBe('GET http://h/workloads/demo'); + expect(calls[0]!.headers.authorization).toBe('Bearer a'); + expect(JSON.parse(o.stdout)).toEqual({ + workloadId: 'demo', + status: 'ready', + contextType: 'workspace', + workspace: { size: '1Gi' }, + revision: REV, + }); + }); + + it('prints a table, with nulls for what an unactivated workload lacks in JSON', async () => { + const { fetch } = scriptedFetch( + json(workload('awaiting_upload')), + json({ workloadId: 'demo', status: 'deleted' }), + ); + const rt = testRuntime({ fetchImpl: fetch }); + const table = io(); + expect(await cmdWorkloadGet(rt, table, { id: 'demo', json: false })).toBe(0); + const lines = table.stdout.trimEnd().split('\n'); + expect(lines[0]).toMatch(/^WORKLOAD\s+STATUS\s+TYPE\s+SIZE\s+REVISION$/); + expect(lines[1]).toMatch(/^demo\s+awaiting_upload\s+workspace\s+1Gi\s+-$/); + const bare = io(); + expect(await cmdWorkloadGet(rt, bare, { id: 'demo', json: true })).toBe(0); + expect(JSON.parse(bare.stdout)).toEqual({ + workloadId: 'demo', + status: 'deleted', + contextType: null, + workspace: null, + revision: null, + }); + }); + + it('exits 1 for a missing workload and 2 for an invalid id', async () => { + const { fetch, calls } = scriptedFetch(json({ error: 'workload_not_found' }, 404)); + const rt = testRuntime({ fetchImpl: fetch }); + const o = io(); + expect(await cmdWorkloadGet(rt, o, { id: 'gone', json: true })).toBe(1); + expect(o.stderr.join('\n')).toContain("no workload 'gone' (or it is not yours)"); + expect(o.stdout).toBe(''); + expect(await cmdWorkloadGet(rt, io(), { id: '../etc', json: false })).toBe(2); + expect(calls).toHaveLength(1); + }); +}); + +describe('workloads delete', () => { + it('deletes and prints {workloadId, status} with --json', async () => { + const { fetch, calls } = scriptedFetch(new Response(null, { status: 204 })); + const o = io(); + expect( + await cmdWorkloadDelete(testRuntime({ fetchImpl: fetch }), o, { id: 'demo', json: true }), + ).toBe(0); + expect(calls[0]!.method + ' ' + calls[0]!.url).toBe('DELETE http://h/workloads/demo'); + expect(calls[0]!.headers.authorization).toBe('Bearer a'); + expect(JSON.parse(o.stdout)).toEqual({ workloadId: 'demo', status: 'deleted' }); + expect(o.stderr.join('\n')).toContain('deleted workload demo'); + }); + + it('surfaces 404 and 409', async () => { + const { fetch } = scriptedFetch( + json({ error: 'workload_not_found' }, 404), + json({ error: 'workload_state_changed' }, 409), + ); + const rt = testRuntime({ fetchImpl: fetch }); + const missing = io(); + expect(await cmdWorkloadDelete(rt, missing, { id: 'demo', json: false })).toBe(1); + expect(missing.stderr.join('\n')).toContain("no workload 'demo'"); + const raced = io(); + expect(await cmdWorkloadDelete(rt, raced, { id: 'demo', json: false })).toBe(1); + expect(raced.stderr.join('\n')).toContain('409 workload_state_changed'); + }); +});