From 757a39aa472c6aff95f0520ad0bc95eec4bf200e Mon Sep 17 00:00:00 2001 From: Nirupam Bhowmick <48842933+heydryft@users.noreply.github.com> Date: Tue, 20 Oct 2026 12:19:18 +0100 Subject: [PATCH 1/3] feat(jobs): runcrate run and runcrate jobs (RUN-486) --- README.md | 24 ++ internal/api/client.go | 9 +- internal/api/instances.go | 34 +- internal/api/jobs.go | 229 ++++++++++++++ internal/api/jobs_test.go | 143 +++++++++ internal/commands/helpers.go | 6 +- internal/commands/jobs.go | 559 +++++++++++++++++++++++++++++++++ internal/commands/jobs_test.go | 381 ++++++++++++++++++++++ internal/commands/root.go | 8 + internal/commands/run.go | 250 +++++++++++++++ 10 files changed, 1633 insertions(+), 10 deletions(-) create mode 100644 internal/api/jobs.go create mode 100644 internal/api/jobs_test.go create mode 100644 internal/commands/jobs.go create mode 100644 internal/commands/jobs_test.go create mode 100644 internal/commands/run.go diff --git a/README.md b/README.md index 28fb2a3..b1bafa2 100644 --- a/README.md +++ b/README.md @@ -35,6 +35,30 @@ runcrate cp ./train.py my-gpu:/root/workspace/ Point the CLI at a different API origin with `runcrate config set url ` or the `RUNCRATE_API_URL` env var (defaults to `https://runcrate.ai`). +## Jobs + +A job runs one command on its own GPU machine, streams its output, and stops the +machine (and its billing) when the command ends. + +```bash +runcrate run --gpu A4000 -- nvidia-smi +runcrate run --gpu H100 --volume --max-runtime 2h -- python train.py --epochs 3 +runcrate run --gpu A4000 --env HF_TOKEN -d -- 'pip install -r requirements.txt && python eval.py' +runcrate jobs list [--status running,queued] [--limit 20] [-q] [--json] +runcrate jobs status # exit code, error, cost, instance, env keys +runcrate jobs logs [-f] # print output; -f follows until the job ends +runcrate jobs cancel +``` + +- The command runs with bash in `/workspace`, which holds the volume's files when + `--volume` is set; files written there are copied back when the machine stops. +- `--max-runtime` takes `90m`, `2h`, `1h30m` or minutes, from 1m to 24h (default 1h). +- `--env KEY=VALUE` is repeatable; `--env KEY` copies the value from your shell. + Values are never stored or shown. +- `runcrate run` and `runcrate jobs logs -f` exit with the job's code: the command's + own code, 124 on timeout, 130 when cancelled. Ctrl-C detaches and leaves the job + running; `-d` / `--detach` prints the job ID and returns. + ## Updating ```bash diff --git a/internal/api/client.go b/internal/api/client.go index fc67afa..86dac01 100644 --- a/internal/api/client.go +++ b/internal/api/client.go @@ -1,6 +1,7 @@ package api import ( + "context" "encoding/json" "fmt" "io" @@ -52,8 +53,14 @@ func NewOAuthClient(baseURL, accessToken, projectID, environment string) *Client } func (c *Client) do(method, path string, body io.Reader) (*http.Response, error) { + return c.doCtx(context.Background(), method, path, body) +} + +// doCtx is do with a request context, for calls that can be cancelled or need +// a deadline other than the client's default timeout. +func (c *Client) doCtx(ctx context.Context, method, path string, body io.Reader) (*http.Response, error) { url := c.baseURL + path - req, err := http.NewRequest(method, url, body) + req, err := http.NewRequestWithContext(ctx, method, url, body) if err != nil { return nil, err } diff --git a/internal/api/instances.go b/internal/api/instances.go index 9131dba..4290c30 100644 --- a/internal/api/instances.go +++ b/internal/api/instances.go @@ -98,7 +98,25 @@ func (c *Client) GetSSHCertificate(instanceID, publicKey string) (*SSHCertificat return &result.Data, nil } -// checkHTTPStatus returns a formatted error for non-2xx responses. +// StatusError is a non-2xx v1 response: `{ "error": { "code", "message", "details"? } }`. +type StatusError struct { + StatusCode int + Code string + Message string + Details map[string]any +} + +func (e *StatusError) Error() string { + if e.StatusCode == http.StatusPaymentRequired && e.Code == "BUDGET_REACHED" { + return "this environment's monthly budget is reached; an owner or admin can raise it in the dashboard (HTTP 402)" + } + if e.Message != "" { + return fmt.Sprintf("%s (HTTP %d)", e.Message, e.StatusCode) + } + return fmt.Sprintf("API error (HTTP %d)", e.StatusCode) +} + +// checkHTTPStatus returns a *StatusError for non-2xx responses. func checkHTTPStatus(resp *http.Response) error { if resp.StatusCode >= 200 && resp.StatusCode < 300 { return nil @@ -106,14 +124,18 @@ func checkHTTPStatus(resp *http.Response) error { var errResp struct { Error struct { - Code string `json:"code"` - Message string `json:"message"` + Code string `json:"code"` + Message string `json:"message"` + Details map[string]any `json:"details"` } `json:"error"` } - if json.NewDecoder(resp.Body).Decode(&errResp) == nil && errResp.Error.Message != "" { - return fmt.Errorf("%s (HTTP %d)", errResp.Error.Message, resp.StatusCode) + e := &StatusError{StatusCode: resp.StatusCode} + if json.NewDecoder(resp.Body).Decode(&errResp) == nil { + e.Code = errResp.Error.Code + e.Message = errResp.Error.Message + e.Details = errResp.Error.Details } - return fmt.Errorf("API error (HTTP %d)", resp.StatusCode) + return e } // ListInstances returns all instances for the authenticated user. diff --git a/internal/api/jobs.go b/internal/api/jobs.go new file mode 100644 index 0000000..70ec9cc --- /dev/null +++ b/internal/api/jobs.go @@ -0,0 +1,229 @@ +package api + +import ( + "context" + "encoding/json" + "fmt" + "net/url" + "strconv" + "strings" + "time" +) + +// Job is one shell command run on its own machine (RUN-486). +type Job struct { + ID string `json:"id"` + Name string `json:"name"` + Status string `json:"status"` + Command string `json:"command"` + GPUType string `json:"gpu_type"` + GPUCount int `json:"gpu_count"` + Template string `json:"template"` + Image string `json:"image"` + EnvKeys []string `json:"env_keys"` + StorageID string `json:"storage_id"` + MaxRuntimeMinutes int `json:"max_runtime_minutes"` + ExitCode *int `json:"exit_code"` + Error string `json:"error"` + InstanceID string `json:"instance_id"` + EnvironmentID string `json:"environment_id"` + CreatedAt string `json:"created_at"` + StartedAt string `json:"started_at"` + EndedAt string `json:"ended_at"` + CostPerHour float64 `json:"cost_per_hour"` + CostUSD *float64 `json:"cost_usd"` + LogBytes int64 `json:"log_bytes"` + LogsTruncated bool `json:"logs_truncated"` +} + +// LogChunk is a raw fragment of a job's combined stdout and stderr. Seq and +// TS are null on the marker chunk that stands where truncated output was dropped. +type LogChunk struct { + Seq *int64 `json:"seq"` + TS *string `json:"ts"` + Text string `json:"text"` +} + +// JobLogs is one page of a job's output. Pass Cursor back as `after`; Done means +// the job has ended and every chunk was read. +type JobLogs struct { + JobID string `json:"job_id"` + Status string `json:"status"` + Chunks []LogChunk `json:"chunks"` + Cursor int64 `json:"cursor"` + Truncated bool `json:"truncated"` + Done bool `json:"done"` +} + +// JobList is one page of jobs, newest first. +type JobList struct { + Jobs []Job + Cursor string + HasMore bool +} + +// CreateJobRequest is the body of POST /api/v1/jobs. +type CreateJobRequest struct { + Command string `json:"command"` + GPUType string `json:"gpu_type"` + MaxRuntimeMinutes int `json:"max_runtime_minutes"` + GPUCount int `json:"gpu_count,omitempty"` + Template string `json:"template,omitempty"` + Image string `json:"image,omitempty"` + Env map[string]string `json:"env,omitempty"` + StorageID string `json:"storage_id,omitempty"` + Name string `json:"name,omitempty"` + Region string `json:"region,omitempty"` +} + +// LongPollTimeout bounds a follow=true logs request. The server holds it for +// up to 25 s, so this leaves room for the response without hanging on a dead +// connection for the client's default timeout. +const LongPollTimeout = 40 * time.Second + +// CreateJob submits a job. When the job was created but its machine could not +// be launched, the *StatusError carries Details["job_id"]. +func (c *Client) CreateJob(req CreateJobRequest) (*Job, error) { + body, err := json.Marshal(req) + if err != nil { + return nil, fmt.Errorf("encoding request: %w", err) + } + + resp, err := c.do("POST", "/api/v1/jobs", strings.NewReader(string(body))) + if err != nil { + return nil, fmt.Errorf("network error: %w", err) + } + defer resp.Body.Close() + + if err := checkHTTPStatus(resp); err != nil { + return nil, err + } + + var result apiEnvelope[Job] + if err := json.NewDecoder(resp.Body).Decode(&result); err != nil { + return nil, fmt.Errorf("parsing response: %w", err) + } + return &result.Data, nil +} + +// ListJobs returns one page of jobs. status is a comma list; empty means all. +// limit <= 0 uses the server default; cursor is meta.cursor from a previous page. +func (c *Client) ListJobs(status string, limit int, cursor string) (*JobList, error) { + q := url.Values{} + if status != "" { + q.Set("status", status) + } + if limit > 0 { + q.Set("limit", strconv.Itoa(limit)) + } + if cursor != "" { + q.Set("cursor", cursor) + } + path := "/api/v1/jobs" + if len(q) > 0 { + path += "?" + q.Encode() + } + + resp, err := c.do("GET", path, nil) + if err != nil { + return nil, fmt.Errorf("network error: %w", err) + } + defer resp.Body.Close() + + if err := checkHTTPStatus(resp); err != nil { + return nil, err + } + + var result struct { + Data []Job `json:"data"` + Meta struct { + Cursor any `json:"cursor"` + HasMore bool `json:"hasMore"` + } `json:"meta"` + } + if err := json.NewDecoder(resp.Body).Decode(&result); err != nil { + return nil, fmt.Errorf("parsing response: %w", err) + } + + list := &JobList{Jobs: result.Data, HasMore: result.Meta.HasMore} + // The cursor is opaque: keep it as the string to send back. + switch v := result.Meta.Cursor.(type) { + case string: + list.Cursor = v + case float64: + list.Cursor = strconv.FormatFloat(v, 'f', -1, 64) + } + return list, nil +} + +// GetJob returns one job. +func (c *Client) GetJob(id string) (*Job, error) { + resp, err := c.do("GET", "/api/v1/jobs/"+url.PathEscape(id), nil) + if err != nil { + return nil, fmt.Errorf("network error: %w", err) + } + defer resp.Body.Close() + + if err := checkHTTPStatus(resp); err != nil { + return nil, err + } + + var result apiEnvelope[Job] + if err := json.NewDecoder(resp.Body).Decode(&result); err != nil { + return nil, fmt.Errorf("parsing response: %w", err) + } + return &result.Data, nil +} + +// GetJobLogs returns chunks after the seq cursor `after`. With follow, the +// server waits up to 25 s for new output or the end, and the request is bounded +// by LongPollTimeout. limit <= 0 uses the server default. Cancelling ctx aborts +// the request. +func (c *Client) GetJobLogs(ctx context.Context, id string, after int64, follow bool, limit int) (*JobLogs, error) { + q := url.Values{} + q.Set("after", strconv.FormatInt(after, 10)) + if follow { + q.Set("follow", "true") + var cancel context.CancelFunc + ctx, cancel = context.WithTimeout(ctx, LongPollTimeout) + defer cancel() + } + if limit > 0 { + q.Set("limit", strconv.Itoa(limit)) + } + + resp, err := c.doCtx(ctx, "GET", "/api/v1/jobs/"+url.PathEscape(id)+"/logs?"+q.Encode(), nil) + if err != nil { + return nil, fmt.Errorf("network error: %w", err) + } + defer resp.Body.Close() + + if err := checkHTTPStatus(resp); err != nil { + return nil, err + } + + var result apiEnvelope[JobLogs] + if err := json.NewDecoder(resp.Body).Decode(&result); err != nil { + return nil, fmt.Errorf("parsing response: %w", err) + } + return &result.Data, nil +} + +// CancelJob stops the job's machine; the job becomes cancelled. +func (c *Client) CancelJob(id string) (*Job, error) { + resp, err := c.do("POST", "/api/v1/jobs/"+url.PathEscape(id)+"/cancel", nil) + if err != nil { + return nil, fmt.Errorf("network error: %w", err) + } + defer resp.Body.Close() + + if err := checkHTTPStatus(resp); err != nil { + return nil, err + } + + var result apiEnvelope[Job] + if err := json.NewDecoder(resp.Body).Decode(&result); err != nil { + return nil, fmt.Errorf("parsing response: %w", err) + } + return &result.Data, nil +} diff --git a/internal/api/jobs_test.go b/internal/api/jobs_test.go new file mode 100644 index 0000000..6e5b20c --- /dev/null +++ b/internal/api/jobs_test.go @@ -0,0 +1,143 @@ +package api + +import ( + "context" + "encoding/json" + "errors" + "net/http" + "net/http/httptest" + "strings" + "testing" +) + +func TestCreateJobSendsContractBody(t *testing.T) { + var body map[string]any + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method != "POST" || r.URL.Path != "/api/v1/jobs" { + w.WriteHeader(http.StatusNotFound) + return + } + json.NewDecoder(r.Body).Decode(&body) + w.WriteHeader(http.StatusCreated) + w.Write([]byte(`{"data":{"id":"job-1","name":"job-job1","status":"queued","gpu_type":"A4000","gpu_count":1,"max_runtime_minutes":90,"exit_code":null,"cost_usd":null}}`)) + })) + defer server.Close() + + job, err := NewClient("rc_live_test", server.URL).CreateJob(CreateJobRequest{ + Command: "python train.py", + GPUType: "A4000", + MaxRuntimeMinutes: 90, + Env: map[string]string{"HF_TOKEN": "secret"}, + StorageID: "vol-1", + }) + if err != nil { + t.Fatalf("CreateJob() error: %v", err) + } + if job.ID != "job-1" || job.Status != "queued" || job.ExitCode != nil || job.CostUSD != nil { + t.Errorf("job = %+v", job) + } + + if body["gpu_type"] != "A4000" || body["command"] != "python train.py" || body["storage_id"] != "vol-1" { + t.Errorf("body = %v", body) + } + if body["max_runtime_minutes"] != float64(90) { + t.Errorf("max_runtime_minutes = %v, want 90", body["max_runtime_minutes"]) + } + env, ok := body["env"].(map[string]any) + if !ok || env["HF_TOKEN"] != "secret" { + t.Errorf("env = %#v, want an object", body["env"]) + } + for _, unset := range []string{"template", "image", "name", "region"} { + if _, ok := body[unset]; ok { + t.Errorf("unset %q was sent: %v", unset, body[unset]) + } + } +} + +func TestCreateJobErrors(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch r.Header.Get("X-Case") { + case "budget": + w.WriteHeader(http.StatusPaymentRequired) + w.Write([]byte(`{"error":{"code":"BUDGET_REACHED","message":"Monthly budget reached"}}`)) + default: + w.WriteHeader(http.StatusServiceUnavailable) + w.Write([]byte(`{"error":{"code":"service_unavailable","message":"No machine available","details":{"job_id":"job-9"}}}`)) + } + })) + defer server.Close() + + do := func(c string) error { + client := NewClient("rc_live_test", server.URL) + client.httpClient.Transport = headerTransport{"X-Case", c} + _, err := client.CreateJob(CreateJobRequest{Command: "true", GPUType: "A4000", MaxRuntimeMinutes: 1}) + return err + } + + err := do("budget") + if err == nil || !strings.Contains(err.Error(), "monthly budget") || !strings.Contains(err.Error(), "owner or admin") { + t.Errorf("budget error = %v", err) + } + + err = do("launch") + var se *StatusError + if !errors.As(err, &se) || se.StatusCode != 503 || se.Details["job_id"] != "job-9" { + t.Errorf("launch failure error = %#v", err) + } + if err.Error() != "No machine available (HTTP 503)" { + t.Errorf("message = %q", err.Error()) + } +} + +type headerTransport struct{ key, value string } + +func (h headerTransport) RoundTrip(r *http.Request) (*http.Response, error) { + r.Header.Set(h.key, h.value) + return http.DefaultTransport.RoundTrip(r) +} + +func TestGetJobLogsQueryAndMarker(t *testing.T) { + var query string + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + query = r.URL.RawQuery + w.Write([]byte(`{"data":{"job_id":"job-1","status":"running","chunks":[{"seq":7,"ts":"2026-10-20T10:01:00.000Z","text":"a"},{"seq":null,"ts":null,"text":"\n[runcrate] output dropped\n"}],"cursor":8,"truncated":true,"done":false}}`)) + })) + defer server.Close() + + logs, err := NewClient("rc_live_test", server.URL).GetJobLogs(context.Background(), "job-1", 6, true, 0) + if err != nil { + t.Fatalf("GetJobLogs() error: %v", err) + } + if query != "after=6&follow=true" { + t.Errorf("query = %q, want after=6&follow=true", query) + } + if logs.Cursor != 8 || !logs.Truncated || len(logs.Chunks) != 2 { + t.Fatalf("logs = %+v", logs) + } + if logs.Chunks[0].Seq == nil || *logs.Chunks[0].Seq != 7 { + t.Errorf("first chunk seq = %v, want 7", logs.Chunks[0].Seq) + } + if logs.Chunks[1].Seq != nil || logs.Chunks[1].TS != nil { + t.Errorf("marker chunk should have null seq and ts: %+v", logs.Chunks[1]) + } +} + +func TestListJobsPassesFiltersAndReadsMeta(t *testing.T) { + var query string + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + query = r.URL.RawQuery + w.Write([]byte(`{"data":[{"id":"job-2","status":"running"},{"id":"job-1","status":"queued"}],"meta":{"cursor":"abc","hasMore":true}}`)) + })) + defer server.Close() + + list, err := NewClient("rc_live_test", server.URL).ListJobs("running,queued", 2, "prev") + if err != nil { + t.Fatalf("ListJobs() error: %v", err) + } + if query != "cursor=prev&limit=2&status=running%2Cqueued" { + t.Errorf("query = %q", query) + } + if len(list.Jobs) != 2 || list.Cursor != "abc" || !list.HasMore { + t.Errorf("list = %+v", list) + } +} diff --git a/internal/commands/helpers.go b/internal/commands/helpers.go index dd437d9..d537cf1 100644 --- a/internal/commands/helpers.go +++ b/internal/commands/helpers.go @@ -75,11 +75,11 @@ func execOrRun(bin string, argv []string) error { // statusColor returns the status with a simple text indicator. func statusColor(status string) string { switch strings.ToLower(status) { - case "running", "deployed": + case "running", "deployed", "succeeded": return "● " + status - case "creating", "deploying", "awaiting_provider": + case "creating", "deploying", "awaiting_provider", "queued", "starting": return "◌ " + status - case "terminated", "failed": + case "terminated", "failed", "cancelled", "timed_out": return "✕ " + status default: return " " + status diff --git a/internal/commands/jobs.go b/internal/commands/jobs.go new file mode 100644 index 0000000..d1e494e --- /dev/null +++ b/internal/commands/jobs.go @@ -0,0 +1,559 @@ +package commands + +import ( + "context" + "errors" + "fmt" + "io" + "net/http" + "os" + "strings" + "time" + + "github.com/runcrate/cli/internal/api" + "github.com/spf13/cobra" +) + +// ── follow ────────────────────────────────────────────────────────────────── + +// errDetached reports that the follow context was cancelled (Ctrl-C). +var errDetached = errors.New("detached") + +// jobFollower streams a job's output by long-polling its logs with the cursor. +type jobFollower struct { + // client is called before every request, so a long follow picks up + // refreshed sign-in tokens. + client func() (*api.Client, error) + stdout io.Writer + stderr io.Writer + retryDelay func(attempt int) time.Duration + // idleDelay spaces out polls that return no output, in case the server + // answers at once instead of holding the request. + idleDelay time.Duration + + atLineStart bool + announced map[string]bool +} + +func newJobFollower() *jobFollower { + return &jobFollower{ + client: loadClient, + stdout: os.Stdout, + stderr: os.Stderr, + retryDelay: backoff, + idleDelay: time.Second, + } +} + +// backoff waits 1s, 2s, 4s ... up to 15s between retries. +func backoff(attempt int) time.Duration { + d := time.Second << min(attempt-1, 4) + return min(d, 15*time.Second) +} + +// retryable reports whether a failed request is worth repeating: network +// errors, rate limits and server errors. 401 is retried a few times because +// the client is rebuilt (and its token refreshed) before each attempt. +func retryable(err error, authFailures int) bool { + var se *api.StatusError + if !errors.As(err, &se) { + return true + } + switch { + case se.StatusCode == http.StatusUnauthorized: + return authFailures <= 3 + case se.StatusCode == http.StatusTooManyRequests, se.StatusCode >= 500: + return true + } + return false +} + +// call runs fn with a fresh client, retrying transient failures until it +// succeeds, fails for good, or ctx is cancelled (errDetached). +func (f *jobFollower) call(ctx context.Context, fn func(*api.Client) error) error { + failures, authFailures := 0, 0 + for { + if ctx.Err() != nil { + return errDetached + } + client, err := f.client() + if err != nil { + return err + } + err = fn(client) + if err == nil { + return nil + } + if ctx.Err() != nil { + return errDetached + } + var se *api.StatusError + if errors.As(err, &se) && se.StatusCode == http.StatusUnauthorized { + authFailures++ + } + if !retryable(err, authFailures) { + return err + } + failures++ + if failures == 1 { + f.lineBreak() + fmt.Fprintf(f.stderr, "Lost the connection to Runcrate (%v). Retrying; the job keeps running.\n", err) + } + if !sleepCtx(ctx, f.retryDelay(failures)) { + return errDetached + } + } +} + +func sleepCtx(ctx context.Context, d time.Duration) bool { + if d <= 0 { + return ctx.Err() == nil + } + t := time.NewTimer(d) + defer t.Stop() + select { + case <-ctx.Done(): + return false + case <-t.C: + return true + } +} + +// lineBreak ends a partial output line on stderr so status lines start clean. +// stdout stays byte-for-byte the job's output. +func (f *jobFollower) lineBreak() { + if !f.atLineStart { + fmt.Fprintln(f.stderr) + f.atLineStart = true + } +} + +func (f *jobFollower) announce(status string) { + var msg string + switch status { + case "starting": + msg = "Machine is starting and running setup." + case "running": + msg = "Command started." + default: + return + } + if f.announced[status] { + return + } + f.announced[status] = true + f.lineBreak() + fmt.Fprintln(f.stderr, msg) +} + +// follow writes the job's output from seq cursor `after` until the job ends, +// then returns the final job. On ctx cancel it returns errDetached. +func (f *jobFollower) follow(ctx context.Context, id string, after int64) (*api.Job, error) { + f.atLineStart = true + f.announced = map[string]bool{} + for { + var logs *api.JobLogs + err := f.call(ctx, func(c *api.Client) error { + var err error + logs, err = c.GetJobLogs(ctx, id, after, true, 0) + return err + }) + if err != nil { + return nil, err + } + + f.announce(logs.Status) + for _, ch := range logs.Chunks { + if ch.Text == "" { + continue + } + if _, err := io.WriteString(f.stdout, ch.Text); err != nil { + return nil, err + } + f.atLineStart = strings.HasSuffix(ch.Text, "\n") + } + if logs.Cursor > after { + after = logs.Cursor + } + if logs.Done { + break + } + if len(logs.Chunks) == 0 && !sleepCtx(ctx, f.idleDelay) { + return nil, errDetached + } + } + + var job *api.Job + err := f.call(ctx, func(c *api.Client) error { + var err error + job, err = c.GetJob(id) + return err + }) + if err != nil { + return nil, err + } + f.lineBreak() + return job, nil +} + +// streamJob follows a job to its end and prints the summary to stderr. It +// returns the exit code for the CLI; on detach it prints how to re-attach or +// cancel and returns 0. It never cancels the job. +func streamJob(ctx context.Context, f *jobFollower, id string) (int, error) { + job, err := f.follow(ctx, id, 0) + if errors.Is(err, errDetached) { + f.lineBreak() + fmt.Fprintln(f.stderr, "Detached. The job keeps running.") + fmt.Fprintf(f.stderr, " Follow: runcrate jobs logs -f %s\n", id) + fmt.Fprintf(f.stderr, " Cancel: runcrate jobs cancel %s\n", id) + return 0, nil + } + if err != nil { + f.lineBreak() + return 0, fmt.Errorf("following job: %w\n The job keeps running. Re-attach with: runcrate jobs logs -f %s", err, id) + } + fmt.Fprintln(f.stderr, jobSummary(job)) + return jobExitCode(job), nil +} + +// jobExitCode maps a finished job to the CLI's exit code. +func jobExitCode(j *api.Job) int { + switch j.Status { + case "succeeded": + return 0 + case "failed": + if j.ExitCode != nil && *j.ExitCode > 0 && *j.ExitCode < 256 { + return *j.ExitCode + } + return 1 + case "timed_out": + return 124 + case "cancelled": + return 130 + } + return 1 +} + +// jobSummary is the one-line final status: outcome, exit code, runtime, cost. +func jobSummary(j *api.Job) string { + var head string + switch j.Status { + case "succeeded": + head = "Job succeeded" + case "failed": + head = "Job failed" + if j.ExitCode != nil { + head += fmt.Sprintf(" (exit code %d)", *j.ExitCode) + } + case "timed_out": + head = fmt.Sprintf("Job timed out after its max runtime of %s", formatMinutes(j.MaxRuntimeMinutes)) + case "cancelled": + head = "Job cancelled" + default: + head = "Job " + j.Status + } + if j.Error != "" { + head += ": " + strings.TrimSuffix(j.Error, ".") + } + head += "." + + var details []string + if d, ok := jobRuntime(j, time.Now()); ok { + details = append(details, "ran "+formatDuration(d)) + } + if j.CostUSD != nil { + details = append(details, "cost "+formatCost(*j.CostUSD)) + } + if len(details) == 0 { + return head + } + s := strings.Join(details, ", ") + return head + " " + strings.ToUpper(s[:1]) + s[1:] + "." +} + +// jobRuntime is how long the command has run: started_at to ended_at, or to +// now while it runs. ok is false before it starts. +func jobRuntime(j *api.Job, now time.Time) (time.Duration, bool) { + start, err := time.Parse(time.RFC3339, j.StartedAt) + if err != nil { + return 0, false + } + end := now + if j.EndedAt != "" { + if e, err := time.Parse(time.RFC3339, j.EndedAt); err == nil { + end = e + } + } + if end.Before(start) { + return 0, true + } + return end.Sub(start), true +} + +// formatDuration renders 45s, 12m5s, 2h3m. +func formatDuration(d time.Duration) string { + if d >= time.Hour { + d = d.Round(time.Minute) + return strings.TrimSuffix(d.String(), "0s") + } + return d.Round(time.Second).String() +} + +// formatShortDuration renders a coarse age: 45s, 12m, 3h, 2d. +func formatShortDuration(d time.Duration) string { + switch { + case d < time.Minute: + return fmt.Sprintf("%ds", int(d.Seconds())) + case d < time.Hour: + return fmt.Sprintf("%dm", int(d.Minutes())) + case d < 48*time.Hour: + h := int(d.Hours()) + if m := int(d.Minutes()) % 60; m > 0 { + return fmt.Sprintf("%dh%dm", h, m) + } + return fmt.Sprintf("%dh", h) + } + return fmt.Sprintf("%dd", int(d.Hours()/24)) +} + +func formatCost(c float64) string { + if c > 0 && c < 0.01 { + return fmt.Sprintf("$%.4f", c) + } + return fmt.Sprintf("$%.2f", c) +} + +// ── commands ──────────────────────────────────────────────────────────────── + +var jobsCmd = &cobra.Command{ + Use: "jobs", + Aliases: []string{"job"}, + Short: "List, follow and cancel jobs started with runcrate run", +} + +var jobsListCmd = &cobra.Command{ + Use: "list", + Aliases: []string{"ls"}, + Short: "List jobs, newest first", + RunE: func(cmd *cobra.Command, args []string) error { + client, err := loadClient() + if err != nil { + return err + } + + status, _ := cmd.Flags().GetString("status") + limit, _ := cmd.Flags().GetInt("limit") + list, err := client.ListJobs(strings.ReplaceAll(status, " ", ""), limit, "") + if err != nil { + return fmt.Errorf("fetching jobs: %w", err) + } + + if jsonOutput { + return printJSON(list.Jobs) + } + + quiet, _ := cmd.Flags().GetBool("quiet") + if quiet { + for _, j := range list.Jobs { + fmt.Println(j.ID) + } + return nil + } + + if len(list.Jobs) == 0 { + fmt.Println("No jobs found. Start one with: runcrate run --gpu A4000 -- ") + return nil + } + + now := time.Now() + w := newTable() + fmt.Fprintf(w, "ID\tNAME\tSTATUS\tGPU\tEXIT\tRUNTIME\tAGE\tCOST\n") + for _, j := range list.Jobs { + gpu := j.GPUType + if j.GPUCount > 1 { + gpu = fmt.Sprintf("%dx %s", j.GPUCount, j.GPUType) + } + exit := "-" + if j.ExitCode != nil { + exit = fmt.Sprintf("%d", *j.ExitCode) + } + runtime := "-" + if d, ok := jobRuntime(&j, now); ok { + runtime = formatShortDuration(d) + } + age := "-" + if c, err := time.Parse(time.RFC3339, j.CreatedAt); err == nil { + age = formatShortDuration(now.Sub(c)) + " ago" + } + cost := "-" + if j.CostUSD != nil { + cost = formatCost(*j.CostUSD) + } + fmt.Fprintf(w, "%s\t%s\t%s\t%s\t%s\t%s\t%s\t%s\n", + j.ID, truncate(j.Name, 24), statusColor(j.Status), gpu, exit, runtime, age, cost) + } + if err := w.Flush(); err != nil { + return err + } + if list.HasMore { + fmt.Fprintf(os.Stderr, "Showing the newest %d jobs. See more with --limit (up to 100).\n", len(list.Jobs)) + } + return nil + }, +} + +var jobsStatusCmd = &cobra.Command{ + Use: "status ", + Aliases: []string{"info"}, + Short: "Show a job's details", + Args: cobra.ExactArgs(1), + RunE: func(cmd *cobra.Command, args []string) error { + client, err := loadClient() + if err != nil { + return err + } + + j, err := client.GetJob(args[0]) + if err != nil { + return fmt.Errorf("fetching job: %w", err) + } + + if jsonOutput { + return printJSON(j) + } + + fmt.Println() + fmt.Printf(" Name: %s\n", j.Name) + fmt.Printf(" ID: %s\n", j.ID) + fmt.Printf(" Status: %s\n", statusColor(j.Status)) + if j.ExitCode != nil { + fmt.Printf(" Exit code: %d\n", *j.ExitCode) + } + if j.Error != "" { + fmt.Printf(" Error: %s\n", j.Error) + } + fmt.Printf(" Command: %s\n", j.Command) + fmt.Printf(" GPU: %dx %s\n", j.GPUCount, j.GPUType) + if j.Image != "" { + fmt.Printf(" Image: %s\n", j.Image) + } else if j.Template != "" { + fmt.Printf(" Template: %s\n", j.Template) + } + if j.StorageID != "" { + fmt.Printf(" Volume: %s\n", j.StorageID) + } + if len(j.EnvKeys) > 0 { + fmt.Printf(" Env: %s\n", strings.Join(j.EnvKeys, ", ")) + } + fmt.Printf(" Max runtime: %s\n", formatMinutes(j.MaxRuntimeMinutes)) + if d, ok := jobRuntime(j, time.Now()); ok { + fmt.Printf(" Runtime: %s\n", formatDuration(d)) + } + if j.CostUSD != nil { + cost := formatCost(*j.CostUSD) + if j.EndedAt == "" { + cost += " so far" + } + if j.CostPerHour > 0 { + cost += fmt.Sprintf(" ($%.2f/hr)", j.CostPerHour) + } + fmt.Printf(" Cost: %s\n", cost) + } else if j.CostPerHour > 0 { + fmt.Printf(" Cost: $%.2f/hr\n", j.CostPerHour) + } + if j.InstanceID != "" { + fmt.Printf(" Instance: %s\n", j.InstanceID) + } + fmt.Printf(" Created: %s\n", j.CreatedAt) + if j.StartedAt != "" { + fmt.Printf(" Started: %s\n", j.StartedAt) + } + if j.EndedAt != "" { + fmt.Printf(" Ended: %s\n", j.EndedAt) + } + if j.LogsTruncated { + fmt.Printf(" Logs: %.1f MB, middle dropped (kept the first 5 MB and the newest output)\n", float64(j.LogBytes)/(1<<20)) + } + fmt.Println() + return nil + }, +} + +var jobsLogsCmd = &cobra.Command{ + Use: "logs ", + Short: "Print a job's output (-f to follow until it ends)", + Args: cobra.ExactArgs(1), + RunE: func(cmd *cobra.Command, args []string) error { + cmd.SilenceUsage = true + follow, _ := cmd.Flags().GetBool("follow") + if follow { + return followAndExit(cmd, args[0]) + } + + client, err := loadClient() + if err != nil { + return err + } + return printAllLogs(cmd.Context(), client, args[0], os.Stdout) + }, +} + +// printAllLogs pages through every chunk from the start until the cursor +// stops advancing. +func printAllLogs(ctx context.Context, client *api.Client, id string, out io.Writer) error { + var after int64 + for { + logs, err := client.GetJobLogs(ctx, id, after, false, 1000) + if err != nil { + return fmt.Errorf("fetching logs: %w", err) + } + for _, ch := range logs.Chunks { + if _, err := io.WriteString(out, ch.Text); err != nil { + return err + } + } + if logs.Done || logs.Cursor <= after { + return nil + } + after = logs.Cursor + } +} + +var jobsCancelCmd = &cobra.Command{ + Use: "cancel ", + Short: "Cancel a job and stop its machine", + Args: cobra.ExactArgs(1), + RunE: func(cmd *cobra.Command, args []string) error { + client, err := loadClient() + if err != nil { + return err + } + + j, err := client.CancelJob(args[0]) + if err != nil { + return fmt.Errorf("cancelling job: %w", err) + } + + if jsonOutput { + return printJSON(j) + } + fmt.Printf("Job %s is %s. Its machine is stopping.\n", j.Name, j.Status) + return nil + }, +} + +func init() { + jobsCmd.PersistentFlags().BoolVar(&jsonOutput, "json", false, "Output as JSON") + + jobsListCmd.Flags().String("status", "", "Only these statuses, comma separated (e.g. running,queued)") + jobsListCmd.Flags().Int("limit", 20, "How many jobs to show (max 100)") + jobsListCmd.Flags().BoolP("quiet", "q", false, "Only show job IDs") + + jobsLogsCmd.Flags().BoolP("follow", "f", false, "Follow the output until the job ends; exit with its exit code") + + jobsCmd.AddCommand(jobsListCmd) + jobsCmd.AddCommand(jobsStatusCmd) + jobsCmd.AddCommand(jobsLogsCmd) + jobsCmd.AddCommand(jobsCancelCmd) + rootCmd.AddCommand(jobsCmd) +} diff --git a/internal/commands/jobs_test.go b/internal/commands/jobs_test.go new file mode 100644 index 0000000..abe06dd --- /dev/null +++ b/internal/commands/jobs_test.go @@ -0,0 +1,381 @@ +package commands + +import ( + "bytes" + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "strconv" + "strings" + "sync" + "testing" + "time" + + "github.com/runcrate/cli/internal/api" +) + +func TestParseMaxRuntime(t *testing.T) { + tests := []struct { + in string + want int + wantErr bool + }{ + {in: "90m", want: 90}, + {in: "2h", want: 120}, + {in: "1h30m", want: 90}, + {in: "45", want: 45}, + {in: "90s", want: 2}, + {in: "1s", want: 1}, + {in: "24h", want: 1440}, + {in: "1440", want: 1440}, + {in: "0", wantErr: true}, + {in: "0s", wantErr: true}, + {in: "-5m", wantErr: true}, + {in: "25h", wantErr: true}, + {in: "1441", wantErr: true}, + {in: "two hours", wantErr: true}, + } + for _, tt := range tests { + t.Run(tt.in, func(t *testing.T) { + got, err := parseMaxRuntime(tt.in) + if tt.wantErr { + if err == nil { + t.Fatalf("parseMaxRuntime(%q) = %d, want error", tt.in, got) + } + return + } + if err != nil || got != tt.want { + t.Fatalf("parseMaxRuntime(%q) = %d, %v; want %d", tt.in, got, err, tt.want) + } + }) + } +} + +func TestJoinCommand(t *testing.T) { + tests := []struct { + args []string + want string + }{ + {[]string{"pip install -r req.txt && python train.py"}, "pip install -r req.txt && python train.py"}, + {[]string{"python", "-c", "print(1)"}, "python -c 'print(1)'"}, + {[]string{"ls", "-la", "/workspace/data_v2"}, "ls -la /workspace/data_v2"}, + {[]string{"echo", "it's here"}, `echo 'it'"'"'s here'`}, + {[]string{"echo", "$HOME", ""}, `echo '$HOME' ''`}, + } + for _, tt := range tests { + if got := joinCommand(tt.args); got != tt.want { + t.Errorf("joinCommand(%q) = %q, want %q", tt.args, got, tt.want) + } + } +} + +func TestBuildJobRequestBody(t *testing.T) { + lookup := func(k string) (string, bool) { + if k == "HF_TOKEN" { + return "from-shell", true + } + return "", false + } + parse := func(argv ...string) (api.CreateJobRequest, error) { + cmd := newRunCmd() + if err := cmd.ParseFlags(argv); err != nil { + t.Fatalf("ParseFlags(%q): %v", argv, err) + } + return buildJobRequest(cmd, cmd.Flags().Args(), lookup) + } + + req, err := parse("--gpu", "A4000", "--max-runtime", "90s", "--volume", "vol-1", + "--env", "A=1,2", "--env", "HF_TOKEN", "--", "python", "-c", "print(1)") + if err != nil { + t.Fatalf("buildJobRequest: %v", err) + } + raw, _ := json.Marshal(req) + var body map[string]any + json.Unmarshal(raw, &body) + want := map[string]any{ + "command": "python -c 'print(1)'", + "gpu_type": "A4000", + "gpu_count": float64(1), + "max_runtime_minutes": float64(2), + "storage_id": "vol-1", + "env": map[string]any{"A": "1,2", "HF_TOKEN": "from-shell"}, + } + if !jsonEqual(body, want) { + t.Errorf("body = %s\nwant %v", raw, want) + } + + // Defaults to 60 minutes; flags stop at the command, so its own flags pass through without --. + req, err = parse("--gpu", "H100:2", "nvidia-smi", "-L") + if err != nil { + t.Fatalf("buildJobRequest without --: %v", err) + } + if req.MaxRuntimeMinutes != 60 || req.Command != "nvidia-smi -L" || req.GPUType != "H100" || req.GPUCount != 2 { + t.Errorf("req = %+v", req) + } + + for name, argv := range map[string][]string{ + "no gpu": {"--", "true"}, + "no command": {"--gpu", "A4000"}, + "bad env key": {"--gpu", "A4000", "--env", "1A=x", "--", "true"}, + "unset env": {"--gpu", "A4000", "--env", "MISSING", "--", "true"}, + "runtime too long": {"--gpu", "A4000", "--max-runtime", "25h", "--", "true"}, + } { + if _, err := parse(argv...); err == nil { + t.Errorf("%s: expected an error", name) + } + } +} + +func jsonEqual(a, b any) bool { + x, _ := json.Marshal(a) + y, _ := json.Marshal(b) + return bytes.Equal(x, y) +} + +// ── follow loop ───────────────────────────────────────────────────────────── + +type logStep struct { + wantAfter int64 + status int // HTTP status to answer instead of the page, when set + resp api.JobLogs +} + +type fakeJobsServer struct { + t *testing.T + mu sync.Mutex + steps []logStep + n int + job api.Job + follows []string + cancels int + finalGet int + held int // long polls held open until the client went away +} + +func (f *fakeJobsServer) ServeHTTP(w http.ResponseWriter, r *http.Request) { + f.mu.Lock() + switch { + case r.Method == "GET" && r.URL.Path == "/api/v1/jobs/job-1/logs": + if f.n >= len(f.steps) { + f.held++ + f.mu.Unlock() + // Hold the long poll like the server does until the client goes away. + <-r.Context().Done() + return + } + step := f.steps[f.n] + f.n++ + f.follows = append(f.follows, r.URL.Query().Get("follow")) + f.mu.Unlock() + after, _ := strconv.ParseInt(r.URL.Query().Get("after"), 10, 64) + if after != step.wantAfter { + f.t.Errorf("logs request %d: after = %d, want %d", f.n, after, step.wantAfter) + } + if step.status != 0 { + w.WriteHeader(step.status) + w.Write([]byte(`{"error":{"code":"x","message":"try later"}}`)) + return + } + json.NewEncoder(w).Encode(map[string]any{"data": step.resp}) + return + case r.Method == "GET" && r.URL.Path == "/api/v1/jobs/job-1": + f.finalGet++ + json.NewEncoder(w).Encode(map[string]any{"data": f.job}) + case r.Method == "POST" && r.URL.Path == "/api/v1/jobs/job-1/cancel": + f.cancels++ + w.WriteHeader(http.StatusInternalServerError) + default: + w.WriteHeader(http.StatusNotFound) + } + f.mu.Unlock() +} + +func seq(n int64) *int64 { return &n } + +func chunk(n int64, text string) api.LogChunk { return api.LogChunk{Seq: seq(n), Text: text} } + +func newTestFollower(url string, stdout, stderr *bytes.Buffer) *jobFollower { + return &jobFollower{ + client: func() (*api.Client, error) { return api.NewClient("rc_live_test", url), nil }, + stdout: stdout, + stderr: stderr, + retryDelay: func(int) time.Duration { return 0 }, + } +} + +func intp(n int) *int { return &n } + +func TestStreamJobPrintsChunksInOrderAndExitsWithJobCode(t *testing.T) { + cost := 0.04 + fake := &fakeJobsServer{t: t, steps: []logStep{ + {wantAfter: 0, resp: api.JobLogs{Status: "starting", Chunks: []api.LogChunk{chunk(1, "hello "), chunk(2, "wor")}, Cursor: 2}}, + {wantAfter: 2, resp: api.JobLogs{Status: "running", Cursor: 2}}, + {wantAfter: 2, status: http.StatusServiceUnavailable}, + {wantAfter: 2, resp: api.JobLogs{Status: "running", Chunks: []api.LogChunk{ + chunk(3, "ld\n"), {Text: "\n[runcrate] output dropped\n"}, chunk(9, "tail"), + }, Cursor: 9, Truncated: true}}, + {wantAfter: 9, resp: api.JobLogs{Status: "failed", Cursor: 9, Done: true}}, + }, job: api.Job{ + ID: "job-1", Status: "failed", ExitCode: intp(3), CostUSD: &cost, + StartedAt: "2026-10-20T10:00:00.000Z", EndedAt: "2026-10-20T10:03:12.000Z", + }} + server := httptest.NewServer(fake) + defer server.Close() + + var stdout, stderr bytes.Buffer + code, err := streamJob(context.Background(), newTestFollower(server.URL, &stdout, &stderr), "job-1") + if err != nil { + t.Fatalf("streamJob: %v", err) + } + if code != 3 { + t.Errorf("exit code = %d, want 3", code) + } + if got, want := stdout.String(), "hello world\n\n[runcrate] output dropped\ntail"; got != want { + t.Errorf("stdout = %q, want %q", got, want) + } + for _, f := range fake.follows { + if f != "true" { + t.Errorf("follow = %q, want true on every request", f) + } + } + errOut := stderr.String() + for _, want := range []string{"Machine is starting", "Command started.", "Lost the connection", "\nJob failed (exit code 3). Ran 3m12s, cost $0.04.\n"} { + if strings.Count(errOut, want) != 1 { + t.Errorf("stderr should contain %q once:\n%s", want, errOut) + } + } +} + +func TestStreamJobExitCodes(t *testing.T) { + tests := []struct { + status string + exit *int + want int + }{ + {"succeeded", intp(0), 0}, + {"failed", intp(2), 2}, + {"failed", nil, 1}, + {"timed_out", nil, 124}, + {"cancelled", nil, 130}, + } + for _, tt := range tests { + t.Run(tt.status, func(t *testing.T) { + fake := &fakeJobsServer{t: t, steps: []logStep{ + {wantAfter: 0, resp: api.JobLogs{Status: tt.status, Chunks: []api.LogChunk{chunk(1, "out\n")}, Cursor: 1, Done: true}}, + }, job: api.Job{ID: "job-1", Status: tt.status, ExitCode: tt.exit, MaxRuntimeMinutes: 60}} + server := httptest.NewServer(fake) + defer server.Close() + + var stdout, stderr bytes.Buffer + code, err := streamJob(context.Background(), newTestFollower(server.URL, &stdout, &stderr), "job-1") + if err != nil { + t.Fatalf("streamJob: %v", err) + } + if code != tt.want { + t.Errorf("exit code = %d, want %d", code, tt.want) + } + if fake.n != 1 || fake.finalGet != 1 { + t.Errorf("logs requests = %d, job reads = %d; want 1 and 1 after done", fake.n, fake.finalGet) + } + }) + } +} + +func TestStreamJobStopsOnPermanentError(t *testing.T) { + fake := &fakeJobsServer{t: t, steps: []logStep{{wantAfter: 0, status: http.StatusNotFound}}} + server := httptest.NewServer(fake) + defer server.Close() + + var stdout, stderr bytes.Buffer + _, err := streamJob(context.Background(), newTestFollower(server.URL, &stdout, &stderr), "job-1") + if err == nil || !strings.Contains(err.Error(), "runcrate jobs logs -f job-1") { + t.Fatalf("err = %v, want a not-found error with the re-attach hint", err) + } + if fake.n != 1 { + t.Errorf("a 404 was retried: %d requests", fake.n) + } +} + +// cancelOnWrite cancels the follow context shortly after output arrives, once +// the next long poll is open, standing in for Ctrl-C. +type cancelOnWrite struct { + buf bytes.Buffer + cancel context.CancelFunc +} + +func (c *cancelOnWrite) Write(p []byte) (int, error) { + time.AfterFunc(100*time.Millisecond, c.cancel) + return c.buf.Write(p) +} + +func TestStreamJobDetachDoesNotCancel(t *testing.T) { + fake := &fakeJobsServer{t: t, steps: []logStep{ + {wantAfter: 0, resp: api.JobLogs{Status: "running", Chunks: []api.LogChunk{chunk(1, "step 1\n")}, Cursor: 1}}, + }} + server := httptest.NewServer(fake) + defer server.Close() + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + var stderr bytes.Buffer + stdout := &cancelOnWrite{cancel: cancel} + f := newTestFollower(server.URL, nil, &stderr) + f.stdout = stdout + + done := make(chan struct{}) + var code int + var err error + go func() { + code, err = streamJob(ctx, f, "job-1") + close(done) + }() + select { + case <-done: + case <-time.After(5 * time.Second): + t.Fatal("streamJob did not return after the context was cancelled") + } + + if err != nil || code != 0 { + t.Fatalf("streamJob = %d, %v; want 0, nil", code, err) + } + if stdout.buf.String() != "step 1\n" { + t.Errorf("stdout = %q", stdout.buf.String()) + } + for _, want := range []string{"Detached. The job keeps running.", "runcrate jobs logs -f job-1", "runcrate jobs cancel job-1"} { + if !strings.Contains(stderr.String(), want) { + t.Errorf("stderr missing %q:\n%s", want, stderr.String()) + } + } + fake.mu.Lock() + defer fake.mu.Unlock() + if fake.cancels != 0 { + t.Errorf("detach called /cancel %d times", fake.cancels) + } + if fake.held != 1 { + t.Errorf("held long polls = %d, want 1 (Ctrl-C should abort the open poll)", fake.held) + } +} + +func TestPrintAllLogsPagesUntilCursorStops(t *testing.T) { + fake := &fakeJobsServer{t: t, steps: []logStep{ + {wantAfter: 0, resp: api.JobLogs{Status: "running", Chunks: []api.LogChunk{chunk(1, "a"), chunk(2, "b")}, Cursor: 2}}, + {wantAfter: 2, resp: api.JobLogs{Status: "running", Chunks: []api.LogChunk{chunk(3, "c")}, Cursor: 3}}, + {wantAfter: 3, resp: api.JobLogs{Status: "running", Cursor: 3}}, + }} + server := httptest.NewServer(fake) + defer server.Close() + + var out bytes.Buffer + if err := printAllLogs(context.Background(), api.NewClient("rc_live_test", server.URL), "job-1", &out); err != nil { + t.Fatalf("printAllLogs: %v", err) + } + if out.String() != "abc" { + t.Errorf("out = %q, want abc", out.String()) + } + for _, f := range fake.follows { + if f != "" { + t.Errorf("printAllLogs sent follow=%q", f) + } + } +} diff --git a/internal/commands/root.go b/internal/commands/root.go index 0526b93..a179938 100644 --- a/internal/commands/root.go +++ b/internal/commands/root.go @@ -69,6 +69,14 @@ func printHelp() { fmt.Println(" instances status Live status from provider") fmt.Println(" instances delete Terminate and stop billing") fmt.Println() + fmt.Println(" Jobs: A command on its own machine, stopped when it ends") + fmt.Println(" run --gpu A4000 -- Run and stream output [--volume X] [--max-runtime 2h]") + fmt.Println(" run --gpu A4000 -d -- Start and return the job ID") + fmt.Println(" jobs list [--status running] List jobs") + fmt.Println(" jobs status Details (exit code, cost, error)") + fmt.Println(" jobs logs [-f] Print output (-f to follow)") + fmt.Println(" jobs cancel Cancel and stop the machine") + fmt.Println() fmt.Println(" Access:") fmt.Println(" ssh Interactive shell (keyless)") fmt.Println(" ssh -- Run command and exit") diff --git a/internal/commands/run.go b/internal/commands/run.go new file mode 100644 index 0000000..1a2ae3d --- /dev/null +++ b/internal/commands/run.go @@ -0,0 +1,250 @@ +package commands + +import ( + "errors" + "fmt" + "math" + "os" + "os/signal" + "regexp" + "strconv" + "strings" + "time" + + "github.com/runcrate/cli/internal/api" + "github.com/spf13/cobra" +) + +const ( + defaultMaxRuntimeMinutes = 60 + maxMaxRuntimeMinutes = 24 * 60 +) + +// parseMaxRuntime reads --max-runtime: a Go duration (90m, 2h, 1h30m) or bare +// minutes (45). Rounds up to whole minutes; allows 1 minute to 24 hours. +func parseMaxRuntime(s string) (int, error) { + s = strings.TrimSpace(s) + var minutes int + if n, err := strconv.Atoi(s); err == nil { + minutes = n + } else { + d, err := time.ParseDuration(s) + if err != nil { + return 0, fmt.Errorf("invalid --max-runtime %q: use a duration like 90m, 2h or 1h30m, or a number of minutes", s) + } + if d <= 0 { + minutes = 0 + } else { + minutes = int(math.Ceil(d.Minutes())) + } + } + if minutes < 1 || minutes > maxMaxRuntimeMinutes { + return 0, fmt.Errorf("--max-runtime must be between 1m and 24h (got %s)", s) + } + return minutes, nil +} + +// formatMinutes renders whole minutes as 45m, 1h0m, 1h30m. +func formatMinutes(m int) string { + return strings.TrimSuffix((time.Duration(m) * time.Minute).String(), "0s") +} + +// shellSafe matches arguments that mean the same thing quoted or not. +var shellSafe = regexp.MustCompile(`^[A-Za-z0-9_@%+=:,./-]+$`) + +// shellQuote quotes an argument for bash when it needs it. +func shellQuote(arg string) string { + if arg == "" { + return "''" + } + if shellSafe.MatchString(arg) { + return arg + } + return "'" + strings.ReplaceAll(arg, "'", `'"'"'`) + "'" +} + +// joinCommand turns the arguments after -- into one bash command. A single +// argument is used as is, so `-- 'make && ./train'` keeps its shell syntax; +// several are quoted where needed so `-- python -c 'print(1)'` runs as typed. +func joinCommand(args []string) string { + if len(args) == 1 { + return args[0] + } + quoted := make([]string, len(args)) + for i, a := range args { + quoted[i] = shellQuote(a) + } + return strings.Join(quoted, " ") +} + +var envKeyPattern = regexp.MustCompile(`^[A-Za-z_][A-Za-z0-9_]*$`) + +// parseEnvFlags reads repeated --env KEY=VALUE. A bare KEY takes its value +// from the local environment, so secrets stay out of shell history. +func parseEnvFlags(vals []string, lookup func(string) (string, bool)) (map[string]string, error) { + if len(vals) == 0 { + return nil, nil + } + env := make(map[string]string, len(vals)) + for _, kv := range vals { + key, value, hasValue := strings.Cut(kv, "=") + if !envKeyPattern.MatchString(key) { + return nil, fmt.Errorf("invalid --env %q: the name must be letters, digits and underscores, not starting with a digit", kv) + } + if !hasValue { + v, ok := lookup(key) + if !ok { + return nil, fmt.Errorf("--env %s: %s is not set in your environment (use --env %s=value)", key, key, key) + } + value = v + } + env[key] = value + } + return env, nil +} + +// buildJobRequest turns run's flags and arguments into the create body. +func buildJobRequest(cmd *cobra.Command, args []string, lookupEnv func(string) (string, bool)) (api.CreateJobRequest, error) { + var req api.CreateJobRequest + + if len(args) == 0 { + return req, fmt.Errorf("give the command to run after --, e.g. runcrate run --gpu A4000 -- python train.py") + } + + gpuRaw, _ := cmd.Flags().GetString("gpu") + gpuCountRaw, _ := cmd.Flags().GetInt("gpu-count") + gpuType, gpuCount, err := parseGPUFlag(gpuRaw, gpuCountRaw, cmd.Flags().Changed("gpu-count")) + if err != nil { + return req, err + } + if gpuType == "" { + return req, fmt.Errorf("--gpu is required (see GPUs with: runcrate instances types)") + } + + minutes := defaultMaxRuntimeMinutes + if cmd.Flags().Changed("max-runtime") { + raw, _ := cmd.Flags().GetString("max-runtime") + if minutes, err = parseMaxRuntime(raw); err != nil { + return req, err + } + } + + envVals, _ := cmd.Flags().GetStringArray("env") + env, err := parseEnvFlags(envVals, lookupEnv) + if err != nil { + return req, err + } + + req.Command = joinCommand(args) + req.GPUType = gpuType + req.GPUCount = gpuCount + req.MaxRuntimeMinutes = minutes + req.Env = env + req.StorageID, _ = cmd.Flags().GetString("volume") + req.Template, _ = cmd.Flags().GetString("template") + req.Image, _ = cmd.Flags().GetString("image") + req.Name, _ = cmd.Flags().GetString("name") + req.Region, _ = cmd.Flags().GetString("region") + return req, nil +} + +func newRunCmd() *cobra.Command { + cmd := &cobra.Command{ + Use: "run --gpu [flags] -- ", + Short: "Run a command on a fresh GPU machine and stream its output", + Long: `Run a command on its own GPU machine. Runcrate launches the machine, runs the +command with bash in /workspace (the volume's files when --volume is set), +streams its output here, and stops the machine when the command ends. + +The exit code matches the job: the command's own code when it fails, 124 when it +hits --max-runtime, 130 when it is cancelled. Ctrl-C detaches and leaves the job +running.`, + Example: ` runcrate run --gpu A4000 -- nvidia-smi + runcrate run --gpu H100 --volume --max-runtime 2h -- python train.py --epochs 3 + runcrate run --gpu A4000 --env HF_TOKEN -- 'pip install -r requirements.txt && python eval.py'`, + RunE: runJob, + } + // Flags end at the first argument, so the command's own flags pass through + // even without --. + cmd.Flags().SetInterspersed(false) + cmd.Flags().String("gpu", "", "GPU type, e.g. A4000 or H100:2 (required)") + cmd.Flags().Int("gpu-count", 1, "Number of GPUs (alternative to colon syntax)") + cmd.Flags().String("volume", "", "Volume ID to mount as /workspace") + cmd.Flags().String("max-runtime", formatMinutes(defaultMaxRuntimeMinutes), "Stop the job after this long: 90m, 2h, or minutes (1m to 24h)") + cmd.Flags().String("template", "", "OS template (default ubuntu-devbox)") + cmd.Flags().String("image", "", "Container image to run the command in") + cmd.Flags().StringArray("env", nil, "Environment variable KEY=VALUE, or KEY to copy it from your shell (repeatable)") + cmd.Flags().String("name", "", "Job name (default job-)") + cmd.Flags().String("region", "", "Preferred region") + cmd.Flags().BoolP("detach", "d", false, "Print the job ID and return without following its output") + return cmd +} + +var runCmd = newRunCmd() + +func runJob(cmd *cobra.Command, args []string) error { + req, err := buildJobRequest(cmd, args, os.LookupEnv) + if err != nil { + return err + } + cmd.SilenceUsage = true + + client, err := loadClient() + if err != nil { + return err + } + + job, err := client.CreateJob(req) + if err != nil { + var se *api.StatusError + if errors.As(err, &se) { + if id, ok := se.Details["job_id"].(string); ok && id != "" { + return fmt.Errorf("creating job: %w\n Job %s was created but its machine could not be launched; see: runcrate jobs status %s", err, id, id) + } + } + return fmt.Errorf("creating job: %w", err) + } + + detach, _ := cmd.Flags().GetBool("detach") + if detach { + fmt.Println(job.ID) + fmt.Fprintf(os.Stderr, "Job %s is %s. Follow it with: runcrate jobs logs -f %s\n", job.Name, job.Status, job.ID) + return nil + } + + gpu := fmt.Sprintf("%dx %s", job.GPUCount, job.GPUType) + if job.CostPerHour > 0 { + gpu += fmt.Sprintf(" at $%.2f/hr", job.CostPerHour) + } + maxRuntime := formatMinutes(job.MaxRuntimeMinutes) + if !cmd.Flags().Changed("max-runtime") { + maxRuntime += " (set --max-runtime to change)" + } + fmt.Fprintf(os.Stderr, "Job %s (%s) is %s.\n", job.Name, job.ID, job.Status) + fmt.Fprintf(os.Stderr, " GPU: %s\n", gpu) + fmt.Fprintf(os.Stderr, " Max runtime: %s\n", maxRuntime) + fmt.Fprintln(os.Stderr, "Launching a machine. Output streams below and the machine stops when the command ends. Ctrl-C detaches.") + + return followAndExit(cmd, job.ID) +} + +// followAndExit streams a job until it ends or the user presses Ctrl-C, then +// exits with the job's exit code. +func followAndExit(cmd *cobra.Command, id string) error { + ctx, stop := signal.NotifyContext(cmd.Context(), os.Interrupt) + defer stop() + + code, err := streamJob(ctx, newJobFollower(), id) + if err != nil { + return err + } + if code != 0 { + stop() + os.Exit(code) + } + return nil +} + +func init() { + rootCmd.AddCommand(runCmd) +} From 74fef3bac732f733cda4ba624e9710c495ae172b Mon Sep 17 00:00:00 2001 From: Nirupam Bhowmick <48842933+heydryft@users.noreply.github.com> Date: Tue, 20 Oct 2026 13:32:15 +0100 Subject: [PATCH 2/3] fix(jobs): reuse the shared formatMinutes after the auto-stop merge --- internal/commands/run.go | 4 ---- 1 file changed, 4 deletions(-) diff --git a/internal/commands/run.go b/internal/commands/run.go index 674c023..84a8f53 100644 --- a/internal/commands/run.go +++ b/internal/commands/run.go @@ -45,10 +45,6 @@ func parseMaxRuntime(s string) (int, error) { } // formatMinutes renders whole minutes as 45m, 1h0m, 1h30m. -func formatMinutes(m int) string { - return strings.TrimSuffix((time.Duration(m) * time.Minute).String(), "0s") -} - // shellSafe matches arguments that mean the same thing quoted or not. var shellSafe = regexp.MustCompile(`^[A-Za-z0-9_@%+=:,./-]+$`) From b92653838c606254fb8f63482ed3ebf1ba3f4c90 Mon Sep 17 00:00:00 2001 From: Nirupam Bhowmick <48842933+heydryft@users.noreply.github.com> Date: Tue, 20 Oct 2026 13:33:24 +0100 Subject: [PATCH 3/3] fix(api): one budget-reached message for every command --- internal/api/client.go | 3 +++ 1 file changed, 3 insertions(+) diff --git a/internal/api/client.go b/internal/api/client.go index cbe0989..f24573e 100644 --- a/internal/api/client.go +++ b/internal/api/client.go @@ -32,6 +32,9 @@ type APIError struct { } func (e *APIError) Error() string { + if e.Status == http.StatusPaymentRequired && strings.EqualFold(e.Code, "BUDGET_REACHED") { + return "this environment's monthly budget is reached; an owner or admin can raise it in the dashboard (HTTP 402)" + } if e.Message != "" { return fmt.Sprintf("%s (HTTP %d)", e.Message, e.Status) }