From ec0bdde9994fa8387040349b237a2479bf1cab65 Mon Sep 17 00:00:00 2001 From: ryank90 Date: Thu, 3 Sep 2026 15:23:28 +0100 Subject: [PATCH] feat(api): introduce invoke and task methods --- docs/runware_serverless_apps.md | 2 + docs/runware_serverless_apps_invoke.md | 59 ++++ docs/runware_serverless_apps_tasks.md | 51 +++ docs/runware_serverless_apps_tasks_show.md | 34 ++ internal/api/serverless/client.go | 37 +- internal/api/serverless/tasks.go | 289 +++++++++++++++ internal/api/serverless/tasks_test.go | 370 ++++++++++++++++++++ internal/cmd/serverless/apps.go | 10 +- internal/cmd/serverless/apps_invoke.go | 138 ++++++++ internal/cmd/serverless/apps_invoke_test.go | 79 +++++ internal/cmd/serverless/apps_tasks.go | 111 ++++++ internal/cmd/serverless/apps_tasks_test.go | 68 ++++ internal/cmd/serverless/display.go | 91 ++++- internal/cmd/serverless/display_test.go | 6 +- 14 files changed, 1320 insertions(+), 25 deletions(-) create mode 100644 docs/runware_serverless_apps_invoke.md create mode 100644 docs/runware_serverless_apps_tasks.md create mode 100644 docs/runware_serverless_apps_tasks_show.md create mode 100644 internal/api/serverless/tasks.go create mode 100644 internal/api/serverless/tasks_test.go create mode 100644 internal/cmd/serverless/apps_invoke.go create mode 100644 internal/cmd/serverless/apps_invoke_test.go create mode 100644 internal/cmd/serverless/apps_tasks.go create mode 100644 internal/cmd/serverless/apps_tasks_test.go diff --git a/docs/runware_serverless_apps.md b/docs/runware_serverless_apps.md index 5c64946..3898fa5 100644 --- a/docs/runware_serverless_apps.md +++ b/docs/runware_serverless_apps.md @@ -32,12 +32,14 @@ runware serverless apps [flags] * [runware serverless apps delete](runware_serverless_apps_delete.md) - Delete a serverless application * [runware serverless apps endpoints](runware_serverless_apps_endpoints.md) - List endpoints for a serverless application * [runware serverless apps env](runware_serverless_apps_env.md) - Manage plain-text environment variables for an application +* [runware serverless apps invoke](runware_serverless_apps_invoke.md) - Invoke an application endpoint * [runware serverless apps list](runware_serverless_apps_list.md) - List serverless applications * [runware serverless apps logs](runware_serverless_apps_logs.md) - Show logs for a serverless application * [runware serverless apps resume](runware_serverless_apps_resume.md) - Resume a stopped serverless application * [runware serverless apps scale](runware_serverless_apps_scale.md) - Scale a serverless application * [runware serverless apps show](runware_serverless_apps_show.md) - Show details for a serverless application * [runware serverless apps stop](runware_serverless_apps_stop.md) - Stop a serverless application +* [runware serverless apps tasks](runware_serverless_apps_tasks.md) - List and inspect application tasks * [runware serverless apps usage](runware_serverless_apps_usage.md) - Show usage for a serverless application * [runware serverless apps versions](runware_serverless_apps_versions.md) - Inspect application versions * [runware serverless apps workers](runware_serverless_apps_workers.md) - List workers for a serverless application diff --git a/docs/runware_serverless_apps_invoke.md b/docs/runware_serverless_apps_invoke.md new file mode 100644 index 0000000..2469160 --- /dev/null +++ b/docs/runware_serverless_apps_invoke.md @@ -0,0 +1,59 @@ +## runware serverless apps invoke + +Invoke an application endpoint + +### Synopsis + +Submit a JSON payload to a named application endpoint. + +endpointPath is a bare lowercase segment as returned by apps endpoints +(e.g. infer). A leading slash is rejected. + +The default is async: the command prints the accepted task id. Pass --wait +to poll until the task is completed or failed. + +--sync uses the sync invocation endpoint. If the platform wait window +expires, the command polls the returned task id; it never treats expiry as +a failure and never resubmits. + +``` +runware serverless apps invoke [flags] +``` + +### Examples + +``` + # list endpoint paths, then invoke asynchronously + runware serverless apps endpoints my-app + runware serverless apps invoke my-app infer -f payload.json + + # wait for a completed task (sync, then poll if the wait window expires) + runware serverless apps invoke my-app infer --sync -f payload.json + + # async invoke and poll + runware serverless apps invoke my-app infer --wait -f payload.json +``` + +### Options + +``` + -f, --body string JSON payload file, or - for stdin (default {}) + -h, --help help for invoke + --poll-interval duration Polling interval when waiting for a task (default 2s) + --sync Use sync invocation and wait for a terminal task + --wait Poll until the task is completed or failed +``` + +### Options inherited from parent commands + +``` + --debug Show full debug output + -F, --format string CLI output format: table, json, yaml (default "table") + --transport string Transport protocol: ws (WebSocket) or http (REST) (default "ws") + -v, --verbose Show request/response details +``` + +### SEE ALSO + +* [runware serverless apps](runware_serverless_apps.md) - Manage deployed serverless applications + diff --git a/docs/runware_serverless_apps_tasks.md b/docs/runware_serverless_apps_tasks.md new file mode 100644 index 0000000..086938e --- /dev/null +++ b/docs/runware_serverless_apps_tasks.md @@ -0,0 +1,51 @@ +## runware serverless apps tasks + +List and inspect application tasks + +### Synopsis + +List TTL-bounded task metadata for an application. + +This is a recovery window, not persisted history. Pending includes queued, +running, and retrying work. A page can be empty and still have nextCursor. + +``` +runware serverless apps tasks [flags] +``` + +### Examples + +``` + # list recent tasks + runware serverless apps tasks my-app --limit 10 + + # filter by status + runware serverless apps tasks my-app --status pending + + # page through results + runware serverless apps tasks my-app --limit 10 --cursor +``` + +### Options + +``` + --cursor string Pagination cursor from a previous nextCursor + -h, --help help for tasks + --limit int Maximum number of tasks to return (1-100) + --status string Filter by status (pending, completed, or failed) +``` + +### Options inherited from parent commands + +``` + --debug Show full debug output + -F, --format string CLI output format: table, json, yaml (default "table") + --transport string Transport protocol: ws (WebSocket) or http (REST) (default "ws") + -v, --verbose Show request/response details +``` + +### SEE ALSO + +* [runware serverless apps](runware_serverless_apps.md) - Manage deployed serverless applications +* [runware serverless apps tasks show](runware_serverless_apps_tasks_show.md) - Show a single application task + diff --git a/docs/runware_serverless_apps_tasks_show.md b/docs/runware_serverless_apps_tasks_show.md new file mode 100644 index 0000000..77b685d --- /dev/null +++ b/docs/runware_serverless_apps_tasks_show.md @@ -0,0 +1,34 @@ +## runware serverless apps tasks show + +Show a single application task + +``` +runware serverless apps tasks show [flags] +``` + +### Examples + +``` + # show a task + runware serverless apps tasks show my-app 7c9e6679-7425-40de-944b-e07fc1f90ae7 +``` + +### Options + +``` + -h, --help help for show +``` + +### Options inherited from parent commands + +``` + --debug Show full debug output + -F, --format string CLI output format: table, json, yaml (default "table") + --transport string Transport protocol: ws (WebSocket) or http (REST) (default "ws") + -v, --verbose Show request/response details +``` + +### SEE ALSO + +* [runware serverless apps tasks](runware_serverless_apps_tasks.md) - List and inspect application tasks + diff --git a/internal/api/serverless/client.go b/internal/api/serverless/client.go index 23eda55..aada4cb 100644 --- a/internal/api/serverless/client.go +++ b/internal/api/serverless/client.go @@ -28,6 +28,11 @@ const defaultTimeout = 30 * time.Second // exceed the default request timeout on slow links or larger codebases. const createAppTimeout = 5 * time.Minute +// invokeSyncTimeout bounds startSyncTask. It must exceed the platform wait +// window so a 504/202 with taskId is received; a client-side timeout would +// lose the id and force a resubmit (a second billable run). +const invokeSyncTimeout = 5 * time.Minute + // GpuType is the public catalogue entry for a supported GPU type. type GpuType = gen.GpuType @@ -82,6 +87,24 @@ type BuildStatus = gen.BuildStatus // ListWorkersParams are optional filters for ListWorkers. type ListWorkersParams = gen.ListWorkersParams +// Task is a serverless invocation. +type Task = gen.Task + +// TaskStatus is a task lifecycle status. +type TaskStatus = gen.TaskStatus + +// TaskPayload is the JSON object forwarded to an endpoint handler. +type TaskPayload = gen.TaskPayload + +// ListTasksParams are optional filters for ListTasks. +type ListTasksParams = gen.ListTasksParams + +const ( + TaskStatusPending TaskStatus = gen.TaskStatusPending + TaskStatusCompleted TaskStatus = gen.TaskStatusCompleted + TaskStatusFailed TaskStatus = gen.TaskStatusFailed +) + // Endpoint is an app HTTP endpoint. type Endpoint = gen.Endpoint @@ -178,19 +201,23 @@ func newGeneratedClient(apiKey, baseURL string, httpClient gen.HttpRequestDoer) } // createInner returns a generated client suitable for createApp. -// When the underlying doer is an *http.Client with a positive Timeout shorter -// than createAppTimeout, a clone with the longer timeout is used so -// large uploads are not cut off. A zero Timeout (no deadline) is left unchanged. func (c *Client) createInner() *gen.ClientWithResponses { + return c.innerWithMinTimeout(createAppTimeout) +} + +// innerWithMinTimeout returns the generated client, cloning the HTTP client +// when its Timeout is shorter than minTimeout. A zero Timeout (no deadline) is left +// unchanged. +func (c *Client) innerWithMinTimeout(minTimeout time.Duration) *gen.ClientWithResponses { hc, ok := c.doer.(*http.Client) if !ok { return c.inner } - if hc.Timeout == 0 || hc.Timeout >= createAppTimeout { + if hc.Timeout == 0 || hc.Timeout >= minTimeout { return c.inner } cloned := *hc - cloned.Timeout = createAppTimeout + cloned.Timeout = minTimeout return newGeneratedClient(c.apiKey, c.baseURL, &cloned) } diff --git a/internal/api/serverless/tasks.go b/internal/api/serverless/tasks.go new file mode 100644 index 0000000..712d456 --- /dev/null +++ b/internal/api/serverless/tasks.go @@ -0,0 +1,289 @@ +package serverless + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "net/http" + "regexp" + "strings" + "time" + + "github.com/runware/runware-cli/internal/api/serverless/gen" + "github.com/runware/runware-cli/internal/api/transport" +) + +// endpointPathPattern is ADR-034: a bare lowercase segment, no leading slash. +var endpointPathPattern = regexp.MustCompile(`^[a-z]([a-z0-9-]{0,62}[a-z0-9])?$`) + +// taskNotFoundRetry is how long WaitTask retries getTask 404s. A freshly +// accepted id can temporarily miss the result store. +const taskNotFoundRetry = 30 * time.Second + +// defaultTaskPollInterval is used when WaitTask is called with a non-positive interval. +const defaultTaskPollInterval = 2 * time.Second + +// ValidateEndpointPath checks ADR-034. A leading slash is rejected by the API +// with 422; fail locally with a hint naming the bare segment. +func ValidateEndpointPath(path string) error { + if path == "" { + return fmt.Errorf("endpoint path is required") + } + if strings.HasPrefix(path, "/") { + bare := strings.TrimLeft(path, "/") + if endpointPathPattern.MatchString(bare) { + return fmt.Errorf("endpoint path %q must be a bare segment without a leading slash (e.g. %q)", path, bare) + } + return fmt.Errorf("endpoint path %q must be a bare lowercase segment without a leading slash", path) + } + if !endpointPathPattern.MatchString(path) { + return fmt.Errorf("endpoint path %q is invalid: use a lowercase segment of 1-64 characters (letters, digits, hyphens)", path) + } + return nil +} + +// InvokeAsync starts a task and returns the accepted (typically pending) task. +func (c *Client) InvokeAsync(ctx context.Context, appID, endpointPath string, body TaskPayload) (*Task, error) { + if c.apiKey == "" { + return nil, transport.ErrNoAPIKey + } + if err := ValidateEndpointPath(endpointPath); err != nil { + return nil, err + } + if body == nil { + body = TaskPayload{} + } + + resp, err := c.inner.StartAsyncTaskWithResponse(ctx, appID, endpointPath, gen.StartAsyncTaskJSONRequestBody(body)) + if err != nil { + return nil, fmt.Errorf("invoke async: %w", err) + } + + c.logResponse(ctx, resp.HTTPResponse, resp.Body) + + switch resp.StatusCode() { + case http.StatusAccepted: + if resp.JSON202 == nil { + return nil, fmt.Errorf("invoke async: empty 202 response") + } + return resp.JSON202, nil + case http.StatusBadRequest: + return nil, problemToError(resp.ApplicationproblemJSON400, http.StatusBadRequest) + case http.StatusUnauthorized: + return nil, problemToError(resp.ApplicationproblemJSON401, http.StatusUnauthorized) + case http.StatusForbidden: + return nil, problemToError(resp.ApplicationproblemJSON403, http.StatusForbidden) + case http.StatusNotFound: + return nil, problemToError(resp.ApplicationproblemJSON404, http.StatusNotFound) + case http.StatusConflict: + return nil, problemToError(resp.ApplicationproblemJSON409, http.StatusConflict) + case http.StatusUnprocessableEntity: + return nil, problemToError(resp.ApplicationproblemJSON422, http.StatusUnprocessableEntity) + default: + return nil, problemFromBody(resp.Body, resp.StatusCode()) + } +} + +// InvokeSync starts a task and waits up to the platform wait window. +// A wait-window expiry (504 today, 202 after RUNSERV-547) is not a failure: +// the accepted task is returned so the caller can poll. Never resubmit. +func (c *Client) InvokeSync(ctx context.Context, appID, endpointPath string, body TaskPayload) (*Task, error) { + if c.apiKey == "" { + return nil, transport.ErrNoAPIKey + } + if err := ValidateEndpointPath(endpointPath); err != nil { + return nil, err + } + if body == nil { + body = TaskPayload{} + } + + resp, err := c.innerWithMinTimeout(invokeSyncTimeout).StartSyncTaskWithResponse(ctx, appID, endpointPath, gen.StartSyncTaskJSONRequestBody(body)) + if err != nil { + return nil, fmt.Errorf("invoke sync: %w", err) + } + + c.logResponse(ctx, resp.HTTPResponse, resp.Body) + + switch resp.StatusCode() { + case http.StatusOK: + if resp.JSON200 == nil { + return nil, fmt.Errorf("invoke sync: empty 200 response") + } + return resp.JSON200, nil + case http.StatusAccepted: + // Settled shape (RUNSERV-547): wait expiry returns 202 + Task. + task, err := taskFromBody(resp.Body) + if err != nil { + if id := taskIDFromProblem(nil, resp.Body); id != "" { + return pendingTask(appID, id), nil + } + return nil, fmt.Errorf("invoke sync: wait window expired without a task id") + } + return task, nil + case http.StatusGatewayTimeout: + if id := taskIDFromProblem(resp.ApplicationproblemJSON504, resp.Body); id != "" { + return pendingTask(appID, id), nil + } + return nil, fmt.Errorf("invoke sync: wait window expired without a task id") + case http.StatusBadRequest: + return nil, problemToError(resp.ApplicationproblemJSON400, http.StatusBadRequest) + case http.StatusUnauthorized: + return nil, problemToError(resp.ApplicationproblemJSON401, http.StatusUnauthorized) + case http.StatusForbidden: + return nil, problemToError(resp.ApplicationproblemJSON403, http.StatusForbidden) + case http.StatusNotFound: + return nil, problemToError(resp.ApplicationproblemJSON404, http.StatusNotFound) + case http.StatusConflict: + return nil, problemToError(resp.ApplicationproblemJSON409, http.StatusConflict) + case http.StatusUnprocessableEntity: + return nil, problemToError(resp.ApplicationproblemJSON422, http.StatusUnprocessableEntity) + default: + return nil, problemFromBody(resp.Body, resp.StatusCode()) + } +} + +// GetTask returns a task by id. +func (c *Client) GetTask(ctx context.Context, appID, taskID string) (*Task, error) { + if c.apiKey == "" { + return nil, transport.ErrNoAPIKey + } + + resp, err := c.inner.GetTaskWithResponse(ctx, appID, taskID) + if err != nil { + return nil, fmt.Errorf("get task: %w", err) + } + + c.logResponse(ctx, resp.HTTPResponse, resp.Body) + + switch resp.StatusCode() { + case http.StatusOK: + if resp.JSON200 == nil { + return nil, fmt.Errorf("get task: empty 200 response") + } + return resp.JSON200, nil + case http.StatusUnauthorized: + return nil, problemToError(resp.ApplicationproblemJSON401, http.StatusUnauthorized) + case http.StatusForbidden: + return nil, problemToError(resp.ApplicationproblemJSON403, http.StatusForbidden) + case http.StatusNotFound: + return nil, problemToError(resp.ApplicationproblemJSON404, http.StatusNotFound) + default: + return nil, problemFromBody(resp.Body, resp.StatusCode()) + } +} + +// ListTasks returns a page of tasks for an app. +func (c *Client) ListTasks(ctx context.Context, appID string, params *ListTasksParams) (Page[Task], error) { + if c.apiKey == "" { + return Page[Task]{}, transport.ErrNoAPIKey + } + + resp, err := c.inner.ListTasksWithResponse(ctx, appID, params) + if err != nil { + return Page[Task]{}, fmt.Errorf("list tasks: %w", err) + } + + c.logResponse(ctx, resp.HTTPResponse, resp.Body) + + switch resp.StatusCode() { + case http.StatusOK: + if resp.JSON200 == nil { + return pageOf[Task](nil, nil), nil + } + return pageOf(resp.JSON200.Data, resp.JSON200.NextCursor), nil + case http.StatusBadRequest: + return Page[Task]{}, problemToError(resp.ApplicationproblemJSON400, http.StatusBadRequest) + case http.StatusUnauthorized: + return Page[Task]{}, problemToError(resp.ApplicationproblemJSON401, http.StatusUnauthorized) + case http.StatusForbidden: + return Page[Task]{}, problemToError(resp.ApplicationproblemJSON403, http.StatusForbidden) + case http.StatusNotFound: + return Page[Task]{}, problemToError(resp.ApplicationproblemJSON404, http.StatusNotFound) + case http.StatusUnprocessableEntity: + return Page[Task]{}, problemToError(resp.ApplicationproblemJSON422, http.StatusUnprocessableEntity) + default: + return Page[Task]{}, problemFromBody(resp.Body, resp.StatusCode()) + } +} + +// WaitTask polls getTask until the task is completed or failed. +// Transient 404s are retried for taskNotFoundRetry; the invocation is never +// resubmitted. +func (c *Client) WaitTask(ctx context.Context, appID, taskID string, interval time.Duration) (*Task, error) { + if interval <= 0 { + interval = defaultTaskPollInterval + } + + var notFoundSince time.Time + for { + task, err := c.GetTask(ctx, appID, taskID) + if err != nil { + if isNotFound(err) { + if notFoundSince.IsZero() { + notFoundSince = time.Now() + } + if time.Since(notFoundSince) > taskNotFoundRetry { + return nil, err + } + } else { + return nil, err + } + } else { + notFoundSince = time.Time{} + if task.Status != TaskStatusPending { + return task, nil + } + } + + timer := time.NewTimer(interval) + select { + case <-ctx.Done(): + timer.Stop() + return nil, ctx.Err() + case <-timer.C: + } + } +} + +func pendingTask(appID, taskID string) *Task { + return &Task{ + Id: taskID, + AppId: appID, + Status: TaskStatusPending, + } +} + +func taskFromBody(body []byte) (*Task, error) { + var task Task + if err := json.Unmarshal(body, &task); err != nil { + return nil, err + } + if task.Id == "" { + return nil, fmt.Errorf("empty task id") + } + return &task, nil +} + +func taskIDFromProblem(p *gen.ProblemDetails, body []byte) string { + if p != nil && p.TaskId != nil && *p.TaskId != "" { + return *p.TaskId + } + if len(body) == 0 { + return "" + } + var parsed gen.ProblemDetails + if err := json.Unmarshal(body, &parsed); err != nil { + return "" + } + if parsed.TaskId == nil { + return "" + } + return *parsed.TaskId +} + +func isNotFound(err error) bool { + var re *transport.RunwareError + return errors.As(err, &re) && re.Code == transport.CodeNotFound +} diff --git a/internal/api/serverless/tasks_test.go b/internal/api/serverless/tasks_test.go new file mode 100644 index 0000000..7ba756b --- /dev/null +++ b/internal/api/serverless/tasks_test.go @@ -0,0 +1,370 @@ +package serverless + +import ( + "context" + "encoding/json" + "errors" + "io" + "log/slog" + "net/http" + "net/http/httptest" + "strings" + "sync/atomic" + "testing" + "time" + + "github.com/runware/runware-cli/internal/api/transport" +) + +const ( + testTaskID = "task-1" + testEndpoint = "infer" + testTaskJSON = `{"id":"task-1","status":"pending","appId":"my-app","createdAt":"2026-07-30T12:00:00Z"}` + testTaskDone = `{"id":"task-1","status":"completed","appId":"my-app","createdAt":"2026-07-30T12:00:00Z","completedAt":"2026-07-30T12:00:05Z","output":{"ok":true}}` + testTaskFailed = `{"id":"task-1","status":"failed","appId":"my-app","createdAt":"2026-07-30T12:00:00Z","error":"oom killed"}` +) + +func TestValidateEndpointPath(t *testing.T) { + cases := []struct { + path string + wantErr string + }{ + {path: "infer"}, + {path: "a"}, + {path: "my-endpoint"}, + {path: "", wantErr: "required"}, + {path: "/infer", wantErr: "bare segment"}, + {path: "/infer", wantErr: `"infer"`}, + {path: "Infer", wantErr: "invalid"}, + {path: "my_endpoint", wantErr: "invalid"}, + {path: "infer/", wantErr: "invalid"}, + {path: "/", wantErr: "leading slash"}, + } + for _, tc := range cases { + err := ValidateEndpointPath(tc.path) + if tc.wantErr == "" { + if err != nil { + t.Errorf("ValidateEndpointPath(%q): %v", tc.path, err) + } + continue + } + if err == nil { + t.Errorf("ValidateEndpointPath(%q): expected error containing %q", tc.path, tc.wantErr) + continue + } + if !strings.Contains(err.Error(), tc.wantErr) { + t.Errorf("ValidateEndpointPath(%q): error %q does not contain %q", tc.path, err, tc.wantErr) + } + } +} + +func TestInvokeAsync(t *testing.T) { + var posts atomic.Int32 + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + want := "/v1/apps/" + testAppID + "/invoke-async/" + testEndpoint + if r.Method != http.MethodPost || r.URL.Path != want { + t.Errorf("unexpected %s %s", r.Method, r.URL.Path) + } + posts.Add(1) + body, err := io.ReadAll(r.Body) + if err != nil { + t.Errorf("read body: %v", err) + } + var payload map[string]any + if err := json.Unmarshal(body, &payload); err != nil { + t.Errorf("body is not JSON: %s", body) + } + if payload["prompt"] != "hi" { + t.Errorf("payload = %s", body) + } + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusAccepted) + _, _ = w.Write([]byte(testTaskJSON)) + })) + defer srv.Close() + + c := newClient("test-key", srv.URL, slog.Default(), srv.Client()) + task, err := c.InvokeAsync(context.Background(), testAppID, testEndpoint, TaskPayload{"prompt": "hi"}) + if err != nil { + t.Fatalf("InvokeAsync: %v", err) + } + if task.Id != testTaskID || task.Status != TaskStatusPending { + t.Fatalf("unexpected task: %+v", task) + } + if posts.Load() != 1 { + t.Fatalf("expected 1 POST, got %d", posts.Load()) + } +} + +func TestInvokeAsync_RejectsLeadingSlash(t *testing.T) { + c := NewClient("test-key", "https://example.invalid", slog.Default()) + _, err := c.InvokeAsync(context.Background(), testAppID, "/infer", nil) + if err == nil || !strings.Contains(err.Error(), "bare segment") { + t.Fatalf("expected leading-slash error, got %v", err) + } +} + +func TestInvokeAsync_NoAPIKey(t *testing.T) { + c := NewClient("", "https://example.invalid", slog.Default()) + if _, err := c.InvokeAsync(context.Background(), testAppID, testEndpoint, nil); !errors.Is(err, transport.ErrNoAPIKey) { + t.Fatalf("expected ErrNoAPIKey, got %v", err) + } +} + +func TestInvokeSync_Completed(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + want := "/v1/apps/" + testAppID + "/invoke-sync/" + testEndpoint + if r.Method != http.MethodPost || r.URL.Path != want { + t.Errorf("unexpected %s %s", r.Method, r.URL.Path) + } + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(testTaskDone)) + })) + defer srv.Close() + + c := newClient("test-key", srv.URL, slog.Default(), srv.Client()) + task, err := c.InvokeSync(context.Background(), testAppID, testEndpoint, nil) + if err != nil { + t.Fatalf("InvokeSync: %v", err) + } + if task.Status != TaskStatusCompleted || task.Output == nil || (*task.Output)["ok"] != true { + t.Fatalf("unexpected task: %+v", task) + } +} + +func TestInvokeSync_WaitExpiry504(t *testing.T) { + var posts atomic.Int32 + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if strings.Contains(r.URL.Path, "/invoke-sync/") { + posts.Add(1) + w.Header().Set("Content-Type", "application/problem+json") + w.WriteHeader(http.StatusGatewayTimeout) + _, _ = w.Write([]byte(`{"type":"about:blank","title":"Gateway Timeout","status":504,"taskId":"task-1"}`)) + return + } + t.Errorf("unexpected %s %s", r.Method, r.URL.Path) + })) + defer srv.Close() + + c := newClient("test-key", srv.URL, slog.Default(), srv.Client()) + task, err := c.InvokeSync(context.Background(), testAppID, testEndpoint, nil) + if err != nil { + t.Fatalf("InvokeSync must not treat 504 as failure: %v", err) + } + if task.Id != testTaskID || task.Status != TaskStatusPending { + t.Fatalf("expected pending task %s, got %+v", testTaskID, task) + } + if posts.Load() != 1 { + t.Fatalf("504 must not resubmit: got %d POSTs", posts.Load()) + } +} + +func TestInvokeSync_WaitExpiry202(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusAccepted) + _, _ = w.Write([]byte(testTaskJSON)) + })) + defer srv.Close() + + c := newClient("test-key", srv.URL, slog.Default(), srv.Client()) + task, err := c.InvokeSync(context.Background(), testAppID, testEndpoint, nil) + if err != nil { + t.Fatalf("InvokeSync must not treat 202 as failure: %v", err) + } + if task.Id != testTaskID || task.Status != TaskStatusPending { + t.Fatalf("unexpected task: %+v", task) + } +} + +func TestInvokeSync_504WithoutTaskID(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/problem+json") + w.WriteHeader(http.StatusGatewayTimeout) + _, _ = w.Write([]byte(`{"type":"about:blank","title":"Gateway Timeout","status":504}`)) + })) + defer srv.Close() + + c := newClient("test-key", srv.URL, slog.Default(), srv.Client()) + _, err := c.InvokeSync(context.Background(), testAppID, testEndpoint, nil) + if err == nil || !strings.Contains(err.Error(), "without a task id") { + t.Fatalf("expected missing-id error, got %v", err) + } + if strings.Contains(err.Error(), "timeout") || strings.Contains(err.Error(), "Timeout") { + t.Fatalf("must not report a timeout error: %v", err) + } +} + +func TestGetTask(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + want := "/v1/apps/" + testAppID + "/tasks/" + testTaskID + if r.Method != http.MethodGet || r.URL.Path != want { + t.Errorf("unexpected %s %s", r.Method, r.URL.Path) + } + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(testTaskDone)) + })) + defer srv.Close() + + c := newClient("test-key", srv.URL, slog.Default(), srv.Client()) + task, err := c.GetTask(context.Background(), testAppID, testTaskID) + if err != nil { + t.Fatalf("GetTask: %v", err) + } + if task.Status != TaskStatusCompleted { + t.Fatalf("unexpected task: %+v", task) + } +} + +func TestGetTask_NotFound(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/problem+json") + w.WriteHeader(http.StatusNotFound) + _, _ = w.Write([]byte(`{"type":"about:blank","title":"Not Found","status":404,"detail":"No task exists"}`)) + })) + defer srv.Close() + + c := newClient("test-key", srv.URL, slog.Default(), srv.Client()) + _, err := c.GetTask(context.Background(), testAppID, testTaskID) + var re *transport.RunwareError + if !errors.As(err, &re) || re.StatusCode != http.StatusNotFound { + t.Fatalf("expected 404 RunwareError, got %v", err) + } +} + +func TestListTasks_CursorAndStatus(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + want := "/v1/apps/" + testAppID + "/tasks" + if r.Method != http.MethodGet || r.URL.Path != want { + t.Errorf("unexpected %s %s", r.Method, r.URL.Path) + } + if got := r.URL.Query().Get("limit"); got != "10" { + t.Errorf("limit query = %q, want 10", got) + } + if got := r.URL.Query().Get("cursor"); got != testCursorPage2 { + t.Errorf("cursor query = %q, want %s", got, testCursorPage2) + } + if got := r.URL.Query().Get("status"); got != "pending" { + t.Errorf("status query = %q, want pending", got) + } + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(`{"data":[` + testTaskJSON + `],"nextCursor":"` + testCursorPage3 + `"}`)) + })) + defer srv.Close() + + limit := Limit(10) + cursor := Cursor(testCursorPage2) + status := TaskStatusPending + c := newClient("test-key", srv.URL, slog.Default(), srv.Client()) + page, err := c.ListTasks(context.Background(), testAppID, &ListTasksParams{ + Limit: &limit, + Cursor: &cursor, + Status: &status, + }) + if err != nil { + t.Fatalf("ListTasks: %v", err) + } + if len(page.Data) != 1 || page.Data[0].Id != testTaskID { + t.Fatalf("unexpected tasks: %+v", page.Data) + } + if page.NextCursor == nil || *page.NextCursor != testCursorPage3 { + t.Fatalf("unexpected nextCursor: %+v", page.NextCursor) + } +} + +func TestListTasks_EmptyPageKeepsCursor(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(`{"data":[],"nextCursor":"` + testCursorPage2 + `"}`)) + })) + defer srv.Close() + + c := newClient("test-key", srv.URL, slog.Default(), srv.Client()) + page, err := c.ListTasks(context.Background(), testAppID, nil) + if err != nil { + t.Fatalf("ListTasks: %v", err) + } + if len(page.Data) != 0 { + t.Fatalf("expected empty data, got %+v", page.Data) + } + if page.NextCursor == nil || *page.NextCursor != testCursorPage2 { + t.Fatalf("unexpected nextCursor: %+v", page.NextCursor) + } +} + +func TestListTasks_NoAPIKey(t *testing.T) { + c := NewClient("", "https://example.invalid", slog.Default()) + if _, err := c.ListTasks(context.Background(), testAppID, nil); !errors.Is(err, transport.ErrNoAPIKey) { + t.Fatalf("expected ErrNoAPIKey, got %v", err) + } +} + +func TestWaitTask_PollsUntilCompleted(t *testing.T) { + var gets atomic.Int32 + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path != "/v1/apps/"+testAppID+"/tasks/"+testTaskID { + t.Errorf("unexpected path %s", r.URL.Path) + } + n := gets.Add(1) + w.Header().Set("Content-Type", "application/json") + if n == 1 { + _, _ = w.Write([]byte(testTaskJSON)) + return + } + _, _ = w.Write([]byte(testTaskDone)) + })) + defer srv.Close() + + c := newClient("test-key", srv.URL, slog.Default(), srv.Client()) + task, err := c.WaitTask(context.Background(), testAppID, testTaskID, time.Millisecond) + if err != nil { + t.Fatalf("WaitTask: %v", err) + } + if task.Status != TaskStatusCompleted { + t.Fatalf("unexpected task: %+v", task) + } + if gets.Load() < 2 { + t.Fatalf("expected at least 2 GETs, got %d", gets.Load()) + } +} + +func TestWaitTask_RetriesTransient404(t *testing.T) { + var gets atomic.Int32 + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + n := gets.Add(1) + if n == 1 { + w.Header().Set("Content-Type", "application/problem+json") + w.WriteHeader(http.StatusNotFound) + _, _ = w.Write([]byte(`{"type":"about:blank","title":"Not Found","status":404}`)) + return + } + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(testTaskFailed)) + })) + defer srv.Close() + + c := newClient("test-key", srv.URL, slog.Default(), srv.Client()) + task, err := c.WaitTask(context.Background(), testAppID, testTaskID, time.Millisecond) + if err != nil { + t.Fatalf("WaitTask: %v", err) + } + if task.Status != TaskStatusFailed || task.Error == nil || *task.Error != "oom killed" { + t.Fatalf("unexpected task: %+v", task) + } +} + +func TestWaitTask_DoesNotPost(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodGet { + t.Errorf("WaitTask must not %s %s", r.Method, r.URL.Path) + } + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(testTaskDone)) + })) + defer srv.Close() + + c := newClient("test-key", srv.URL, slog.Default(), srv.Client()) + if _, err := c.WaitTask(context.Background(), testAppID, testTaskID, time.Millisecond); err != nil { + t.Fatalf("WaitTask: %v", err) + } +} diff --git a/internal/cmd/serverless/apps.go b/internal/cmd/serverless/apps.go index 36898d6..36285ba 100644 --- a/internal/cmd/serverless/apps.go +++ b/internal/cmd/serverless/apps.go @@ -23,6 +23,8 @@ func newAppsCmd(logger *log.Logger) *cobra.Command { newAppsListCmd(logger), newAppsShowCmd(logger), newAppsEndpointsCmd(logger), + newAppsInvokeCmd(logger), + newAppsTasksCmd(logger), newAppsEnvCmd(logger), newAppsVersionsCmd(logger), newAppsBuildsCmd(logger), @@ -241,7 +243,7 @@ func newAppsWorkersCmd(logger *log.Logger) *cobra.Command { } spin.Stop() - return printPage(cmdutil.FormatFor(cmd), page, workersResult(page.Data), cmd.ErrOrStderr(), extraCursorFlag("--status", status)) + return printPage(cmdutil.FormatFor(cmd), page, workersResult(page.Data), cmd.ErrOrStderr(), extraStatusCursorFlag(status)) }, } @@ -301,9 +303,9 @@ func extraListCursorFlags(query, gpuType, sort, status string) string { return strings.Join(parts, " ") } -// extraCursorFlag formats a single filter flag for a next-page --cursor hint. -func extraCursorFlag(name, value string) string { - return strings.Join(appendFlag(nil, name, value), " ") +// extraStatusCursorFlag formats --status for a next-page --cursor hint. +func extraStatusCursorFlag(value string) string { + return strings.Join(appendFlag(nil, "--status", value), " ") } func appendFlag(parts []string, name, value string) []string { diff --git a/internal/cmd/serverless/apps_invoke.go b/internal/cmd/serverless/apps_invoke.go new file mode 100644 index 0000000..d163f57 --- /dev/null +++ b/internal/cmd/serverless/apps_invoke.go @@ -0,0 +1,138 @@ +package serverless + +import ( + "encoding/json" + "fmt" + "io" + "log/slog" + "time" + + "github.com/charmbracelet/log" + serverlessapi "github.com/runware/runware-cli/internal/api/serverless" + "github.com/runware/runware-cli/internal/cmdutil" + "github.com/runware/runware-cli/internal/config" + "github.com/runware/runware-cli/internal/output" + "github.com/spf13/cobra" +) + +func newAppsInvokeCmd(logger *log.Logger) *cobra.Command { + var ( + sync bool + wait bool + bodyFile string + pollInterval time.Duration + ) + + cmd := &cobra.Command{ + Use: "invoke ", + Short: "Invoke an application endpoint", + Long: `Submit a JSON payload to a named application endpoint. + +endpointPath is a bare lowercase segment as returned by apps endpoints +(e.g. infer). A leading slash is rejected. + +The default is async: the command prints the accepted task id. Pass --wait +to poll until the task is completed or failed. + +--sync uses the sync invocation endpoint. If the platform wait window +expires, the command polls the returned task id; it never treats expiry as +a failure and never resubmits.`, + Example: ` # list endpoint paths, then invoke asynchronously + runware serverless apps endpoints my-app + runware serverless apps invoke my-app infer -f payload.json + + # wait for a completed task (sync, then poll if the wait window expires) + runware serverless apps invoke my-app infer --sync -f payload.json + + # async invoke and poll + runware serverless apps invoke my-app infer --wait -f payload.json`, + Args: cobra.ExactArgs(2), + RunE: func(cmd *cobra.Command, args []string) error { + appID := args[0] + endpointPath, payload, err := parseInvokeInput(args[1], bodyFile, cmd.InOrStdin()) + if err != nil { + return err + } + + client := serverlessapi.NewClient(config.GetAPIKey(), config.GetServerlessBaseURL(), slog.New(logger)) + spin := cmdutil.NewSpinner(fmt.Sprintf("Invoking %s on %s...", endpointPath, appID)) + spin.Start() + + var task *serverlessapi.Task + if sync { + task, err = client.InvokeSync(cmd.Context(), appID, endpointPath, payload) + } else { + task, err = client.InvokeAsync(cmd.Context(), appID, endpointPath, payload) + } + if err != nil { + spin.Stop() + return err + } + + shouldWait := sync || wait + if shouldWait && task.Status == serverlessapi.TaskStatusPending { + _, _ = fmt.Fprintf(cmd.ErrOrStderr(), "Task %s accepted; waiting...\n", task.Id) + spin.SetMessage(fmt.Sprintf("Waiting for task %s...", task.Id)) + task, err = client.WaitTask(cmd.Context(), appID, task.Id, pollInterval) + if err != nil { + spin.Stop() + return err + } + } + spin.Stop() + + if err := output.Print(cmdutil.FormatFor(cmd), taskResult(*task)); err != nil { + return err + } + return taskFailedErr(task) + }, + } + + cmd.Flags().BoolVar(&sync, "sync", false, "Use sync invocation and wait for a terminal task") + cmd.Flags().BoolVar(&wait, "wait", false, "Poll until the task is completed or failed") + cmd.Flags().StringVarP(&bodyFile, "body", "f", "", "JSON payload file, or - for stdin (default {})") + cmd.Flags().DurationVar(&pollInterval, "poll-interval", 2*time.Second, "Polling interval when waiting for a task") + return cmd +} + +func parseInvokeInput(endpointPath, bodyFile string, stdin io.Reader) (string, serverlessapi.TaskPayload, error) { + if err := serverlessapi.ValidateEndpointPath(endpointPath); err != nil { + return "", nil, err + } + payload, err := readTaskPayload(bodyFile, stdin) + if err != nil { + return "", nil, err + } + return endpointPath, payload, nil +} + +func readTaskPayload(path string, stdin io.Reader) (serverlessapi.TaskPayload, error) { + if path == "" { + return serverlessapi.TaskPayload{}, nil + } + raw, err := readValueFlag("", path, stdin) + if err != nil { + return nil, err + } + if raw == "" { + return serverlessapi.TaskPayload{}, nil + } + var payload serverlessapi.TaskPayload + if err := json.Unmarshal([]byte(raw), &payload); err != nil { + return nil, fmt.Errorf("parse body JSON: %w", err) + } + if payload == nil { + return serverlessapi.TaskPayload{}, nil + } + return payload, nil +} + +func taskFailedErr(task *serverlessapi.Task) error { + if task == nil || task.Status != serverlessapi.TaskStatusFailed { + return nil + } + if task.Error != nil && *task.Error != "" { + return fmt.Errorf("%s", *task.Error) + } + return fmt.Errorf("task failed") +} diff --git a/internal/cmd/serverless/apps_invoke_test.go b/internal/cmd/serverless/apps_invoke_test.go new file mode 100644 index 0000000..9a26d14 --- /dev/null +++ b/internal/cmd/serverless/apps_invoke_test.go @@ -0,0 +1,79 @@ +package serverless + +import ( + "os" + "path/filepath" + "strings" + "testing" + + serverlessapi "github.com/runware/runware-cli/internal/api/serverless" +) + +func TestParseInvokeInput_RejectsLeadingSlash(t *testing.T) { + _, _, err := parseInvokeInput("/infer", "", strings.NewReader("")) + if err == nil || !strings.Contains(err.Error(), "bare segment") { + t.Fatalf("expected leading-slash error, got %v", err) + } +} + +func TestParseInvokeInput_DefaultEmptyObject(t *testing.T) { + path, payload, err := parseInvokeInput("infer", "", strings.NewReader("")) + if err != nil { + t.Fatalf("parseInvokeInput: %v", err) + } + if path != "infer" { + t.Fatalf("path = %q", path) + } + if payload == nil || len(payload) != 0 { + t.Fatalf("expected empty object, got %#v", payload) + } +} + +func TestReadTaskPayload_FromFile(t *testing.T) { + path := filepath.Join(t.TempDir(), "payload.json") + if err := os.WriteFile(path, []byte("{\"prompt\":\"hi\"}\n"), 0o600); err != nil { + t.Fatal(err) + } + payload, err := readTaskPayload(path, strings.NewReader("")) + if err != nil { + t.Fatalf("readTaskPayload: %v", err) + } + if payload["prompt"] != "hi" { + t.Fatalf("payload = %#v", payload) + } +} + +func TestReadTaskPayload_FromStdin(t *testing.T) { + payload, err := readTaskPayload("-", strings.NewReader(`{"n":1}`)) + if err != nil { + t.Fatalf("readTaskPayload: %v", err) + } + if payload["n"] != float64(1) { + t.Fatalf("payload = %#v", payload) + } +} + +func TestReadTaskPayload_InvalidJSON(t *testing.T) { + _, err := readTaskPayload("-", strings.NewReader("not-json")) + if err == nil || !strings.Contains(err.Error(), "parse body JSON") { + t.Fatalf("expected JSON error, got %v", err) + } +} + +func TestTaskFailedErr(t *testing.T) { + if err := taskFailedErr(&serverlessapi.Task{Status: serverlessapi.TaskStatusCompleted}); err != nil { + t.Fatalf("completed: %v", err) + } + msg := "oom killed" + err := taskFailedErr(&serverlessapi.Task{ + Status: serverlessapi.TaskStatusFailed, + Error: &msg, + }) + if err == nil || err.Error() != msg { + t.Fatalf("failed: got %v, want %q", err, msg) + } + err = taskFailedErr(&serverlessapi.Task{Status: serverlessapi.TaskStatusFailed}) + if err == nil || err.Error() != "task failed" { + t.Fatalf("failed without string: %v", err) + } +} diff --git a/internal/cmd/serverless/apps_tasks.go b/internal/cmd/serverless/apps_tasks.go new file mode 100644 index 0000000..9e46618 --- /dev/null +++ b/internal/cmd/serverless/apps_tasks.go @@ -0,0 +1,111 @@ +package serverless + +import ( + "fmt" + "log/slog" + + "github.com/charmbracelet/log" + serverlessapi "github.com/runware/runware-cli/internal/api/serverless" + "github.com/runware/runware-cli/internal/cmdutil" + "github.com/runware/runware-cli/internal/config" + "github.com/runware/runware-cli/internal/output" + "github.com/spf13/cobra" +) + +func newAppsTasksCmd(logger *log.Logger) *cobra.Command { + var ( + limit int + cursor string + status string + ) + + cmd := &cobra.Command{ + Use: "tasks ", + Short: "List and inspect application tasks", + Long: `List TTL-bounded task metadata for an application. + +This is a recovery window, not persisted history. Pending includes queued, +running, and retrying work. A page can be empty and still have nextCursor.`, + Example: ` # list recent tasks + runware serverless apps tasks my-app --limit 10 + + # filter by status + runware serverless apps tasks my-app --status pending + + # page through results + runware serverless apps tasks my-app --limit 10 --cursor `, + Args: cobra.MaximumNArgs(1), + RunE: func(cmd *cobra.Command, args []string) error { + if len(args) == 0 { + return cmd.Help() + } + if err := validateListLimit(limit); err != nil { + return err + } + id := args[0] + statusVal, err := parseTaskStatus(status) + if err != nil { + return err + } + var params *serverlessapi.ListTasksParams + if limit > 0 || cursor != "" || status != "" { + params = &serverlessapi.ListTasksParams{} + params.Limit, params.Cursor = listPageParams(limit, cursor) + params.Status = statusVal + } + + spin := cmdutil.NewSpinner(fmt.Sprintf("Fetching tasks for %s...", id)) + spin.Start() + + client := serverlessapi.NewClient(config.GetAPIKey(), config.GetServerlessBaseURL(), slog.New(logger)) + page, err := client.ListTasks(cmd.Context(), id, params) + if err != nil { + spin.Stop() + return err + } + spin.Stop() + + return printPage(cmdutil.FormatFor(cmd), page, tasksResult(page.Data), cmd.ErrOrStderr(), extraStatusCursorFlag(status)) + }, + } + + cmd.Flags().IntVar(&limit, "limit", 0, "Maximum number of tasks to return (1-100)") + cmd.Flags().StringVar(&cursor, "cursor", "", "Pagination cursor from a previous nextCursor") + cmd.Flags().StringVar(&status, "status", "", "Filter by status (pending, completed, or failed)") + cmd.AddCommand(newAppsTasksShowCmd(logger)) + return cmd +} + +func newAppsTasksShowCmd(logger *log.Logger) *cobra.Command { + return &cobra.Command{ + Use: "show ", + Short: "Show a single application task", + Example: ` # show a task + runware serverless apps tasks show my-app 7c9e6679-7425-40de-944b-e07fc1f90ae7`, + Args: cobra.ExactArgs(2), + RunE: func(cmd *cobra.Command, args []string) error { + appID := args[0] + taskID := args[1] + + spin := cmdutil.NewSpinner(fmt.Sprintf("Fetching task %s...", taskID)) + spin.Start() + + client := serverlessapi.NewClient(config.GetAPIKey(), config.GetServerlessBaseURL(), slog.New(logger)) + task, err := client.GetTask(cmd.Context(), appID, taskID) + if err != nil { + spin.Stop() + return err + } + spin.Stop() + + if err := output.Print(cmdutil.FormatFor(cmd), taskResult(*task)); err != nil { + return err + } + return taskFailedErr(task) + }, + } +} + +func parseTaskStatus(status string) (*serverlessapi.TaskStatus, error) { + return parseValidFlag[serverlessapi.TaskStatus]("--status", status, "pending, completed, or failed") +} diff --git a/internal/cmd/serverless/apps_tasks_test.go b/internal/cmd/serverless/apps_tasks_test.go new file mode 100644 index 0000000..7250b26 --- /dev/null +++ b/internal/cmd/serverless/apps_tasks_test.go @@ -0,0 +1,68 @@ +package serverless + +import ( + "strings" + "testing" + "time" + + serverlessapi "github.com/runware/runware-cli/internal/api/serverless" +) + +const testTaskID = "task-1" + +func TestParseTaskStatus(t *testing.T) { + got, err := parseTaskStatus("") + if err != nil || got != nil { + t.Fatalf("unset status: got=%v err=%v", got, err) + } + + got, err = parseTaskStatus("pending") + if err != nil || got == nil || *got != serverlessapi.TaskStatusPending { + t.Fatalf("pending: got=%v err=%v", got, err) + } + + _, err = parseTaskStatus("running") + if err == nil || !strings.Contains(err.Error(), "invalid --status") { + t.Fatalf("expected invalid --status, got %v", err) + } +} + +func TestTaskResult_IncludesOutputAndError(t *testing.T) { + created := time.Date(2026, 7, 30, 12, 0, 0, 0, time.UTC) + completed := created.Add(5 * time.Second) + errMsg := "oom killed" + output := map[string]any{"ok": true} + r := taskResult{ + Id: testTaskID, + AppId: testAppID, + Status: serverlessapi.TaskStatusFailed, + Error: &errMsg, + Output: &output, + CreatedAt: created, + CompletedAt: &completed, + } + rows := r.Rows() + got := make(map[string]any, len(rows)) + for _, row := range rows { + got[row[0].(string)] = row[1] + } + if got[colID] != testTaskID || got[colStatus] != "failed" || got[colError] != errMsg { + t.Fatalf("rows = %#v", got) + } + if got["Output"] != `{"ok":true}` { + t.Fatalf("output = %#v", got["Output"]) + } +} + +func TestTasksResult_Headers(t *testing.T) { + r := tasksResult{ + {Id: testTaskID, Status: serverlessapi.TaskStatusPending, CreatedAt: time.Date(2026, 7, 30, 12, 0, 0, 0, time.UTC)}, + } + if got := r.Headers(); len(got) != 5 { + t.Fatalf("headers: %v", got) + } + rows := r.Rows() + if len(rows) != 1 || rows[0][0] != testTaskID || rows[0][1] != "pending" { + t.Fatalf("rows: %#v", rows) + } +} diff --git a/internal/cmd/serverless/display.go b/internal/cmd/serverless/display.go index fe9252b..9c64cdd 100644 --- a/internal/cmd/serverless/display.go +++ b/internal/cmd/serverless/display.go @@ -1,6 +1,7 @@ package serverless import ( + "encoding/json" "fmt" "io" "time" @@ -11,17 +12,19 @@ import ( ) const ( - colID = "ID" - colName = "Name" - colStatus = "Status" - colCreated = "Created" - colUpdated = "Updated" - colType = "Type" - colField = "Field" - colValue = "Value" - colApp = "App" - colKey = "Key" - colEnvVar = "Env var" + colID = "ID" + colName = "Name" + colStatus = "Status" + colCreated = "Created" + colUpdated = "Updated" + colType = "Type" + colField = "Field" + colValue = "Value" + colApp = "App" + colKey = "Key" + colEnvVar = "Env var" + colError = "Error" + colCompleted = "Completed" colComputeType = "Compute type" colGPUType = "GPU type" @@ -166,6 +169,68 @@ func (r workersResult) Rows() [][]any { return rows } +// tasksResult wraps task lists for table display. Output is omitted. +type tasksResult []serverlessapi.Task + +func (r tasksResult) Headers() []string { + return []string{colID, colStatus, colError, colCreated, colCompleted} +} + +func (r tasksResult) Rows() [][]any { + rows := make([][]any, len(r)) + for i := range r { + task := &r[i] + rows[i] = []any{ + task.Id, + string(task.Status), + formatOptionalString(task.Error), + formatTaskTime(task.CreatedAt), + formatOptionalTime(task.CompletedAt), + } + } + return rows +} + +// taskResult wraps a single task for table/json/yaml display. +type taskResult serverlessapi.Task + +func (r taskResult) Headers() []string { + return []string{colField, colValue} +} + +func (r taskResult) Rows() [][]any { + rows := [][]any{ + {colID, r.Id}, + {colApp, r.AppId}, + {colStatus, string(r.Status)}, + {colCreated, formatTaskTime(r.CreatedAt)}, + {colCompleted, formatOptionalTime(r.CompletedAt)}, + {colError, formatOptionalString(r.Error)}, + } + if r.Output != nil { + rows = append(rows, []any{"Output", formatJSONValue(*r.Output)}) + } + return rows +} + +func formatTaskTime(t time.Time) string { + if t.IsZero() { + return "" + } + return t.Format(time.RFC3339) +} + +func formatJSONValue(v any) string { + if v == nil { + return "" + } + b, err := json.Marshal(v) + if err != nil { + return fmt.Sprint(v) + } + return string(b) +} + func formatOptionalInt32(v *int32) string { if v == nil { return "" @@ -177,7 +242,7 @@ func formatOptionalInt32(v *int32) string { type buildsResult []serverlessapi.Build func (r buildsResult) Headers() []string { - return []string{colID, colStatus, "Error", colCreated} + return []string{colID, colStatus, colError, colCreated} } func (r buildsResult) Rows() [][]any { @@ -206,7 +271,7 @@ func (r buildResult) Rows() [][]any { return [][]any{ {colID, r.Id.String()}, {colStatus, string(r.Status)}, - {"Error", formatOptionalString(r.Error)}, + {colError, formatOptionalString(r.Error)}, {"Exit code", formatOptionalInt32(r.ExitCode)}, {colCreated, formatOptionalTime(r.CreatedAt)}, } diff --git a/internal/cmd/serverless/display_test.go b/internal/cmd/serverless/display_test.go index cc734d2..2b1f97c 100644 --- a/internal/cmd/serverless/display_test.go +++ b/internal/cmd/serverless/display_test.go @@ -147,11 +147,11 @@ func TestExtraListCursorFlags(t *testing.T) { } } -func TestExtraCursorFlag(t *testing.T) { - if got := extraCursorFlag("--status", "ready"); got != "--status ready" { +func TestExtraStatusCursorFlag(t *testing.T) { + if got := extraStatusCursorFlag("ready"); got != "--status ready" { t.Fatalf("got %q", got) } - if got := extraCursorFlag("--status", ""); got != "" { + if got := extraStatusCursorFlag(""); got != "" { t.Fatalf("empty: got %q", got) } }