diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index 11d0e46..609f35e 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -4,7 +4,7 @@ Thanks for your interest in contributing. This document covers how to get the pr ## Prerequisites -- **Go 1.26.8+** — the version in `go.mod`, which is where the standard +- **Go 1.26.9+** — the version in `go.mod`, which is where the standard library carries the current security fixes. The project uses [mise](https://mise.jdx.dev/) to pin the version. Run `mise install` in the repo root and Go will be available automatically. - A [nan.builders](https://nan.builders) account with an API key (needed to test the TUI at runtime). @@ -35,9 +35,11 @@ INSTALL_DIR=~/.local/bin curl -fsSL https://nan.builders/install | bash ``` main.go Entry point — delegates to cmd.Execute() -cmd/ Cobra subcommands (auth, me, metrics) +cmd/ Cobra subcommands (auth, me, metrics, run, runs) internal/ api/client.go HTTP client for the nan.builders REST API + api/runs.go /v1/runs client and its SSE event stream + runs/render.go Run events to terminal lines, guest text sanitised session/session.go Session persistence (~/.config/nan/session.json) tui/tui.go Entire TUI — layout, renderers, and tool writers ``` diff --git a/README.md b/README.md index e695b6a..06225d4 100644 --- a/README.md +++ b/README.md @@ -86,9 +86,80 @@ tool takes ours back out. The key you paste is checked against the cluster before you apply it, so a mistyped one is caught here rather than as a 401 inside each tool later. +## Agent runs + +`nan run` starts an agent (Pi or Hermes) in one of your +[workspaces](https://cloud.nan.builders) and streams what it does. The run +happens in the workspace, not on your machine: it keeps going if the command +stops, and its log stays on nan.builders. + +```bash +nan run "review PR 42 and write the findings to artifacts/review.md" +nan run --ws develop --cwd /home/nan/projects/api -f task.md +git diff | nan run - +id=$(nan run --detach "upgrade the dependencies") +``` + +| Flag | Default | | +|---|---|---| +| `--ws NAME` | your only workspace | Workspace to run in. Required when you have more than one. | +| `--agent pi\|hermes` | `pi` | Agent to run. | +| `--model M` | the agent's own | Model for the agent. | +| `--cwd PATH` | `/home/nan` | Directory in the workspace. | +| `--worktree` / `--no-worktree` | auto | Run in a separate git worktree (auto: when `--cwd` is in a git repo). | +| `--timeout D` | `30m` | Stop the run after this long, `1m` to `2h`. | +| `--detach` | | Queue the run, print its id and return. | +| `--json` | | Events as JSON lines, then the finished run object. With `--detach`, the run object. | +| `--idempotency-key K` | | Retrying with the same key and request returns the run already created. | +| `-f FILE` | | Read the prompt from a file (`-` for stdin). | + +While it streams, **Ctrl-C** detaches and leaves the run going, and prints how +to pick it up again. A second Ctrl-C within 2 seconds cancels the run. + +`--detach` exits 75 on purpose (the run is still going), so under `set -e` +allow for it: + +```bash +id=$(nan run --detach "upgrade the dependencies") || [ $? -eq 75 ] +``` + +A script that may retry should pass `--idempotency-key`: a retry after a lost +connection then gets the run it already started instead of a second one. + +Everything the agent prints is shown with escape sequences and control +characters removed: the output comes from your workspace, and it is not +allowed to drive your terminal. + +```bash +nan runs ls [--ws NAME] [--state S] [--limit 20] [--json] # newest first +nan runs show [--json] # state, result, prompt +nan runs logs [-f] [--after N] [--json] # -f follows until it ends +nan runs cancel [--json] +``` + +### Exit codes + +`nan run` and `nan runs logs -f` exit with the run's outcome, so a script can +branch on it without parsing text: + +| Code | Meaning | +|---|---| +| 0 | succeeded | +| 1 | failed | +| 2 | timed out | +| 3 | cancelled (after a double Ctrl-C: cancel requested — `nan runs show` has the final state) | +| 4 | the workspace is not set up for it: agent not installed, no inference key, bad configuration | +| 64 | usage error (also: more than one workspace and no `--ws`) | +| 65 | not signed in, session expired, or not allowed | +| 69 | nan.builders unavailable, or the stream was lost after retries (the run goes on; the id is printed) | +| 75 | `--detach`, or Ctrl-C: the run was accepted and is still going | + +The runs commands use your `nan auth login` session, or the API key from the +Setup tab when there is no session. + ## Build from source -Requires Go 1.26.8+ ([mise](https://mise.jdx.dev/) recommended) — the version +Requires Go 1.26.9+ ([mise](https://mise.jdx.dev/) recommended) — the version in `go.mod`, which is where the standard library carries the current security fixes: diff --git a/cmd/exit.go b/cmd/exit.go new file mode 100644 index 0000000..f9ba4dd --- /dev/null +++ b/cmd/exit.go @@ -0,0 +1,128 @@ +package cmd + +import ( + "errors" + "fmt" + "strings" + + "github.com/nxssie/nan-cli/internal/api" + "github.com/nxssie/nan-cli/internal/runs" + "github.com/nxssie/nan-cli/internal/session" +) + +// Exit codes of `nan run` and `nan runs`. A script that starts a run needs to +// know how it ended without parsing text, so these are a contract: the first +// five are the run's own outcome, the rest follow sysexits.h. +const ( + exitSucceeded = 0 + exitFailed = 1 + exitTimedOut = 2 + exitCancelled = 3 + exitConfig = 4 + exitUsage = 64 + exitAuth = 65 + exitUnavailable = 69 + exitDetached = 75 +) + +// ExitError ends the process with Code, printing Err first when there is one. +type ExitError struct { + Code int + Err error +} + +func (e *ExitError) Error() string { + if e.Err == nil { + return fmt.Sprintf("exit status %d", e.Code) + } + return e.Err.Error() +} + +func (e *ExitError) Unwrap() error { return e.Err } + +func exitf(code int, format string, args ...any) *ExitError { + return &ExitError{Code: code, Err: fmt.Errorf(format, args...)} +} + +func usageErrorf(format string, args ...any) *ExitError { + return exitf(exitUsage, format, args...) +} + +// authError is a refused or missing credential, with the one command that +// fixes it. +func authError(usingKey bool) *ExitError { + if usingKey { + return exitf(exitAuth, "nan.builders refused your API key — run: nan auth login") + } + return exitf(exitAuth, "%v — run: nan auth login", api.ErrSessionExpired) +} + +// apiExit maps a failed request onto an exit code and a message that says +// what to do next. The platform's own message is guest-adjacent text too, so +// it is sanitised like everything else. +func apiExit(err error, usingKey bool) *ExitError { + var exit *ExitError + if errors.As(err, &exit) { + return exit + } + if errors.Is(err, api.ErrSessionExpired) { + return authError(usingKey) + } + if errors.Is(err, session.ErrNotLoggedIn) { + return &ExitError{Code: exitAuth, Err: err} + } + var apiErr *api.APIError + if !errors.As(err, &apiErr) { + return &ExitError{Code: exitUnavailable, Err: err} + } + msg := runs.SanitizeLine(apiErr.Error()) + switch apiErr.Code { + case "tier_restricted": + return exitf(exitAuth, "your membership does not include runs — see https://cloud.nan.builders/membership") + case "workspace_key_not_allowed": + return exitf(exitAuth, "a workspace key cannot start or read runs — sign in with: nan auth login") + case "agent_not_installed", "no_inference_key", "config_error": + return exitf(exitConfig, "%s", msg) + case "idempotency_conflict": + return usageErrorf("that idempotency key was already used for a different run") + case "run_queue_full": + return exitf(exitUnavailable, "too many runs are queued — wait for some to finish (%s)", msg) + case "rate_limited": + return exitf(exitUnavailable, "too many runs started in the last hour — try again later") + } + switch { + case apiErr.Status == 400: + if apiErr.Param != "" { + return usageErrorf("%s: %s", runs.SanitizeLine(apiErr.Param), msg) + } + return usageErrorf("%s", msg) + case apiErr.Status == 401: + return exitf(exitAuth, "%v", session.ErrNotLoggedIn) + case apiErr.Status == 403: + return exitf(exitAuth, "%s", msg) + case apiErr.Status == 404: + return usageErrorf("%s", msg) + } + return exitf(exitUnavailable, "nan.builders could not do that right now: %s", msg) +} + +// runExit is the exit for a run that ended - or the detach code for one +// that has not. +func runExit(run *api.Run) error { + code, ok := runs.ExitCode(run) + if !ok { + return &ExitError{Code: exitDetached} + } + if code == exitSucceeded { + return nil + } + return &ExitError{Code: code} +} + +func joinNames(names []string) string { + clean := make([]string, len(names)) + for i, n := range names { + clean[i] = " " + runs.SanitizeLine(n) + } + return strings.Join(clean, "\n") +} diff --git a/cmd/review_test.go b/cmd/review_test.go new file mode 100644 index 0000000..59e2b84 --- /dev/null +++ b/cmd/review_test.go @@ -0,0 +1,257 @@ +package cmd + +import ( + "bytes" + "context" + "errors" + "io" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" + + "github.com/nxssie/nan-cli/internal/api" +) + +// `nan runs logs -f` meets the same 401 and 404 as the rest, only on the +// stream: it has to end the same way. +func TestRunsLogsFollowStreamRefusals(t *testing.T) { + defer resetRunsFlags() + for _, tc := range []struct { + status int + want int + says string + }{ + {http.StatusUnauthorized, exitAuth, "nan auth login"}, + {http.StatusNotFound, exitUsage, "not found"}, + } { + resetRunsFlags() + logsFollow = true + h := newHarness(t, &fakePlatform{streamStatus: tc.status}) + err := doRunsLogs(context.Background(), h.env, testRunID) + if exitCode(err) != tc.want || !strings.Contains(err.Error(), tc.says) { + t.Errorf("%d: exit %d (%v)", tc.status, exitCode(err), err) + } + if h.fake.streams != 1 { + t.Errorf("%d: retried %d times", tc.status, h.fake.streams-1) + } + } +} + +func TestRuntimeErrorsExit1NotUsage(t *testing.T) { + err := runtimeError(errors.New("write /dev/stdout: broken pipe")) + if code, _ := exitCodeFor(runCmd, err); code != 1 { + t.Errorf("a runtime failure exits %d", code) + } + if runtimeError(nil) != nil { + t.Error("nil became an error") + } + if exitCode(runtimeError(&ExitError{Code: exitAuth})) != exitAuth { + t.Error("an ExitError lost its code") + } +} + +// Neither credential may reach the terminal on any path: success, refusals, +// a lost stream, detach. +func TestCredentialsNeverReachTheOutput(t *testing.T) { + const token, key = "sess-SECRET-token", "sk-SECRET-key" + for _, fake := range []*fakePlatform{ + {}, + {createStatus: 401, createBody: `{"error":{"message":"invalid","code":"invalid_session"}}`}, + {createStatus: 403, createBody: `{"error":{"message":"no","code":"tier_restricted"}}`}, + {createStatus: 500, createBody: `{"error":{"message":"boom"}}`}, + {streamStatus: 502}, + {streamStatus: 401}, + } { + for _, creds := range [][2]string{{token, ""}, {"", key}} { + h := newHarness(t, fake) + c := api.NewRunsClient(creds[0], creds[1]).WithBaseURL(h.url) + c.Backoff = func(int) time.Duration { return time.Millisecond } + h.env.client, h.env.usingKey = c, creds[0] == "" + for _, detach := range []bool{false, true} { + opts := defaultOpts() + opts.detach = detach + err := doRun(context.Background(), h.env, opts, []string{"x"}) + var all bytes.Buffer + all.Write(h.stdout.Bytes()) + all.Write(h.stderr.Bytes()) + if err != nil { + all.WriteString(err.Error()) + } + if strings.Contains(all.String(), "SECRET") { + t.Errorf("a credential reached the output: %q", all.String()) + } + } + fake.created = nil + } + } +} + +func TestStdinPromptOnATerminalSaysItIsWaiting(t *testing.T) { + var notice bytes.Buffer + _, err := buildRunRequest(promptInput{r: strings.NewReader("p"), tty: true, notice: ¬ice}, defaultOpts(), []string{"-"}) + if err != nil { + t.Fatal(err) + } + if !strings.Contains(notice.String(), "Ctrl-D") { + t.Errorf("notice = %q", notice.String()) + } + notice.Reset() + _, _ = buildRunRequest(promptInput{r: strings.NewReader("p"), notice: ¬ice}, defaultOpts(), []string{"-"}) + if notice.Len() != 0 { + t.Error("a piped prompt printed the terminal notice") + } +} + +func TestAServerIDThatIsNotARunIDIsRefused(t *testing.T) { + h := newHarness(t, &fakePlatform{createBody: `{"id":"../../api/keys","object":"run","state":"queued"}`}) + err := doRun(context.Background(), h.env, defaultOpts(), []string{"x"}) + if exitCode(err) != exitUnavailable || h.fake.streams != 0 { + t.Errorf("exit %d (%v), streams %d", exitCode(err), err, h.fake.streams) + } +} + +func TestARepeatedStateIsPrintedOnce(t *testing.T) { + h := newHarness(t, &fakePlatform{finalState: "succeeded", events: []string{ + "event: state\ndata: {\"state\":\"running\"}\n\n", + "event: state\ndata: {\"state\":\"running\"}\n\n", + }}) + if err := doRun(context.Background(), h.env, defaultOpts(), []string{"x"}); err != nil { + t.Fatal(err) + } + if n := strings.Count(h.stderr.String(), "── running"); n != 1 { + t.Errorf("printed running %d times: %q", n, h.stderr.String()) + } +} + +// A usage mistake is reported as one even where nobody is signed in. +func TestUsageIsCheckedBeforeTheSession(t *testing.T) { + t.Setenv("HOME", t.TempDir()) + t.Setenv("USERPROFILE", t.TempDir()) + for _, args := range [][]string{ + {"run", "a", "b"}, + {"runs", "show", "not-an-id"}, + {"runs", "show"}, + {"runs", "ls", "--limit", "0"}, + } { + rootCmd.SetArgs(args) + rootCmd.SetOut(io.Discard) + rootCmd.SetErr(io.Discard) + cmd, err := rootCmd.ExecuteC() + if code, msg := exitCodeFor(cmd, err); code != exitUsage { + t.Errorf("%v: exit %d (%s)", args, code, msg) + } + resetRunsFlags() + } +} + +func TestRunsLogsSaysWhenTheRunIsStillGoing(t *testing.T) { + defer resetRunsFlags() + resetRunsFlags() + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + _, _ = io.WriteString(w, `{"data":[],"next_after":0,"done":false}`) + })) + defer srv.Close() + var stderr bytes.Buffer + env := &runsEnv{client: api.NewRunsClient("t", "").WithBaseURL(srv.URL), stdout: io.Discard, stderr: &stderr} + if err := doRunsLogs(context.Background(), env, testRunID); err != nil { + t.Fatal(err) + } + if !strings.Contains(stderr.String(), "nan runs logs "+testRunID+" -f") { + t.Errorf("stderr = %q", stderr.String()) + } +} + +// Without -f the log comes a page at a time: each page asks after the last, +// every event prints once, and a page that does not advance ends the loop. +func TestRunsLogsPaginates(t *testing.T) { + defer resetRunsFlags() + for name, pages := range map[string][]string{ + "until done": { + `{"data":[{"v":1,"seq":1,"type":"log","data":{"stream":"nan-run","text":"one"}},{"v":1,"seq":2,"type":"log","data":{"stream":"nan-run","text":"two"}}],"next_after":2,"done":false}`, + `{"data":[{"v":1,"seq":3,"type":"log","data":{"stream":"nan-run","text":"three"}}],"next_after":3,"done":true}`, + }, + "stuck cursor": { + `{"data":[{"v":1,"seq":1,"type":"log","data":{"stream":"nan-run","text":"one"}}],"next_after":1,"done":false}`, + `{"data":[{"v":1,"seq":2,"type":"log","data":{"stream":"nan-run","text":"two"}}],"next_after":1,"done":false}`, + `{"data":[{"v":1,"seq":3,"type":"log","data":{"stream":"nan-run","text":"never"}}],"next_after":5,"done":true}`, + }, + } { + resetRunsFlags() + var afters []string + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + afters = append(afters, r.URL.Query().Get("after")) + _, _ = io.WriteString(w, pages[len(afters)-1]) + })) + var stdout bytes.Buffer + env := &runsEnv{client: api.NewRunsClient("t", "").WithBaseURL(srv.URL), stdout: &stdout, stderr: io.Discard} + if err := doRunsLogs(context.Background(), env, testRunID); err != nil { + t.Fatalf("%s: %v", name, err) + } + srv.Close() + switch name { + case "until done": + if strings.Join(afters, ",") != "0,2" || stdout.String() != "nan-run │ one\nnan-run │ two\nnan-run │ three\n" { + t.Errorf("%s: afters=%v stdout=%q", name, afters, stdout.String()) + } + case "stuck cursor": + if strings.Join(afters, ",") != "0,1" || strings.Contains(stdout.String(), "never") { + t.Errorf("%s: afters=%v stdout=%q", name, afters, stdout.String()) + } + } + } +} + +func TestJSONOutputNeutralisesALoneC1Byte(t *testing.T) { + defer resetRunsFlags() + resetRunsFlags() + logsJSON = true + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + _, _ = io.WriteString(w, "{\"data\":[{\"v\":1,\"seq\":1,\"type\":\"log\",\"data\":{\"text\":\"a\x9b31m\"}}],\"next_after\":1,\"done\":true}") + })) + defer srv.Close() + var stdout bytes.Buffer + env := &runsEnv{client: api.NewRunsClient("t", "").WithBaseURL(srv.URL), stdout: &stdout, stderr: io.Discard} + if err := doRunsLogs(context.Background(), env, testRunID); err != nil { + t.Fatal(err) + } + if bytes.IndexByte(stdout.Bytes(), 0x9b) >= 0 { + t.Errorf("raw 0x9b in --json output: %q", stdout.String()) + } + + h := newHarness(t, &fakePlatform{events: []string{"id: 1\nevent: run_event\ndata: {\"v\":1,\"seq\":1,\"type\":\"log\",\"data\":{\"text\":\"a\x9b31m\"}}\n\n"}}) + opts := defaultOpts() + opts.json = true + if err := doRun(context.Background(), h.env, opts, []string{"x"}); err != nil { + t.Fatal(err) + } + if bytes.IndexByte(h.stdout.Bytes(), 0x9b) >= 0 { + t.Errorf("raw 0x9b in nan run --json: %q", h.stdout.String()) + } +} + +func TestContradictoryWorktreeFlagsSayWhatIsWrong(t *testing.T) { + t.Setenv("HOME", t.TempDir()) + rootCmd.SetArgs([]string{"run", "--worktree", "--no-worktree", "x"}) + rootCmd.SetOut(io.Discard) + rootCmd.SetErr(io.Discard) + cmd, err := rootCmd.ExecuteC() + code, msg := exitCodeFor(cmd, err) + runOpts = runOptions{agent: "pi", timeout: 30 * time.Minute} + _ = runCmd.Flags().Set("worktree", "false") + _ = runCmd.Flags().Set("no-worktree", "false") + if code != exitUsage || !strings.Contains(msg, "pick one") { + t.Errorf("exit %d: %s", code, msg) + } +} + +func TestReconnectsAreAnnounced(t *testing.T) { + h := newHarness(t, &fakePlatform{dropFirst: true}) + if err := doRun(context.Background(), h.env, defaultOpts(), []string{"x"}); err != nil { + t.Fatal(err) + } + if !strings.Contains(h.stderr.String(), "reconnecting (1/5)") { + t.Errorf("stderr = %q", h.stderr.String()) + } +} diff --git a/cmd/root.go b/cmd/root.go index 414ad64..db7a060 100644 --- a/cmd/root.go +++ b/cmd/root.go @@ -1,6 +1,7 @@ package cmd import ( + "errors" "fmt" "os" @@ -41,8 +42,32 @@ func init() { } func Execute() { - if err := rootCmd.Execute(); err != nil { - fmt.Fprintln(os.Stderr, err) - os.Exit(1) + cmd, err := rootCmd.ExecuteC() + if err == nil { + return } + code, msg := exitCodeFor(cmd, err) + if msg != "" { + fmt.Fprintln(os.Stderr, msg) + } + os.Exit(code) +} + +// exitCodeFor is how a failed command ends. `nan run` and `nan runs` have a +// documented table, and a script reads their codes: there anything that is +// not already an ExitError is cobra refusing the arguments - an unknown +// flag, a missing id, two flags that exclude each other - which is a usage +// error. Every other command keeps exiting 1. +func exitCodeFor(cmd *cobra.Command, err error) (int, string) { + var exit *ExitError + if errors.As(err, &exit) { + if exit.Err == nil { + return exit.Code, "" + } + return exit.Code, exit.Err.Error() + } + if cmd != nil && cmd.Annotations[sysexitsAnnotation] != "" { + return exitUsage, err.Error() + } + return 1, err.Error() } diff --git a/cmd/run.go b/cmd/run.go new file mode 100644 index 0000000..f886e49 --- /dev/null +++ b/cmd/run.go @@ -0,0 +1,490 @@ +package cmd + +import ( + "context" + "errors" + "fmt" + "io" + "os" + "os/signal" + "regexp" + "strings" + "time" + "unicode/utf8" + + "github.com/nxssie/nan-cli/internal/api" + "github.com/nxssie/nan-cli/internal/runs" + "github.com/nxssie/nan-cli/internal/session" + "github.com/spf13/cobra" +) + +// The limits the platform enforces, checked here too so a mistake costs a +// clear message instead of a round trip and a 400. +const ( + maxPromptBytes = 32768 + minTimeout = time.Minute + maxTimeout = 2 * time.Hour +) + +var ( + modelPattern = regexp.MustCompile(`^[A-Za-z0-9._:/-]{1,128}$`) + idempotencyPattern = regexp.MustCompile(`^[A-Za-z0-9._:-]{1,128}$`) + runIDPattern = regexp.MustCompile(`^[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}$`) +) + +// How long after a Ctrl-C that detached a second one still means "cancel". +var cancelWindow = 2 * time.Second + +type runOptions struct { + workspace string + agent string + model string + cwd string + worktree bool + noWorktree bool + timeout time.Duration + timeoutSet bool + detach bool + json bool + idempotencyKey string + file string +} + +var runOpts runOptions + +var runCmd = &cobra.Command{ + Use: `run [flags] ( "prompt" | -f FILE | - )`, + Short: "Run an agent in one of your workspaces", + Long: `Run an agent in one of your workspaces and stream what it does. + +The prompt is the argument, the contents of -f FILE, or stdin when the +argument is "-". The run happens in the workspace, not on this machine: it +goes on if this command stops, and its log stays on nan.builders. + +Ctrl-C detaches and leaves the run going. A second Ctrl-C within 2 seconds +cancels it. Scripts that may retry should pass --idempotency-key, so a +retry after a lost connection returns the run instead of starting another. + +A prompt given as an argument ends up in your shell history and in ps; for +anything sensitive use -f FILE or "-". + +Exit codes: + 0 the run succeeded + 1 the run failed + 2 the run timed out + 3 the run was cancelled (after a double Ctrl-C: cancel requested) + 4 the workspace is not set up for it (agent not installed, no inference + key, bad configuration) + 64 usage error + 65 not signed in, or not allowed + 69 nan.builders unavailable, or the stream was lost (the run goes on) + 75 detached: the run was accepted and is still going`, + Example: ` nan run "review PR 42 and write the findings to artifacts/review.md" + nan run --ws develop --cwd /home/nan/projects/api -f task.md + git diff | nan run - + id=$(nan run --detach "upgrade the dependencies")`, + Args: cobra.ArbitraryArgs, + Annotations: sysexits, + RunE: func(cmd *cobra.Command, args []string) error { + opts := runOpts + opts.timeoutSet = cmd.Flags().Changed("timeout") + // The arguments first: a usage mistake is the thing to report, even + // on a machine that is not signed in. + req, err := buildRunRequest(promptInput{r: os.Stdin, tty: isTerminal(os.Stdin), notice: os.Stderr}, opts, args) + if err != nil { + return err + } + env, err := newRunsEnv() + if err != nil { + return err + } + return runtimeError(startRun(cmd.Context(), env, opts, req)) + }, +} + +func init() { + rootCmd.AddCommand(runCmd) + f := runCmd.Flags() + f.StringVar(&runOpts.workspace, "ws", "", "Workspace to run in (default: your only workspace)") + f.StringVar(&runOpts.agent, "agent", "pi", "Agent to run: pi or hermes") + f.StringVar(&runOpts.model, "model", "", "Model for the agent (default: the agent's own)") + f.StringVar(&runOpts.cwd, "cwd", "", "Directory in the workspace to run in (default: /home/nan)") + f.BoolVar(&runOpts.worktree, "worktree", false, "Always run in a separate git worktree") + f.BoolVar(&runOpts.noWorktree, "no-worktree", false, "Run in the directory itself, even inside a git repository") + f.DurationVar(&runOpts.timeout, "timeout", 30*time.Minute, "Stop the run after this long, 1m to 2h (default 30m)") + f.BoolVar(&runOpts.detach, "detach", false, "Queue the run, print its id and return without streaming") + f.BoolVar(&runOpts.json, "json", false, "Print events as JSON lines, then the finished run (with --detach: the run)") + f.StringVar(&runOpts.idempotencyKey, "idempotency-key", "", "Retry-safe key: the same key and request return the run already created") + f.StringVarP(&runOpts.file, "file", "f", "", `Read the prompt from FILE ("-" for stdin)`) + // cobra would print the default as "30m0s"; the usage line says 30m. + f.Lookup("timeout").DefValue = "0s" +} + +// runsEnv is everything the run commands touch outside themselves, so the +// tests can hand them a mock server, buffers and a fake Ctrl-C. +type runsEnv struct { + client *api.RunsClient + usingKey bool + stdout io.Writer + stderr io.Writer + interrupts <-chan os.Signal +} + +func newRunsEnv() (*runsEnv, error) { + sess, err := session.Load() + if err != nil { + if errors.Is(err, session.ErrNotLoggedIn) { + return nil, &ExitError{Code: exitAuth, Err: err} + } + return nil, exitf(exitAuth, "could not read your session (%v) — run: nan auth login", err) + } + if sess.Token == "" && sess.APIKey == "" { + return nil, &ExitError{Code: exitAuth, Err: session.ErrNotLoggedIn} + } + return &runsEnv{ + client: api.NewRunsClient(sess.Token, sess.APIKey), + usingKey: sess.Token == "", + stdout: os.Stdout, + stderr: os.Stderr, + }, nil +} + +func startRun(ctx context.Context, env *runsEnv, opts runOptions, req *api.CreateRunRequest) error { + if req.Workspace == "" { + name, err := defaultWorkspace(ctx, env) + if err != nil { + return err + } + req.Workspace = name + } + + run, replayed, err := env.client.CreateRun(ctx, *req, opts.idempotencyKey) + if err != nil { + var apiErr *api.APIError + if errors.As(err, &apiErr) && apiErr.Status == 404 { + return usageErrorf("workspace %q not found (or runs are not enabled for your account)", runs.SanitizeLine(req.Workspace)) + } + if errors.As(err, &apiErr) && apiErr.Code == "workspace_not_running" { + return exitf(exitUnavailable, "workspace %q is not running — start it at https://cloud.nan.builders and try again", runs.SanitizeLine(req.Workspace)) + } + return apiExit(err, env.usingKey) + } + if !runIDPattern.MatchString(run.ID) { + return exitf(exitUnavailable, "nan.builders answered with something that is not a run id") + } + id := run.ID + + if replayed { + fmt.Fprintf(env.stderr, "run %s already exists for this idempotency key (%s)\n", id, runs.SanitizeLine(run.State)) + } + + if opts.detach { + if opts.json { + if err := printJSON(env.stdout, run, true); err != nil { + return err + } + } else { + fmt.Fprintln(env.stdout, id) + fmt.Fprintf(env.stderr, "run %s %s on %s\n", id, runs.SanitizeLine(run.State), runs.SanitizeLine(run.Workspace.Name)) + printFollowHints(env.stderr, id) + } + return runExit(run) + } + + if !replayed { + where := runs.SanitizeLine(run.Workspace.Name) + if where == "" { + where = runs.SanitizeLine(req.Workspace) + } + fmt.Fprintf(env.stderr, "run %s queued on %s (%s) — Ctrl-C to detach\n", id, where, runs.SanitizeLine(run.Agent)) + } + return follow(ctx, env, run.ID, 0, opts.json, true) +} + +// follow streams a run until it ends and exits with its code. With +// detachable, Ctrl-C stops following instead of killing the process, and a +// second one within cancelWindow cancels the run. +func follow(ctx context.Context, env *runsEnv, id string, after int64, asJSON, detachable bool) error { + streamCtx, stopStream := context.WithCancel(ctx) + defer stopStream() + + renderer := &runs.Renderer{Out: env.stdout} + var writeErr error + lastState := "" + onFrame := func(f api.Frame) { + switch { + case f.Event != nil && asJSON: + if err := printJSON(env.stdout, f.Event, false); err != nil && writeErr == nil { + writeErr = err + } + case f.Event != nil: + renderer.Event(*f.Event) + case f.State != nil && !asJSON: + // A reconnect can repeat the state the stream was already in. + if f.State.State == lastState { + return + } + lastState = f.State.State + renderer.Flush() + state := runs.SanitizeLine(strings.ReplaceAll(f.State.State, "_", " ")) + if f.State.ErrorCode != nil && *f.State.ErrorCode != "" { + state += " (" + runs.SanitizeLine(*f.State.ErrorCode) + ")" + } + fmt.Fprintf(env.stderr, "── %s\n", state) + } + } + + type outcome struct { + run *api.Run + err error + } + env.client.OnRetry = func(attempt, max int, err error) { + renderer.Flush() + fmt.Fprintf(env.stderr, "── stream dropped, reconnecting (%d/%d)…\n", attempt, max) + } + done := make(chan outcome, 1) + go func() { + run, err := env.client.Follow(streamCtx, id, after, onFrame) + done <- outcome{run, err} + }() + + // Ctrl-C is only taken over once the run exists and is being followed: + // before that there is nothing to detach from, and a member stuck behind + // a slow request should be able to leave the ordinary way. + var interrupts <-chan os.Signal + release := func() {} + if detachable { + interrupts = env.interrupts + if interrupts == nil { + ch := make(chan os.Signal, 2) + signal.Notify(ch, os.Interrupt) + release = func() { signal.Stop(ch) } + defer release() + interrupts = ch + } + } + + finish := func(res outcome) error { + renderer.Flush() + if res.err != nil { + return streamExit(res.err, env, id) + } + if writeErr != nil { + return writeErr + } + if asJSON { + if err := printJSON(env.stdout, res.run, false); err != nil { + return err + } + } else { + fmt.Fprintln(env.stderr, runs.Outcome(res.run)) + } + return runExit(res.run) + } + + select { + case res := <-done: + return finish(res) + + case <-interrupts: + stopStream() + res := <-done + // The run may have ended in the same instant: then there is nothing + // to detach from, and its outcome is the answer. + if res.run != nil { + return finish(res) + } + renderer.Flush() + safeID := runs.SanitizeLine(id) + fmt.Fprintf(env.stderr, "\ndetached — run %s goes on in the workspace\n", safeID) + printFollowHints(env.stderr, safeID) + fmt.Fprintf(env.stderr, "press Ctrl-C again within %s to cancel it now\n", cancelWindow) + + select { + case <-interrupts: + case <-time.After(cancelWindow): + return &ExitError{Code: exitDetached} + } + // From here a further Ctrl-C leaves the ordinary way, rather than + // sitting unheard behind the cancel request. + release() + cancelCtx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + run, changed, err := env.client.CancelRun(cancelCtx, id) + if err != nil { + return apiExit(err, env.usingKey) + } + if !changed { + fmt.Fprintf(env.stderr, "run %s had already ended: %s\n", safeID, runs.SanitizeLine(run.State)) + return runExit(run) + } + fmt.Fprintf(env.stderr, "cancel requested for run %s — it stops within seconds; check with: nan runs show %s\n", safeID, safeID) + return &ExitError{Code: exitCancelled} + } +} + +func streamExit(err error, env *runsEnv, id string) error { + safeID := runs.SanitizeLine(id) + if errors.Is(err, api.ErrStreamLost) { + return exitf(exitUnavailable, "%s — run %s goes on in the workspace; follow it with: nan runs logs %s -f", runs.SanitizeLine(err.Error()), safeID, safeID) + } + exit := notFound(err, safeID, env.usingKey) + if exit.Code == exitUnavailable { + return exitf(exitUnavailable, "%s (run %s)", runs.SanitizeLine(exit.Error()), safeID) + } + return exit +} + +func printFollowHints(w io.Writer, id string) { + fmt.Fprintf(w, " follow: nan runs logs %s -f\n", id) + fmt.Fprintf(w, " cancel: nan runs cancel %s\n", id) +} + +func printJSON(w io.Writer, v any, indent bool) error { + out, err := runs.SafeJSON(v, indent) + if err != nil { + return err + } + _, err = fmt.Fprintln(w, string(out)) + return err +} + +// buildRunRequest checks the flags and reads the prompt. Everything here is +// a usage error: nothing has been sent yet. +// promptInput is where a "-" prompt comes from. With tty set, a notice says +// the command is waiting for it, which otherwise looks exactly like a hang. +type promptInput struct { + r io.Reader + tty bool + notice io.Writer +} + +func isTerminal(f *os.File) bool { + info, err := f.Stat() + return err == nil && info.Mode()&os.ModeCharDevice != 0 +} + +func buildRunRequest(in promptInput, opts runOptions, args []string) (*api.CreateRunRequest, error) { + if opts.agent != "pi" && opts.agent != "hermes" { + return nil, usageErrorf("--agent must be pi or hermes, not %q", opts.agent) + } + if opts.worktree && opts.noWorktree { + return nil, usageErrorf("--worktree and --no-worktree contradict each other; pick one") + } + if opts.model != "" && !modelPattern.MatchString(opts.model) { + return nil, usageErrorf("--model %q is not a model name (letters, digits and . _ : / -, up to 128)", opts.model) + } + if opts.idempotencyKey != "" && !idempotencyPattern.MatchString(opts.idempotencyKey) { + return nil, usageErrorf("--idempotency-key must be 1-128 of letters, digits and . _ : -") + } + + prompt, err := readPrompt(in, opts, args) + if err != nil { + return nil, err + } + + req := &api.CreateRunRequest{ + Workspace: opts.workspace, + Agent: opts.agent, + Prompt: prompt, + Trigger: "cli", + } + if opts.cwd != "" { + cwd := opts.cwd + req.Cwd = &cwd + } + if opts.model != "" { + model := opts.model + req.Model = &model + } + if opts.worktree || opts.noWorktree { + iso := opts.worktree + req.GitIsolation = &iso + } + if opts.timeoutSet { + if opts.timeout < minTimeout || opts.timeout > maxTimeout { + return nil, usageErrorf("--timeout must be between 1m and 2h, not %s", opts.timeout) + } + secs := int(opts.timeout / time.Second) + req.TimeoutSeconds = &secs + } + return req, nil +} + +func readPrompt(in promptInput, opts runOptions, args []string) (string, error) { + const how = `pass it as an argument, with -f FILE, or "-" to read stdin` + if len(args) > 1 { + return "", usageErrorf("the prompt is one argument — quote it, or %s", how) + } + if opts.file != "" && len(args) == 1 { + return "", usageErrorf("give the prompt once: as an argument or with -f, not both") + } + + var ( + raw []byte + err error + ) + switch { + case opts.file == "-" || (len(args) == 1 && args[0] == "-"): + if in.tty && in.notice != nil { + fmt.Fprintln(in.notice, "reading the prompt from stdin — end it with Ctrl-D") + } + if in.r == nil { + return "", usageErrorf("no stdin to read the prompt from") + } + raw, err = io.ReadAll(io.LimitReader(in.r, maxPromptBytes+1)) + if err != nil { + return "", usageErrorf("could not read the prompt from stdin: %v", err) + } + case opts.file != "": + f, openErr := os.Open(opts.file) + if openErr != nil { + return "", usageErrorf("could not read the prompt: %v", openErr) + } + raw, err = io.ReadAll(io.LimitReader(f, maxPromptBytes+1)) + f.Close() + if err != nil { + return "", usageErrorf("could not read the prompt: %v", err) + } + case len(args) == 1: + raw = []byte(args[0]) + default: + return "", usageErrorf("no prompt — %s", how) + } + + switch { + case len(raw) > maxPromptBytes: + return "", usageErrorf("the prompt is over %d bytes, the most a run takes", maxPromptBytes) + case strings.TrimSpace(string(raw)) == "": + return "", usageErrorf("the prompt is empty — %s", how) + case !utf8.Valid(raw): + return "", usageErrorf("the prompt is not UTF-8 text") + case strings.ContainsRune(string(raw), 0): + return "", usageErrorf("the prompt contains a NUL byte, which a run cannot take") + } + return string(raw), nil +} + +// defaultWorkspace is the member's only workspace. With none, or more than +// one, there is nothing to guess: the member has to say. +func defaultWorkspace(ctx context.Context, env *runsEnv) (string, error) { + list, err := env.client.ListWorkspaces(ctx) + if err != nil { + exit := apiExit(err, env.usingKey) + if exit.Code == exitAuth { + return "", exit + } + return "", exitf(exitUnavailable, "could not list your workspaces (%v) — name one with --ws", exit.Err) + } + switch len(list) { + case 0: + return "", usageErrorf("you have no workspaces — create one at https://cloud.nan.builders") + case 1: + return list[0].Name, nil + } + names := make([]string, len(list)) + for i, w := range list { + names[i] = w.Name + } + return "", usageErrorf("you have %d workspaces; pick one with --ws:\n%s", len(list), joinNames(names)) +} diff --git a/cmd/run_test.go b/cmd/run_test.go new file mode 100644 index 0000000..faa5c63 --- /dev/null +++ b/cmd/run_test.go @@ -0,0 +1,662 @@ +package cmd + +import ( + "bytes" + "context" + "encoding/json" + "errors" + "fmt" + "io" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "strings" + "sync" + "testing" + "time" + + "github.com/nxssie/nan-cli/internal/api" +) + +const testRunID = "6f1c2a9b-0000-4000-8000-000000000001" + +// fakePlatform is the /v1/runs contract (AGENT-AUTOMATIONS-PHASE1 §2.1.3) as +// a test server. Each test sets the parts it cares about. +type fakePlatform struct { + mu sync.Mutex + + workspaces string // body of GET /api/workspaces + createStatus int + createBody string + finalState string + finalError string + events []string // SSE frames before `end` + streamHang bool // never send `end`: the run keeps going + dropFirst bool // close the first stream after its events, without `end` + streamStatus int + cancelStatus int + + created map[string]any + idemKey string + lastEventIDs []string + streams int + cancels int + listQuery string + eventsQuery []string +} + +func (f *fakePlatform) run(state string) string { + ec := "null" + if f.finalError != "" && api.Terminal(state) { + ec = fmt.Sprintf("%q", f.finalError) + } + return fmt.Sprintf(`{"id":%q,"object":"run","workspace":{"id":"w1","name":"develop"},"agent":"pi","model":null,"trigger":"cli","state":%q,"queue_position":null,"prompt_preview":"review","cwd":"/home/nan","git_isolation":null,"timeout_seconds":1800,"cancel_requested":false,"exit_code":null,"error_code":%s,"result":null,"usage":{"tokens_in":null,"tokens_out":null,"source":null},"last_seq":2,"created_at":"2026-10-10T12:00:00Z","started_at":null,"finished_at":null}`, testRunID, state, ec) +} + +func (f *fakePlatform) ServeHTTP(w http.ResponseWriter, r *http.Request) { + f.mu.Lock() + defer f.mu.Unlock() + switch { + case r.Method == http.MethodGet && r.URL.Path == "/api/workspaces": + _, _ = io.WriteString(w, f.workspaces) + + case r.Method == http.MethodPost && r.URL.Path == "/v1/runs": + f.idemKey = r.Header.Get("Idempotency-Key") + _ = json.NewDecoder(r.Body).Decode(&f.created) + status := f.createStatus + if status == 0 { + status = http.StatusCreated + } + w.WriteHeader(status) + body := f.createBody + if body == "" { + body = f.run(api.StateQueued) + } + _, _ = io.WriteString(w, body) + + case r.Method == http.MethodGet && r.URL.Path == "/v1/runs": + f.listQuery = r.URL.RawQuery + _, _ = io.WriteString(w, `{"object":"list","data":[`+f.run("running")+`],"next_cursor":null}`) + + case r.Method == http.MethodGet && r.URL.Path == "/v1/runs/"+testRunID: + _, _ = io.WriteString(w, f.run(f.finalState)) + + case r.Method == http.MethodPost && r.URL.Path == "/v1/runs/"+testRunID+"/cancel": + f.cancels++ + status := f.cancelStatus + if status == 0 { + status = http.StatusAccepted + } + w.WriteHeader(status) + _, _ = io.WriteString(w, f.run("running")) + + case r.Method == http.MethodGet && r.URL.Path == "/v1/runs/"+testRunID+"/events": + if r.Header.Get("Accept") != "text/event-stream" { + f.eventsQuery = append(f.eventsQuery, r.URL.RawQuery) + _, _ = io.WriteString(w, `{"data":[{"v":1,"seq":1,"ts":"t","type":"message","data":{"role":"assistant","text":"hello\u001b[2J","final":true}}],"next_after":1,"done":true}`) + return + } + f.streams++ + f.lastEventIDs = append(f.lastEventIDs, r.Header.Get("Last-Event-ID")) + if f.streamStatus != 0 { + w.WriteHeader(f.streamStatus) + return + } + w.Header().Set("Content-Type", "text/event-stream") + if f.dropFirst && f.streams == 1 { + _, _ = io.WriteString(w, f.events[0]) + return + } + start := 0 + if f.dropFirst { + start = 1 + } + for _, e := range f.events[start:] { + _, _ = io.WriteString(w, e) + } + if f.streamHang { + w.(http.Flusher).Flush() + f.mu.Unlock() + <-r.Context().Done() + f.mu.Lock() + return + } + _, _ = io.WriteString(w, "event: state\ndata: {\"state\":\""+f.finalState+"\",\"error_code\":null}\n\n") + _, _ = io.WriteString(w, "event: end\ndata: "+f.run(f.finalState)+"\n\n") + + default: + http.NotFound(w, r) + } +} + +func sse(seq int, typ, data string) string { + return fmt.Sprintf("id: %d\nevent: run_event\ndata: {\"v\":1,\"seq\":%d,\"ts\":\"2026-10-10T12:00:00Z\",\"type\":%q,\"data\":%s}\n\n", seq, seq, typ, data) +} + +type harness struct { + fake *fakePlatform + url string + env *runsEnv + stdout *bytes.Buffer + stderr *bytes.Buffer + interrupts chan os.Signal +} + +func newHarness(t *testing.T, fake *fakePlatform) *harness { + t.Helper() + if fake.workspaces == "" { + fake.workspaces = `[{"workspace_uuid":"w1","name":"develop","status":"running"}]` + } + if fake.finalState == "" { + fake.finalState = api.StateSucceeded + } + if fake.events == nil { + fake.events = []string{ + sse(1, "started", `{"agent":"pi","agent_version":"1.0.2","cwd":"/home/nan"}`), + sse(2, "message", `{"role":"assistant","text":"all good","final":true}`), + } + } + srv := httptest.NewServer(fake) + t.Cleanup(srv.Close) + client := api.NewRunsClient("session-token", "").WithBaseURL(srv.URL) + client.Backoff = func(int) time.Duration { return time.Millisecond } + client.IdleTimeout = 2 * time.Second + h := &harness{ + fake: fake, + url: srv.URL, + stdout: &bytes.Buffer{}, + stderr: &bytes.Buffer{}, + interrupts: make(chan os.Signal, 2), + } + h.env = &runsEnv{client: client, stdout: h.stdout, stderr: h.stderr, interrupts: h.interrupts} + return h +} + +func exitCode(err error) int { + if err == nil { + return 0 + } + var exit *ExitError + if errors.As(err, &exit) { + return exit.Code + } + return -1 +} + +func defaultOpts() runOptions { return runOptions{agent: "pi"} } + +// doRun is what `nan run`'s RunE does once it has a session: build the +// request from the arguments, then start and follow the run. +func doRun(ctx context.Context, env *runsEnv, opts runOptions, args []string) error { + return doRunWithStdin(ctx, env, opts, args, strings.NewReader("")) +} + +func doRunWithStdin(ctx context.Context, env *runsEnv, opts runOptions, args []string, stdin io.Reader) error { + req, err := buildRunRequest(promptInput{r: stdin}, opts, args) + if err != nil { + return err + } + return startRun(ctx, env, opts, req) +} + +func TestRunStreamsAndExitsWithTheRunsCode(t *testing.T) { + for _, tc := range []struct { + state, errorCode string + want int + }{ + {api.StateSucceeded, "", 0}, + {api.StateFailed, "agent_failed", 1}, + {api.StateTimedOut, "timed_out", 2}, + {api.StateCancelled, "cancelled", 3}, + {api.StateFailed, "config_error", 4}, + {api.StateFailed, "agent_not_installed", 4}, + {api.StateFailed, "no_inference_key", 4}, + } { + t.Run(tc.state+"/"+tc.errorCode, func(t *testing.T) { + h := newHarness(t, &fakePlatform{finalState: tc.state, finalError: tc.errorCode}) + err := doRun(context.Background(), h.env, defaultOpts(), []string{"review PR 42"}) + if got := exitCode(err); got != tc.want { + t.Fatalf("exit %d (%v), want %d", got, err, tc.want) + } + if !strings.Contains(h.stdout.String(), "all good") { + t.Errorf("stdout = %q", h.stdout.String()) + } + if !strings.Contains(h.stderr.String(), "run "+testRunID) { + t.Errorf("stderr does not name the run: %q", h.stderr.String()) + } + if h.fake.created["trigger"] != "cli" || h.fake.created["prompt"] != "review PR 42" || h.fake.created["agent"] != "pi" { + t.Errorf("created with %v", h.fake.created) + } + }) + } +} + +func TestRunReconnectsWithLastEventID(t *testing.T) { + h := newHarness(t, &fakePlatform{dropFirst: true}) + if err := doRun(context.Background(), h.env, defaultOpts(), []string{"x"}); err != nil { + t.Fatal(err) + } + if fmt.Sprint(h.fake.lastEventIDs) != "[ 1]" { + t.Errorf("Last-Event-ID per connection = %q", h.fake.lastEventIDs) + } + if strings.Count(h.stdout.String(), "started pi") != 1 || !strings.Contains(h.stdout.String(), "all good") { + t.Errorf("stdout = %q", h.stdout.String()) + } +} + +func TestRunStreamLostExits69AndNamesTheRun(t *testing.T) { + h := newHarness(t, &fakePlatform{streamStatus: http.StatusBadGateway}) + err := doRun(context.Background(), h.env, defaultOpts(), []string{"x"}) + if exitCode(err) != exitUnavailable { + t.Fatalf("exit %d (%v)", exitCode(err), err) + } + if !strings.Contains(err.Error(), "nan runs logs "+testRunID+" -f") { + t.Errorf("message does not say how to pick it up again: %v", err) + } + if h.fake.streams != 6 { + t.Errorf("connected %d times", h.fake.streams) + } +} + +func waitFor(t *testing.T, cond func() bool) { + t.Helper() + deadline := time.Now().Add(3 * time.Second) + for !cond() { + if time.Now().After(deadline) { + t.Fatal("timed out waiting") + } + time.Sleep(5 * time.Millisecond) + } +} + +func TestCtrlCDetaches(t *testing.T) { + old := cancelWindow + cancelWindow = 50 * time.Millisecond + defer func() { cancelWindow = old }() + + h := newHarness(t, &fakePlatform{streamHang: true}) + done := make(chan error, 1) + go func() { done <- doRun(context.Background(), h.env, defaultOpts(), []string{"x"}) }() + waitFor(t, func() bool { h.fake.mu.Lock(); defer h.fake.mu.Unlock(); return h.fake.streams == 1 }) + h.interrupts <- os.Interrupt + + err := <-done + if exitCode(err) != exitDetached { + t.Fatalf("exit %d (%v)", exitCode(err), err) + } + for _, want := range []string{"nan runs logs " + testRunID + " -f", "nan runs cancel " + testRunID} { + if !strings.Contains(h.stderr.String(), want) { + t.Errorf("stderr lacks %q: %q", want, h.stderr.String()) + } + } + if h.fake.cancels != 0 { + t.Error("a single Ctrl-C cancelled the run") + } +} + +func TestDoubleCtrlCCancels(t *testing.T) { + old := cancelWindow + cancelWindow = 5 * time.Second + defer func() { cancelWindow = old }() + + h := newHarness(t, &fakePlatform{streamHang: true}) + done := make(chan error, 1) + go func() { done <- doRun(context.Background(), h.env, defaultOpts(), []string{"x"}) }() + waitFor(t, func() bool { h.fake.mu.Lock(); defer h.fake.mu.Unlock(); return h.fake.streams == 1 }) + h.interrupts <- os.Interrupt + h.interrupts <- os.Interrupt + + err := <-done + if exitCode(err) != exitCancelled { + t.Fatalf("exit %d (%v)", exitCode(err), err) + } + if h.fake.cancels != 1 { + t.Errorf("cancel sent %d times", h.fake.cancels) + } + if !strings.Contains(h.stderr.String(), "cancel requested") { + t.Errorf("stderr = %q", h.stderr.String()) + } +} + +func TestDefaultWorkspace(t *testing.T) { + t.Run("exactly one is used", func(t *testing.T) { + h := newHarness(t, &fakePlatform{workspaces: `[{"workspace_uuid":"w9","name":"solo","status":"running"}]`}) + if err := doRun(context.Background(), h.env, defaultOpts(), []string{"x"}); err != nil { + t.Fatal(err) + } + if h.fake.created["workspace"] != "solo" { + t.Errorf("created in %v", h.fake.created["workspace"]) + } + }) + t.Run("more than one is ambiguous", func(t *testing.T) { + h := newHarness(t, &fakePlatform{workspaces: `[{"workspace_uuid":"a","name":"develop"},{"workspace_uuid":"b","name":"prod\u001b[31m"}]`}) + err := doRun(context.Background(), h.env, defaultOpts(), []string{"x"}) + if exitCode(err) != exitUsage { + t.Fatalf("exit %d (%v)", exitCode(err), err) + } + if !strings.Contains(err.Error(), "develop") || !strings.Contains(err.Error(), "prod") || strings.Contains(err.Error(), "\x1b") { + t.Errorf("message = %q", err.Error()) + } + if h.fake.created != nil { + t.Error("created a run anyway") + } + }) + t.Run("none", func(t *testing.T) { + h := newHarness(t, &fakePlatform{workspaces: `[]`}) + if err := doRun(context.Background(), h.env, defaultOpts(), []string{"x"}); exitCode(err) != exitUsage { + t.Fatalf("exit %d (%v)", exitCode(err), err) + } + }) + t.Run("--ws skips the lookup", func(t *testing.T) { + h := newHarness(t, &fakePlatform{workspaces: `not json`}) + opts := defaultOpts() + opts.workspace = "named" + if err := doRun(context.Background(), h.env, opts, []string{"x"}); err != nil { + t.Fatal(err) + } + if h.fake.created["workspace"] != "named" { + t.Errorf("created in %v", h.fake.created["workspace"]) + } + }) +} + +func TestExpiredSessionExits65(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusUnauthorized) + _, _ = io.WriteString(w, `{"error":{"message":"invalid session","type":"authentication_error","code":"invalid_session"}}`) + })) + defer srv.Close() + env := &runsEnv{client: api.NewRunsClient("stale", "").WithBaseURL(srv.URL), stdout: io.Discard, stderr: io.Discard} + + for name, call := range map[string]func() error{ + "run": func() error { return doRun(context.Background(), env, defaultOpts(), []string{"x"}) }, + "ls": func() error { return doRunsLs(context.Background(), env) }, + "show": func() error { return doRunsShow(context.Background(), env, testRunID) }, + "cancel": func() error { return doRunsCancel(context.Background(), env, testRunID) }, + } { + err := call() + if exitCode(err) != exitAuth || !errors.Is(err, api.ErrSessionExpired) && !strings.Contains(err.Error(), "session has expired") { + t.Errorf("%s: exit %d (%v)", name, exitCode(err), err) + } + if !strings.Contains(err.Error(), "nan auth login") { + t.Errorf("%s: does not say how to fix it: %v", name, err) + } + } +} + +func TestNoSessionExits65(t *testing.T) { + t.Setenv("HOME", t.TempDir()) + t.Setenv("USERPROFILE", os.Getenv("HOME")) + _, err := newRunsEnv() + if exitCode(err) != exitAuth { + t.Fatalf("exit %d (%v)", exitCode(err), err) + } +} + +func TestAPIKeyIsUsedWithoutASession(t *testing.T) { + home := t.TempDir() + t.Setenv("HOME", home) + t.Setenv("USERPROFILE", home) + if err := os.MkdirAll(filepath.Join(home, ".config", "nan"), 0o700); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(filepath.Join(home, ".config", "nan", "session.json"), []byte(`{"token":"","apiKey":"sk-member"}`), 0o600); err != nil { + t.Fatal(err) + } + env, err := newRunsEnv() + if err != nil { + t.Fatal(err) + } + if !env.usingKey { + t.Error("a session with only an API key does not use it") + } +} + +func TestControlCharsAreStrippedFromTheStream(t *testing.T) { + h := newHarness(t, &fakePlatform{events: []string{ + sse(1, "message", `{"role":"assistant","text":"hi\u001b]0;pwned\u0007\u001b[2J\r","final":true}`), + sse(2, "tool_call", `{"id":"1","name":"bash\u001b[31m","args_preview":"rm\u009b"}`), + }}) + if err := doRun(context.Background(), h.env, defaultOpts(), []string{"x"}); err != nil { + t.Fatal(err) + } + all := h.stdout.String() + h.stderr.String() + if strings.ContainsAny(all, "\x1b\x07\r\u009b") { + t.Errorf("control bytes reached the terminal: %q", all) + } + if !strings.Contains(h.stdout.String(), "hi") || !strings.Contains(h.stdout.String(), "→ bash rm") { + t.Errorf("stdout = %q", h.stdout.String()) + } +} + +func TestRunJSONStreamsEventLinesThenTheRun(t *testing.T) { + h := newHarness(t, &fakePlatform{}) + opts := defaultOpts() + opts.json = true + if err := doRun(context.Background(), h.env, opts, []string{"x"}); err != nil { + t.Fatal(err) + } + lines := strings.Split(strings.TrimSpace(h.stdout.String()), "\n") + if len(lines) != 3 { + t.Fatalf("stdout lines = %q", lines) + } + var ev api.Event + if err := json.Unmarshal([]byte(lines[0]), &ev); err != nil || ev.Seq != 1 || ev.Type != "started" { + t.Errorf("first line %q: %v", lines[0], err) + } + var run api.Run + if err := json.Unmarshal([]byte(lines[2]), &run); err != nil || run.Object != "run" || run.State != api.StateSucceeded { + t.Errorf("last line %q: %v", lines[2], err) + } +} + +func TestDetach(t *testing.T) { + h := newHarness(t, &fakePlatform{}) + opts := defaultOpts() + opts.detach = true + err := doRun(context.Background(), h.env, opts, []string{"x"}) + if exitCode(err) != exitDetached { + t.Fatalf("exit %d (%v)", exitCode(err), err) + } + if h.stdout.String() != testRunID+"\n" { + t.Errorf("stdout = %q, want only the id", h.stdout.String()) + } + if h.fake.streams != 0 { + t.Error("--detach streamed") + } + + h = newHarness(t, &fakePlatform{}) + opts.json = true + _ = doRun(context.Background(), h.env, opts, []string{"x"}) + var run api.Run + if err := json.Unmarshal(h.stdout.Bytes(), &run); err != nil || run.ID != testRunID { + t.Errorf("--detach --json printed %q", h.stdout.String()) + } +} + +func TestIdempotencyKeyAndReplay(t *testing.T) { + h := newHarness(t, &fakePlatform{createStatus: http.StatusOK}) + opts := defaultOpts() + opts.idempotencyKey = "deploy-42" + if err := doRun(context.Background(), h.env, opts, []string{"x"}); err != nil { + t.Fatal(err) + } + if h.fake.idemKey != "deploy-42" { + t.Errorf("Idempotency-Key = %q", h.fake.idemKey) + } + if !strings.Contains(h.stderr.String(), "already exists") { + t.Errorf("a replay was not called one: %q", h.stderr.String()) + } + + opts.idempotencyKey = "has space" + if err := doRun(context.Background(), h.env, opts, []string{"x"}); exitCode(err) != exitUsage { + t.Errorf("a bad key: exit %d", exitCode(err)) + } +} + +func TestRunFlagsShapeTheRequest(t *testing.T) { + h := newHarness(t, &fakePlatform{}) + opts := defaultOpts() + opts.agent = "hermes" + opts.model = "glm5.3" + opts.cwd = "/home/nan/projects/api" + opts.noWorktree = true + opts.timeout, opts.timeoutSet = 10*time.Minute, true + if err := doRun(context.Background(), h.env, opts, []string{"x"}); err != nil { + t.Fatal(err) + } + c := h.fake.created + if c["agent"] != "hermes" || c["model"] != "glm5.3" || c["cwd"] != "/home/nan/projects/api" || + c["git_isolation"] != false || c["timeout_seconds"] != float64(600) { + t.Errorf("created with %v", c) + } + + h = newHarness(t, &fakePlatform{}) + if err := doRun(context.Background(), h.env, defaultOpts(), []string{"x"}); err != nil { + t.Fatal(err) + } + for _, k := range []string{"timeout_seconds", "git_isolation", "model", "cwd"} { + if _, ok := h.fake.created[k]; ok { + t.Errorf("sent %s without the flag, overriding the platform default", k) + } + } +} + +func TestRunUsageErrorsExit64BeforeSendingAnything(t *testing.T) { + long := strings.Repeat("a", maxPromptBytes+1) + file := filepath.Join(t.TempDir(), "prompt.md") + _ = os.WriteFile(file, []byte("from a file"), 0o600) + for name, tc := range map[string]struct { + mutate func(*runOptions) + args []string + }{ + "no prompt": {func(*runOptions) {}, nil}, + "two arguments": {func(*runOptions) {}, []string{"a", "b"}}, + "argument and -f": {func(o *runOptions) { o.file = file }, []string{"a"}}, + "empty prompt": {func(*runOptions) {}, []string{" "}}, + "too long": {func(*runOptions) {}, []string{long}}, + "NUL": {func(*runOptions) {}, []string{"a\x00b"}}, + "bad agent": {func(o *runOptions) { o.agent = "codex" }, []string{"x"}}, + "bad model": {func(o *runOptions) { o.model = "a b" }, []string{"x"}}, + "timeout too short": {func(o *runOptions) { o.timeout, o.timeoutSet = 30*time.Second, true }, []string{"x"}}, + "timeout too long": {func(o *runOptions) { o.timeout, o.timeoutSet = 3*time.Hour, true }, []string{"x"}}, + "both worktree flags": {func(o *runOptions) { o.worktree, o.noWorktree = true, true }, []string{"x"}}, + "missing prompt file": {func(o *runOptions) { o.file = file + ".nope" }, nil}, + } { + h := newHarness(t, &fakePlatform{}) + opts := defaultOpts() + tc.mutate(&opts) + err := doRun(context.Background(), h.env, opts, tc.args) + if exitCode(err) != exitUsage { + t.Errorf("%s: exit %d (%v)", name, exitCode(err), err) + } + if h.fake.created != nil { + t.Errorf("%s: sent a run anyway", name) + } + } +} + +func TestPromptFromFileAndStdin(t *testing.T) { + file := filepath.Join(t.TempDir(), "prompt.md") + _ = os.WriteFile(file, []byte("# Task\nfrom a file\n"), 0o600) + + h := newHarness(t, &fakePlatform{}) + opts := defaultOpts() + opts.file = file + if err := doRun(context.Background(), h.env, opts, nil); err != nil { + t.Fatal(err) + } + if h.fake.created["prompt"] != "# Task\nfrom a file\n" { + t.Errorf("prompt = %q", h.fake.created["prompt"]) + } + + h = newHarness(t, &fakePlatform{}) + if err := doRunWithStdin(context.Background(), h.env, defaultOpts(), []string{"-"}, strings.NewReader("from stdin")); err != nil { + t.Fatal(err) + } + if h.fake.created["prompt"] != "from stdin" { + t.Errorf("prompt = %q", h.fake.created["prompt"]) + } +} + +func TestAdmissionRefusalsMapToExitCodes(t *testing.T) { + for _, tc := range []struct { + status int + code string + param string + want int + mentioned string + }{ + {http.StatusConflict, "agent_not_installed", "", exitConfig, "not installed"}, + {http.StatusConflict, "no_inference_key", "", exitConfig, ""}, + {http.StatusConflict, "workspace_not_running", "", exitUnavailable, "not running"}, + {http.StatusConflict, "idempotency_conflict", "", exitUsage, "idempotency key"}, + {http.StatusForbidden, "tier_restricted", "", exitAuth, "membership"}, + {http.StatusForbidden, "workspace_key_not_allowed", "", exitAuth, "workspace key"}, + {http.StatusBadRequest, "invalid_request", "cwd", exitUsage, "cwd"}, + {http.StatusNotFound, "not_found", "", exitUsage, "not found"}, + {http.StatusTooManyRequests, "run_queue_full", "", exitUnavailable, "queued"}, + {http.StatusTooManyRequests, "rate_limited", "", exitUnavailable, "last hour"}, + {http.StatusServiceUnavailable, "", "", exitUnavailable, ""}, + } { + param := "null" + if tc.param != "" { + param = fmt.Sprintf("%q", tc.param) + } + h := newHarness(t, &fakePlatform{ + createStatus: tc.status, + createBody: fmt.Sprintf(`{"error":{"message":"the agent is not installed","type":"invalid_request_error","param":%s,"code":%q}}`, param, tc.code), + }) + err := doRun(context.Background(), h.env, defaultOpts(), []string{"x"}) + if exitCode(err) != tc.want { + t.Errorf("%d %s: exit %d (%v), want %d", tc.status, tc.code, exitCode(err), err, tc.want) + continue + } + if tc.mentioned != "" && !strings.Contains(err.Error(), tc.mentioned) { + t.Errorf("%d %s: message %q lacks %q", tc.status, tc.code, err.Error(), tc.mentioned) + } + if h.fake.streams != 0 { + t.Errorf("%d %s: streamed a run that was refused", tc.status, tc.code) + } + } +} + +// What cobra refuses before RunE - an unknown flag, a missing id, flags that +// exclude each other - is a usage error on these commands, exit 64. Elsewhere +// it stays 1, which is what every other command has always exited with. +func TestCobraRefusalsExit64OnRunCommands(t *testing.T) { + for _, args := range [][]string{ + {"run", "--worktree", "--no-worktree", "x"}, + {"run", "--no-such-flag", "x"}, + {"runs", "show"}, + {"runs", "logs", "a", "b"}, + } { + rootCmd.SetArgs(args) + rootCmd.SetOut(io.Discard) + rootCmd.SetErr(io.Discard) + cmd, err := rootCmd.ExecuteC() + code, _ := exitCodeFor(cmd, err) + if err == nil || code != exitUsage { + t.Errorf("%v: exit %d (%v)", args, code, err) + } + runOpts = runOptions{agent: "pi", timeout: 30 * time.Minute} + _ = runCmd.Flags().Set("worktree", "false") + _ = runCmd.Flags().Set("no-worktree", "false") + } + if code, _ := exitCodeFor(meCmd, errors.New("boom")); code != 1 { + t.Errorf("an ordinary command's failure exits %d", code) + } + if code, msg := exitCodeFor(runCmd, &ExitError{Code: exitDetached}); code != exitDetached || msg != "" { + t.Errorf("a silent ExitError: %d %q", code, msg) + } +} + +func TestRunsShowRejectsWhatIsNotAnID(t *testing.T) { + h := newHarness(t, &fakePlatform{}) + for _, id := range []string{"../../api/keys", "6f1c", "6f1c2a9b-0000-4000-8000-00000000000g"} { + if err := doRunsShow(context.Background(), h.env, id); exitCode(err) != exitUsage { + t.Errorf("%q: exit %d (%v)", id, exitCode(err), err) + } + } +} diff --git a/cmd/runs.go b/cmd/runs.go new file mode 100644 index 0000000..043f2ae --- /dev/null +++ b/cmd/runs.go @@ -0,0 +1,384 @@ +package cmd + +import ( + "context" + "errors" + "fmt" + "strings" + "text/tabwriter" + "time" + + "github.com/nxssie/nan-cli/internal/api" + "github.com/nxssie/nan-cli/internal/runs" + "github.com/spf13/cobra" +) + +// sysexitsAnnotation marks the commands whose exit codes are the documented +// table in `nan run --help`. On them a flag mistake exits 64, not 1. +const sysexitsAnnotation = "nan/sysexits" + +var sysexits = map[string]string{sysexitsAnnotation: "true"} + +var ( + lsWorkspace string + lsState string + lsLimit int + lsJSON bool + showJSON bool + logsFollow bool + logsAfter int64 + logsJSON bool + cancelJSON bool +) + +var runsCmd = &cobra.Command{ + Use: "runs", + Short: "List, inspect, follow and cancel agent runs", + Annotations: sysexits, + Args: cobra.ArbitraryArgs, + RunE: func(cmd *cobra.Command, args []string) error { + if len(args) > 0 { + return usageErrorf("unknown command %q for \"nan runs\" — try: nan runs --help", args[0]) + } + return cmd.Help() + }, +} + +var runsLsCmd = &cobra.Command{ + Use: "ls", + Short: "List your runs, newest first", + Args: cobra.NoArgs, + Annotations: sysexits, + RunE: func(cmd *cobra.Command, args []string) error { + if err := checkLsFlags(); err != nil { + return err + } + return withRunsEnv(func(env *runsEnv) error { return doRunsLs(cmd.Context(), env) }) + }, +} + +var runsShowCmd = &cobra.Command{ + Use: "show ", + Short: "Show one run: state, result, prompt", + Args: oneRunID, + Annotations: sysexits, + RunE: func(cmd *cobra.Command, args []string) error { + return withRunsEnv(func(env *runsEnv) error { return doRunsShow(cmd.Context(), env, args[0]) }) + }, +} + +var runsLogsCmd = &cobra.Command{ + Use: "logs ", + Short: "Print a run's events (-f to follow until it ends)", + Long: `Print a run's events. + +With -f, follow the run until it ends and exit with its code, the same codes +as nan run: 0 succeeded, 1 failed, 2 timed out, 3 cancelled, 4 workspace +not set up for it. Ctrl-C stops following; the run goes on.`, + Args: oneRunID, + Annotations: sysexits, + RunE: func(cmd *cobra.Command, args []string) error { + return withRunsEnv(func(env *runsEnv) error { return doRunsLogs(cmd.Context(), env, args[0]) }) + }, +} + +var runsCancelCmd = &cobra.Command{ + Use: "cancel ", + Short: "Cancel a queued or running run", + Args: oneRunID, + Annotations: sysexits, + RunE: func(cmd *cobra.Command, args []string) error { + return withRunsEnv(func(env *runsEnv) error { return doRunsCancel(cmd.Context(), env, args[0]) }) + }, +} + +func init() { + rootCmd.AddCommand(runsCmd) + runsCmd.AddCommand(runsLsCmd, runsShowCmd, runsLogsCmd, runsCancelCmd) + + runsLsCmd.Flags().StringVar(&lsWorkspace, "ws", "", "Only runs in this workspace") + runsLsCmd.Flags().StringVar(&lsState, "state", "", "Only runs in this state: queued, starting, running, succeeded, failed, cancelled, timed_out") + runsLsCmd.Flags().IntVar(&lsLimit, "limit", 20, "How many runs to list (1-100)") + runsLsCmd.Flags().BoolVar(&lsJSON, "json", false, "Print the API's list object as JSON") + + runsShowCmd.Flags().BoolVar(&showJSON, "json", false, "Print the run object as JSON") + + runsLogsCmd.Flags().BoolVarP(&logsFollow, "follow", "f", false, "Follow the run until it ends, and exit with its code") + runsLogsCmd.Flags().Int64Var(&logsAfter, "after", 0, "Start after this event number") + runsLogsCmd.Flags().BoolVar(&logsJSON, "json", false, "Print events as JSON lines (with -f, then the finished run)") + + runsCancelCmd.Flags().BoolVar(&cancelJSON, "json", false, "Print the run object as JSON") +} + +func withRunsEnv(fn func(*runsEnv) error) error { + env, err := newRunsEnv() + if err != nil { + return err + } + return runtimeError(fn(env)) +} + +// runtimeError keeps a failure that happened while running - stdout closed +// under us, say - from reading as a usage error, which is what exitCodeFor +// makes of any plain error from these commands. +func runtimeError(err error) error { + if err == nil { + return nil + } + var exit *ExitError + if errors.As(err, &exit) { + return err + } + return &ExitError{Code: 1, Err: err} +} + +func checkRunID(id string) error { + if !runIDPattern.MatchString(id) { + return usageErrorf("%q is not a run id — ids look like 6f1c2a9b-…, see: nan runs ls", runs.Truncate(runs.SanitizeLine(id), 64)) + } + return nil +} + +// oneRunID is cobra.ExactArgs(1) with a message that says what is missing. +func oneRunID(cmd *cobra.Command, args []string) error { + switch { + case len(args) == 0: + return usageErrorf("missing run id — see: nan runs ls") + case len(args) > 1: + return usageErrorf("one run id at a time") + } + return checkRunID(args[0]) +} + +func notFound(err error, id string, usingKey bool) *ExitError { + var apiErr *api.APIError + if errors.As(err, &apiErr) && apiErr.Status == 404 { + return usageErrorf("run %s not found", id) + } + return apiExit(err, usingKey) +} + +var validStates = map[string]bool{ + api.StateQueued: true, api.StateStarting: true, api.StateRunning: true, + api.StateSucceeded: true, api.StateFailed: true, api.StateCancelled: true, api.StateTimedOut: true, +} + +func checkLsFlags() error { + if lsState != "" && !validStates[lsState] { + return usageErrorf("--state must be one of queued, starting, running, succeeded, failed, cancelled, timed_out") + } + if lsLimit < 1 || lsLimit > 100 { + return usageErrorf("--limit must be between 1 and 100") + } + return nil +} + +func doRunsLs(ctx context.Context, env *runsEnv) error { + list, err := env.client.ListRuns(ctx, api.ListRunsParams{Workspace: lsWorkspace, State: lsState, Limit: lsLimit}) + if err != nil { + return apiExit(err, env.usingKey) + } + if lsJSON { + return printJSON(env.stdout, list, true) + } + if len(list.Data) == 0 { + if lsWorkspace != "" || lsState != "" { + fmt.Fprintln(env.stderr, "No runs match.") + } else { + fmt.Fprintln(env.stderr, "No runs yet. Start one with: nan run \"\"") + } + return nil + } + tw := tabwriter.NewWriter(env.stdout, 0, 0, 2, ' ', 0) + fmt.Fprintln(tw, "ID\tSTATE\tWORKSPACE\tAGENT\tCREATED\tPROMPT") + for _, r := range list.Data { + fmt.Fprintf(tw, "%s\t%s\t%s\t%s\t%s\t%s\n", + runs.SanitizeLine(r.ID), + runs.SanitizeLine(r.State), + runs.SanitizeLine(r.Workspace.Name), + runs.SanitizeLine(r.Agent), + localTime(r.CreatedAt), + runs.Truncate(runs.SanitizeLine(r.PromptPreview), 60)) + } + if err := tw.Flush(); err != nil { + return err + } + if list.NextCursor != nil { + if lsLimit < 100 { + fmt.Fprintf(env.stderr, "showing the %d most recent — raise --limit (up to 100) for more\n", len(list.Data)) + } else { + fmt.Fprintf(env.stderr, "showing the %d most recent — narrow it with --ws or --state\n", len(list.Data)) + } + } + return nil +} + +func localTime(ts string) string { + t, err := time.Parse(time.RFC3339Nano, ts) + if err != nil { + return runs.SanitizeLine(ts) + } + return t.Local().Format("2006-01-02 15:04") +} + +func doRunsShow(ctx context.Context, env *runsEnv, id string) error { + if err := checkRunID(id); err != nil { + return err + } + run, err := env.client.GetRun(ctx, id) + if err != nil { + return notFound(err, id, env.usingKey) + } + if showJSON { + return printJSON(env.stdout, run, true) + } + + tw := tabwriter.NewWriter(env.stdout, 0, 0, 2, ' ', 0) + row := func(k, v string) { + if v != "" { + fmt.Fprintf(tw, "%s\t%s\n", k, v) + } + } + state := runs.SanitizeLine(run.State) + if run.QueuePosition != nil { + state += fmt.Sprintf(" (position %d)", *run.QueuePosition) + } + if run.CancelRequested && !api.Terminal(run.State) { + state += ", cancel requested" + } + row("id", runs.SanitizeLine(run.ID)) + row("state", state) + if run.ErrorCode != nil { + row("error", runs.SanitizeLine(*run.ErrorCode)) + } + if run.ExitCode != nil { + row("exit code", fmt.Sprint(*run.ExitCode)) + } + row("workspace", runs.SanitizeLine(run.Workspace.Name)) + agent := runs.SanitizeLine(run.Agent) + if run.Model != nil && *run.Model != "" { + agent += " · model " + runs.SanitizeLine(*run.Model) + } + row("agent", agent) + row("trigger", runs.SanitizeLine(run.Trigger)) + row("cwd", runs.SanitizeLine(run.Cwd)) + worktree := "auto" + if run.GitIsolation != nil { + worktree = map[bool]string{true: "yes", false: "no"}[*run.GitIsolation] + } + row("worktree", worktree) + if run.TimeoutSeconds > 0 { + row("timeout", (time.Duration(run.TimeoutSeconds) * time.Second).String()) + } + row("created", localTime(run.CreatedAt)) + if run.StartedAt != nil { + row("started", localTime(*run.StartedAt)) + } + if run.FinishedAt != nil { + row("finished", localTime(*run.FinishedAt)) + } + if res := run.Result; res != nil { + row("branch", runs.SanitizeLine(res.Branch)) + row("commit", runs.SanitizeLine(res.Commit)) + if res.ChangedFiles > 0 { + row("changed files", fmt.Sprint(res.ChangedFiles)) + } + for _, a := range res.Artifacts { + row("artifact", fmt.Sprintf("%s (%d bytes)", runs.SanitizeLine(a.Path), a.Bytes)) + } + } + if u := run.Usage; u != nil && (u.TokensIn != nil || u.TokensOut != nil) { + row("tokens", fmt.Sprintf("%s in, %s out", optInt(u.TokensIn), optInt(u.TokensOut))) + } + if err := tw.Flush(); err != nil { + return err + } + + prompt := run.Prompt + if prompt == "" { + prompt = run.PromptPreview + } + printBlock(env, "prompt", prompt) + if run.Result != nil { + printBlock(env, "summary", run.Result.Summary) + } + return nil +} + +func optInt(v *int64) string { + if v == nil { + return "?" + } + return fmt.Sprint(*v) +} + +func printBlock(env *runsEnv, title, text string) { + text = strings.TrimRight(runs.Sanitize(text), "\n") + if strings.TrimSpace(text) == "" { + return + } + fmt.Fprintf(env.stdout, "\n%s:\n", title) + for _, l := range strings.Split(text, "\n") { + fmt.Fprintf(env.stdout, " %s\n", l) + } +} + +func doRunsLogs(ctx context.Context, env *runsEnv, id string) error { + if err := checkRunID(id); err != nil { + return err + } + if logsAfter < 0 { + return usageErrorf("--after cannot be negative") + } + if logsFollow { + return follow(ctx, env, id, logsAfter, logsJSON, false) + } + + renderer := &runs.Renderer{Out: env.stdout} + after := logsAfter + done := false + for { + page, err := env.client.EventsPage(ctx, id, after, 500) + if err != nil { + return notFound(err, id, env.usingKey) + } + for _, ev := range page.Data { + if logsJSON { + if err := printJSON(env.stdout, ev, false); err != nil { + return err + } + } else { + renderer.Event(ev) + } + } + done = page.Done + if page.Done || len(page.Data) == 0 || page.NextAfter <= after { + break + } + after = page.NextAfter + } + renderer.Flush() + if !done { + fmt.Fprintf(env.stderr, "run %s is still going — follow it with: nan runs logs %s -f\n", id, id) + } + return nil +} + +func doRunsCancel(ctx context.Context, env *runsEnv, id string) error { + if err := checkRunID(id); err != nil { + return err + } + run, changed, err := env.client.CancelRun(ctx, id) + if err != nil { + return notFound(err, id, env.usingKey) + } + if cancelJSON { + return printJSON(env.stdout, run, true) + } + if !changed { + fmt.Fprintf(env.stdout, "run %s had already ended: %s\n", id, runs.SanitizeLine(run.State)) + return nil + } + fmt.Fprintf(env.stdout, "cancel requested for run %s — it stops within seconds; check with: nan runs show %s\n", id, id) + return nil +} diff --git a/cmd/runs_test.go b/cmd/runs_test.go new file mode 100644 index 0000000..9ee78d3 --- /dev/null +++ b/cmd/runs_test.go @@ -0,0 +1,212 @@ +package cmd + +import ( + "context" + "encoding/json" + "net/http" + "strings" + "testing" + + "github.com/nxssie/nan-cli/internal/api" +) + +func resetRunsFlags() { + lsWorkspace, lsState, lsLimit, lsJSON = "", "", 20, false + showJSON = false + logsFollow, logsAfter, logsJSON = false, 0, false + cancelJSON = false +} + +func TestRunsLsJSON(t *testing.T) { + defer resetRunsFlags() + resetRunsFlags() + h := newHarness(t, &fakePlatform{}) + lsJSON, lsWorkspace, lsState, lsLimit = true, "develop", "running", 5 + if err := doRunsLs(context.Background(), h.env); err != nil { + t.Fatal(err) + } + var list api.RunList + if err := json.Unmarshal(h.stdout.Bytes(), &list); err != nil { + t.Fatalf("%v: %q", err, h.stdout.String()) + } + if list.Object != "list" || len(list.Data) != 1 || list.Data[0].ID != testRunID { + t.Errorf("list = %+v", list) + } + if h.fake.listQuery != "limit=5&state=running&workspace=develop" { + t.Errorf("query = %q", h.fake.listQuery) + } +} + +func TestRunsLsTable(t *testing.T) { + defer resetRunsFlags() + resetRunsFlags() + h := newHarness(t, &fakePlatform{}) + if err := doRunsLs(context.Background(), h.env); err != nil { + t.Fatal(err) + } + out := h.stdout.String() + for _, want := range []string{"ID", "STATE", testRunID, "running", "develop", "pi", "review"} { + if !strings.Contains(out, want) { + t.Errorf("table lacks %q:\n%s", want, out) + } + } +} + +func TestRunsLsValidatesFlags(t *testing.T) { + defer resetRunsFlags() + for _, set := range []func(){ + func() { lsState = "done" }, + func() { lsLimit = 0 }, + func() { lsLimit = 101 }, + } { + resetRunsFlags() + set() + if err := checkLsFlags(); exitCode(err) != exitUsage { + t.Errorf("exit %d (%v)", exitCode(err), err) + } + } +} + +func TestRunsShow(t *testing.T) { + defer resetRunsFlags() + resetRunsFlags() + h := newHarness(t, &fakePlatform{finalState: "failed", finalError: "agent_failed"}) + if err := doRunsShow(context.Background(), h.env, testRunID); err != nil { + t.Fatal(err) + } + for _, want := range []string{testRunID, "failed", "agent_failed", "develop", "worktree", "auto", "prompt:"} { + if !strings.Contains(h.stdout.String(), want) { + t.Errorf("show lacks %q:\n%s", want, h.stdout.String()) + } + } + + h = newHarness(t, &fakePlatform{}) + showJSON = true + if err := doRunsShow(context.Background(), h.env, testRunID); err != nil { + t.Fatal(err) + } + var run api.Run + if err := json.Unmarshal(h.stdout.Bytes(), &run); err != nil || run.ID != testRunID { + t.Errorf("--json: %v %q", err, h.stdout.String()) + } +} + +func TestRunsLogsJSONIsOneEventPerLine(t *testing.T) { + defer resetRunsFlags() + resetRunsFlags() + h := newHarness(t, &fakePlatform{}) + logsJSON = true + logsAfter = 7 + if err := doRunsLogs(context.Background(), h.env, testRunID); err != nil { + t.Fatal(err) + } + lines := strings.Split(strings.TrimSpace(h.stdout.String()), "\n") + if len(lines) != 1 { + t.Fatalf("lines = %q", lines) + } + var ev api.Event + if err := json.Unmarshal([]byte(lines[0]), &ev); err != nil || ev.Seq != 1 || ev.Type != "message" { + t.Errorf("%v: %q", err, lines[0]) + } + if strings.Contains(h.stdout.String(), "\x1b") { + t.Error("a raw ESC in JSON output") + } + if len(h.fake.eventsQuery) != 1 || !strings.Contains(h.fake.eventsQuery[0], "after=7") { + t.Errorf("queries = %q", h.fake.eventsQuery) + } +} + +func TestRunsLogsTextIsSanitised(t *testing.T) { + defer resetRunsFlags() + resetRunsFlags() + h := newHarness(t, &fakePlatform{}) + if err := doRunsLogs(context.Background(), h.env, testRunID); err != nil { + t.Fatal(err) + } + if h.stdout.String() != "hello\n" { + t.Errorf("stdout = %q", h.stdout.String()) + } +} + +func TestRunsLogsFollowExitsWithTheRunsCode(t *testing.T) { + defer resetRunsFlags() + for _, tc := range []struct { + state, errorCode string + want int + }{{"succeeded", "", 0}, {"failed", "agent_failed", 1}, {"timed_out", "", 2}, {"cancelled", "", 3}, {"failed", "config_error", 4}} { + state, want := tc.state, tc.want + resetRunsFlags() + logsFollow = true + logsAfter = 1 + h := newHarness(t, &fakePlatform{finalState: state, finalError: tc.errorCode}) + err := doRunsLogs(context.Background(), h.env, testRunID) + if exitCode(err) != want { + t.Errorf("%s: exit %d (%v), want %d", state, exitCode(err), err, want) + } + if h.fake.lastEventIDs[0] != "1" { + t.Errorf("%s: --after 1 sent Last-Event-ID %q", state, h.fake.lastEventIDs[0]) + } + } +} + +func TestRunsLogsFollowJSONEndsWithTheRun(t *testing.T) { + defer resetRunsFlags() + resetRunsFlags() + logsFollow, logsJSON = true, true + h := newHarness(t, &fakePlatform{}) + if err := doRunsLogs(context.Background(), h.env, testRunID); err != nil { + t.Fatal(err) + } + lines := strings.Split(strings.TrimSpace(h.stdout.String()), "\n") + var run api.Run + if err := json.Unmarshal([]byte(lines[len(lines)-1]), &run); err != nil || run.Object != "run" { + t.Errorf("last line %q: %v", lines[len(lines)-1], err) + } +} + +func TestRunsCancel(t *testing.T) { + defer resetRunsFlags() + resetRunsFlags() + h := newHarness(t, &fakePlatform{}) + if err := doRunsCancel(context.Background(), h.env, testRunID); err != nil { + t.Fatal(err) + } + if h.fake.cancels != 1 || !strings.Contains(h.stdout.String(), "cancel requested") { + t.Errorf("cancels=%d stdout=%q", h.fake.cancels, h.stdout.String()) + } + + h = newHarness(t, &fakePlatform{cancelStatus: http.StatusOK}) + if err := doRunsCancel(context.Background(), h.env, testRunID); err != nil { + t.Fatal(err) + } + if !strings.Contains(h.stdout.String(), "already ended") { + t.Errorf("stdout = %q", h.stdout.String()) + } + + h = newHarness(t, &fakePlatform{}) + cancelJSON = true + if err := doRunsCancel(context.Background(), h.env, testRunID); err != nil { + t.Fatal(err) + } + var run api.Run + if err := json.Unmarshal(h.stdout.Bytes(), &run); err != nil || run.ID != testRunID { + t.Errorf("--json: %v %q", err, h.stdout.String()) + } +} + +func TestRunsNotFoundExits64(t *testing.T) { + defer resetRunsFlags() + resetRunsFlags() + h := newHarness(t, &fakePlatform{}) + other := "00000000-0000-4000-8000-000000000000" + for name, call := range map[string]func() error{ + "show": func() error { return doRunsShow(context.Background(), h.env, other) }, + "cancel": func() error { return doRunsCancel(context.Background(), h.env, other) }, + "logs": func() error { return doRunsLogs(context.Background(), h.env, other) }, + } { + err := call() + if exitCode(err) != exitUsage || !strings.Contains(err.Error(), "not found") { + t.Errorf("%s: exit %d (%v)", name, exitCode(err), err) + } + } +} diff --git a/go.mod b/go.mod index dcd01d2..fe4ee2a 100644 --- a/go.mod +++ b/go.mod @@ -1,6 +1,6 @@ module github.com/nxssie/nan-cli -go 1.26.8 +go 1.26.9 require ( github.com/charmbracelet/bubbles v1.0.0 diff --git a/internal/api/runs.go b/internal/api/runs.go new file mode 100644 index 0000000..80aea33 --- /dev/null +++ b/internal/api/runs.go @@ -0,0 +1,757 @@ +package api + +import ( + "bufio" + "bytes" + "context" + "encoding/json" + "errors" + "fmt" + "io" + "mime" + "net/http" + "net/url" + "strconv" + "strings" + "sync/atomic" + "time" +) + +// RunsBaseURL is where /v1/runs lives. Not api.nan.builders: that host sits +// behind an edge Worker that rewrites response bodies and may buffer a stream, +// and agent output is exactly the kind of text it rewrites. +const RunsBaseURL = "https://cloud-api.nan.builders" + +// Run states, as the platform names them. +const ( + StateQueued = "queued" + StateStarting = "starting" + StateRunning = "running" + StateSucceeded = "succeeded" + StateFailed = "failed" + StateCancelled = "cancelled" + StateTimedOut = "timed_out" +) + +// Terminal reports whether a run in this state will never change again. +func Terminal(state string) bool { + switch state { + case StateSucceeded, StateFailed, StateCancelled, StateTimedOut: + return true + } + return false +} + +type RunWorkspace struct { + ID string `json:"id"` + Name string `json:"name"` +} + +type RunArtifact struct { + Path string `json:"path"` + Bytes int64 `json:"bytes"` + SHA256 string `json:"sha256"` +} + +type RunResult struct { + Status string `json:"status"` + Summary string `json:"summary"` + Branch string `json:"branch"` + Commit string `json:"commit"` + ChangedFiles int `json:"changed_files"` + Artifacts []RunArtifact `json:"artifacts"` +} + +type RunUsage struct { + TokensIn *int64 `json:"tokens_in"` + TokensOut *int64 `json:"tokens_out"` + Source *string `json:"source"` +} + +// Run mirrors the run object of the platform's /v1/runs API. Every string in +// here except the ids is text a member's agent may have produced or shaped, +// so nothing prints it without sanitising it first. +type Run struct { + ID string `json:"id"` + Object string `json:"object"` + Workspace RunWorkspace `json:"workspace"` + Agent string `json:"agent"` + Model *string `json:"model"` + Trigger string `json:"trigger"` + State string `json:"state"` + QueuePosition *int `json:"queue_position"` + PromptPreview string `json:"prompt_preview"` + Prompt string `json:"prompt,omitempty"` + Cwd string `json:"cwd"` + GitIsolation *bool `json:"git_isolation"` + TimeoutSeconds int `json:"timeout_seconds"` + CancelRequested bool `json:"cancel_requested"` + ExitCode *int `json:"exit_code"` + ErrorCode *string `json:"error_code"` + Result *RunResult `json:"result"` + Usage *RunUsage `json:"usage"` + LastSeq int64 `json:"last_seq"` + CreatedAt string `json:"created_at"` + StartedAt *string `json:"started_at"` + FinishedAt *string `json:"finished_at"` +} + +// CreateRunRequest is the body of POST /v1/runs. Pointers are the fields the +// platform defaults when they are left out, which is not the same as zero. +type CreateRunRequest struct { + Workspace string `json:"workspace"` + Agent string `json:"agent"` + Prompt string `json:"prompt"` + Cwd *string `json:"cwd,omitempty"` + GitIsolation *bool `json:"git_isolation,omitempty"` + TimeoutSeconds *int `json:"timeout_seconds,omitempty"` + Model *string `json:"model,omitempty"` + Trigger string `json:"trigger,omitempty"` +} + +type RunList struct { + Object string `json:"object"` + Data []Run `json:"data"` + NextCursor *string `json:"next_cursor"` +} + +type ListRunsParams struct { + Workspace string + State string + Limit int + Cursor string +} + +// Event is one line of a run's event log. Data stays raw: its shape depends +// on Type, and the renderer is the one place that decides what to show. +type Event struct { + V int `json:"v"` + Seq int64 `json:"seq"` + TS string `json:"ts"` + Type string `json:"type"` + Data json.RawMessage `json:"data"` +} + +type EventPage struct { + Data []Event `json:"data"` + NextAfter int64 `json:"next_after"` + Done bool `json:"done"` +} + +// StateChange is the `state` frame of the event stream. +type StateChange struct { + State string `json:"state"` + ErrorCode *string `json:"error_code"` +} + +// Workspace is the part of GET /api/workspaces this CLI reads. +type Workspace struct { + ID string `json:"workspace_uuid"` + Name string `json:"name"` + Status string `json:"status"` +} + +// APIError is a refusal from the platform, in the OpenAI envelope /v1/runs +// answers with. Code is what callers branch on; Message is for people. +type APIError struct { + Status int + Type string + Code string + Param string + Message string +} + +func (e *APIError) Error() string { + msg := e.Message + if msg == "" { + msg = http.StatusText(e.Status) + } + if e.Code != "" && !strings.Contains(msg, e.Code) { + return fmt.Sprintf("%s (%s)", msg, e.Code) + } + return msg +} + +// ErrStreamLost is the event stream failing more times in a row than the +// client retries. The run itself is not affected: it goes on in the +// workspace, and following it again picks up where this left off. +var ErrStreamLost = errors.New("lost the event stream") + +// RunsClient talks to /v1/runs. It is separate from Client because it can +// authenticate with the member's API key when there is no session, and +// because one of its requests - the event stream - must not have a timeout. +type RunsClient struct { + baseURL string + token string + apiKey string + http *http.Client + stream *http.Client + + // The stream's retry policy. Fields so the tests can run it in + // milliseconds instead of minutes. + StreamRetries int + IdleTimeout time.Duration + HealthyAfter time.Duration + Backoff func(attempt int) time.Duration + + // OnRetry, when set, hears about each reconnect before its backoff, so + // a caller can say why nothing is arriving. + OnRetry func(attempt, max int, err error) +} + +// NewRunsClient builds a client for the session token or, when there is none, +// the member's API key. A session wins because it is what every other +// command uses; the key is the fallback for a machine that only ran Setup. +func NewRunsClient(token, apiKey string) *RunsClient { + return &RunsClient{ + baseURL: RunsBaseURL, + token: token, + apiKey: apiKey, + http: &http.Client{Timeout: requestTimeout, CheckRedirect: noRedirects}, + stream: &http.Client{CheckRedirect: noRedirects}, + StreamRetries: 5, + IdleTimeout: 45 * time.Second, + HealthyAfter: time.Minute, + Backoff: defaultBackoff, + } +} + +// WithBaseURL points the client somewhere else, for tests. +func (c *RunsClient) WithBaseURL(u string) *RunsClient { + c.baseURL = strings.TrimRight(u, "/") + return c +} + +// noRedirects keeps the credential on the request it was meant for. /v1/runs +// never redirects, and Go's default policy would forward the cookie or the +// Authorization header to a same-host http:// target. +func noRedirects(*http.Request, []*http.Request) error { return http.ErrUseLastResponse } + +func defaultBackoff(attempt int) time.Duration { + d := time.Second << attempt + if d > 16*time.Second { + d = 16 * time.Second + } + return d +} + +func (c *RunsClient) hasCredentials() bool { return c.token != "" || c.apiKey != "" } + +func (c *RunsClient) authorize(req *http.Request) { + switch { + case c.token != "": + req.Header.Set("Cookie", "nan_session="+c.token) + case c.apiKey != "": + req.Header.Set("Authorization", "Bearer "+c.apiKey) + } +} + +func (c *RunsClient) newRequest(ctx context.Context, method, path string, body any) (*http.Request, error) { + var reader io.Reader + if body != nil { + buf, err := json.Marshal(body) + if err != nil { + return nil, err + } + reader = bytes.NewReader(buf) + } + req, err := http.NewRequestWithContext(ctx, method, c.baseURL+path, reader) + if err != nil { + return nil, err + } + if body != nil { + req.Header.Set("Content-Type", "application/json") + } + req.Header.Set("Accept", "application/json") + req.Header.Set("User-Agent", "nan-cli") + c.authorize(req) + return req, nil +} + +// maxResponse caps what a JSON response may make this process hold. The +// largest this CLI asks for is a page of 500 events of at most 64 KiB each. +const maxResponse = 40 << 20 + +// do sends a request and decodes a 2xx JSON answer into out. It returns the +// status so callers can tell a 201 from an idempotent 200. +func (c *RunsClient) do(req *http.Request, out any) (int, error) { + resp, err := c.http.Do(req) + if err != nil { + return 0, fmt.Errorf("could not reach nan.builders: %w", err) + } + defer resp.Body.Close() + body, err := io.ReadAll(io.LimitReader(resp.Body, maxResponse)) + if err != nil { + return resp.StatusCode, fmt.Errorf("could not read the answer from nan.builders: %w", err) + } + if err := c.statusError(resp.StatusCode, body); err != nil { + return resp.StatusCode, err + } + if out != nil { + if err := json.Unmarshal(body, out); err != nil { + return resp.StatusCode, fmt.Errorf("unexpected answer from nan.builders: %w", err) + } + } + return resp.StatusCode, nil +} + +// statusError turns a non-2xx answer into an error. 401 with credentials is +// ErrSessionExpired; everything else is an APIError carrying the code from +// either envelope the platform uses: OpenAI's {"error":{...}} on /v1, and the +// bare {"error":"..."} on /api. +func (c *RunsClient) statusError(status int, body []byte) error { + if status < 400 { + return nil + } + if status == http.StatusUnauthorized && c.hasCredentials() { + return ErrSessionExpired + } + apiErr := &APIError{Status: status} + var openai struct { + Error struct { + Message string `json:"message"` + Type string `json:"type"` + Param *string `json:"param"` + Code any `json:"code"` + } `json:"error"` + } + var bare struct { + Error string `json:"error"` + } + switch { + case json.Unmarshal(body, &openai) == nil && (openai.Error.Message != "" || openai.Error.Code != nil): + apiErr.Message = openai.Error.Message + apiErr.Type = openai.Error.Type + if openai.Error.Param != nil { + apiErr.Param = *openai.Error.Param + } + switch code := openai.Error.Code.(type) { + case string: + apiErr.Code = code + case float64: + apiErr.Code = strconv.Itoa(int(code)) + } + case json.Unmarshal(body, &bare) == nil && bare.Error != "": + apiErr.Message = bare.Error + } + return apiErr +} + +// CreateRun queues a run. replayed is true when the platform answered an +// Idempotency-Key it had already seen with the run it created then. +func (c *RunsClient) CreateRun(ctx context.Context, in CreateRunRequest, idempotencyKey string) (run *Run, replayed bool, err error) { + req, err := c.newRequest(ctx, http.MethodPost, "/v1/runs", in) + if err != nil { + return nil, false, err + } + if idempotencyKey != "" { + req.Header.Set("Idempotency-Key", idempotencyKey) + } + run = &Run{} + status, err := c.do(req, run) + if err != nil { + return nil, false, err + } + return run, status == http.StatusOK, nil +} + +func (c *RunsClient) ListRuns(ctx context.Context, p ListRunsParams) (*RunList, error) { + q := url.Values{} + if p.Workspace != "" { + q.Set("workspace", p.Workspace) + } + if p.State != "" { + q.Set("state", p.State) + } + if p.Limit > 0 { + q.Set("limit", strconv.Itoa(p.Limit)) + } + if p.Cursor != "" { + q.Set("cursor", p.Cursor) + } + path := "/v1/runs" + if len(q) > 0 { + path += "?" + q.Encode() + } + req, err := c.newRequest(ctx, http.MethodGet, path, nil) + if err != nil { + return nil, err + } + list := &RunList{} + if _, err := c.do(req, list); err != nil { + return nil, err + } + return list, nil +} + +func (c *RunsClient) GetRun(ctx context.Context, id string) (*Run, error) { + req, err := c.newRequest(ctx, http.MethodGet, "/v1/runs/"+url.PathEscape(id), nil) + if err != nil { + return nil, err + } + run := &Run{} + if _, err := c.do(req, run); err != nil { + return nil, err + } + return run, nil +} + +// CancelRun asks for a run to stop. changed is false when the run had +// already finished and the platform left it as it was (200 instead of 202). +func (c *RunsClient) CancelRun(ctx context.Context, id string) (run *Run, changed bool, err error) { + req, err := c.newRequest(ctx, http.MethodPost, "/v1/runs/"+url.PathEscape(id)+"/cancel", nil) + if err != nil { + return nil, false, err + } + run = &Run{} + status, err := c.do(req, run) + if err != nil { + return nil, false, err + } + return run, status == http.StatusAccepted, nil +} + +// EventsPage reads one page of a run's events after seq `after`. +func (c *RunsClient) EventsPage(ctx context.Context, id string, after int64, limit int) (*EventPage, error) { + q := url.Values{} + q.Set("after", strconv.FormatInt(after, 10)) + if limit > 0 { + q.Set("limit", strconv.Itoa(limit)) + } + req, err := c.newRequest(ctx, http.MethodGet, "/v1/runs/"+url.PathEscape(id)+"/events?"+q.Encode(), nil) + if err != nil { + return nil, err + } + page := &EventPage{} + if _, err := c.do(req, page); err != nil { + return nil, err + } + return page, nil +} + +// ListWorkspaces is GET /api/workspaces, which `nan run` reads to pick the +// workspace when the member has exactly one. +func (c *RunsClient) ListWorkspaces(ctx context.Context) ([]Workspace, error) { + req, err := c.newRequest(ctx, http.MethodGet, "/api/workspaces", nil) + if err != nil { + return nil, err + } + var list []Workspace + if _, err := c.do(req, &list); err != nil { + return nil, err + } + return list, nil +} + +// Frame is what the event stream delivers to a Follow caller: either an +// event of the run's log or a change of its state. +type Frame struct { + Event *Event + State *StateChange +} + +// maxSSELine bounds one line of the stream. The platform caps an event at +// 64 KiB; a line this long is not one of ours and is not worth buffering. +const maxSSELine = 1 << 20 + +// maxSSEFrame bounds the data of one frame, which can span many lines. +const maxSSEFrame = 4 << 20 + +var errIdle = errors.New("the stream went quiet") + +// fatalStreamError is a failure reconnecting cannot fix. +type fatalStreamError struct{ err error } + +func (e *fatalStreamError) Error() string { return e.err.Error() } +func (e *fatalStreamError) Unwrap() error { return e.err } + +// Follow streams a run's events after seq `after`, calling onFrame for each +// one in order, until the platform sends `end` - whose run object it returns. +// +// The stream is expected to break: the platform closes it after 30 minutes, +// and a tunnel or a proxy in between can drop it at any time. A break is a +// reconnect with Last-Event-ID, so nothing is lost or repeated; only +// StreamRetries failed connections in a row give up, as ErrStreamLost. A 401 or a 404 is never retried. +func (c *RunsClient) Follow(ctx context.Context, id string, after int64, onFrame func(Frame)) (*Run, error) { + lastSeq := after + failures := 0 + var lastErr error + for { + if err := ctx.Err(); err != nil { + return nil, err + } + end, progressed, err := c.followOnce(ctx, id, &lastSeq, onFrame) + if end != nil { + return end, nil + } + if ctx.Err() != nil { + return nil, ctx.Err() + } + var fatal *fatalStreamError + if errors.As(err, &fatal) { + return nil, fatal.err + } + if progressed { + failures = 0 + } + if err != nil { + lastErr = err + } + failures++ + if failures > c.StreamRetries { + if lastErr == nil { + lastErr = errors.New("the platform closed the stream") + } + return nil, fmt.Errorf("%w: %v", ErrStreamLost, lastErr) + } + if c.OnRetry != nil { + c.OnRetry(failures, c.StreamRetries, lastErr) + } + backoff := c.Backoff + if backoff == nil { + backoff = defaultBackoff + } + select { + case <-ctx.Done(): + return nil, ctx.Err() + case <-time.After(backoff(failures - 1)): + } + } +} + +// followOnce runs one connection of the stream. progressed is whether the +// connection was a working one: it delivered a frame, or it stayed up and +// heartbeating for HealthyAfter. Only a connection that did neither counts +// towards giving up - otherwise something that answers 200 and hangs up +// straight away, over and over, would never be given up on, and a quiet run +// whose stream the platform recycles every 30 minutes would be. +func (c *RunsClient) followOnce(ctx context.Context, id string, lastSeq *int64, onFrame func(Frame)) (end *Run, progressed bool, err error) { + connCtx, cancel := context.WithCancel(ctx) + defer cancel() + + idle := c.IdleTimeout + if idle <= 0 { + idle = 45 * time.Second + } + healthyAfter := c.HealthyAfter + if healthyAfter <= 0 { + healthyAfter = time.Minute + } + + path := "/v1/runs/" + url.PathEscape(id) + "/events?after=" + strconv.FormatInt(*lastSeq, 10) + req, err := c.newRequest(connCtx, http.MethodGet, path, nil) + if err != nil { + return nil, false, &fatalStreamError{err} + } + req.Header.Set("Accept", "text/event-stream") + req.Header.Set("Cache-Control", "no-cache") + if *lastSeq > 0 { + req.Header.Set("Last-Event-ID", strconv.FormatInt(*lastSeq, 10)) + } + + // The idle limit covers the wait for the response headers too: a + // connection that is accepted and never answered is as dead as one that + // goes quiet halfway. + var headerTimedOut atomic.Bool + headerTimer := time.AfterFunc(idle, func() { + headerTimedOut.Store(true) + cancel() + }) + started := time.Now() + resp, err := c.stream.Do(req) + headerTimer.Stop() + if err != nil { + if headerTimedOut.Load() && ctx.Err() == nil { + return nil, false, errIdle + } + return nil, false, err + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusOK { + body, _ := io.ReadAll(io.LimitReader(resp.Body, 64<<10)) + statusErr := c.statusError(resp.StatusCode, body) + if statusErr == nil { + statusErr = fmt.Errorf("HTTP %d", resp.StatusCode) + } + if resp.StatusCode == http.StatusTooManyRequests || resp.StatusCode >= 500 { + return nil, false, statusErr + } + return nil, false, &fatalStreamError{statusErr} + } + if mt, _, _ := mime.ParseMediaType(resp.Header.Get("Content-Type")); mt != "text/event-stream" { + return nil, false, fmt.Errorf("expected an event stream, got %q", resp.Header.Get("Content-Type")) + } + + // The idle limit is about the connection going quiet, so it is measured + // from the last bytes read, not the last complete line: a long line that + // is still arriving is not idle, and must end on its own terms (as the + // fatal oversize error below) rather than as a reconnect. + var lastRead atomic.Int64 + lastRead.Store(time.Now().UnixNano()) + body := &activityReader{r: resp.Body, last: &lastRead} + + lines := make(chan string) + readErr := make(chan error, 1) + go func() { + scanner := bufio.NewScanner(body) + scanner.Buffer(make([]byte, 0, 64<<10), maxSSELine) + for scanner.Scan() { + select { + case lines <- scanner.Text(): + case <-connCtx.Done(): + return + } + } + if err := scanner.Err(); err != nil { + readErr <- err + } else { + readErr <- io.EOF + } + }() + + timer := time.NewTimer(idle) + defer timer.Stop() + + var ( + eventName string + eventID string + data []string + dataBytes int + heard bool + ) + healthy := func() bool { return progressed || (heard && time.Since(started) >= healthyAfter) } + readEnded := func(err error) (*Run, bool, error) { + if errors.Is(err, bufio.ErrTooLong) { + return nil, healthy(), &fatalStreamError{fmt.Errorf("the platform sent a stream line over %d bytes", maxSSELine)} + } + if errors.Is(err, io.EOF) { + return nil, healthy(), nil + } + return nil, healthy(), err + } + for { + select { + case <-ctx.Done(): + return nil, healthy(), ctx.Err() + case <-timer.C: + // A read that already ended decides the outcome, whichever of + // the two select happened to pick. + select { + case err := <-readErr: + return readEnded(err) + default: + } + if quiet := time.Since(time.Unix(0, lastRead.Load())); quiet < idle { + timer.Reset(idle - quiet) + continue + } + return nil, healthy(), errIdle + case err := <-readErr: + return readEnded(err) + case line := <-lines: + heard = true + if !timer.Stop() { + select { + case <-timer.C: + default: + } + } + timer.Reset(idle) + + line = strings.TrimSuffix(line, "\r") + if line == "" { + if len(data) == 0 && eventName == "" { + continue + } + payload := strings.Join(data, "\n") + name := eventName + frameID := eventID + eventName, eventID, data, dataBytes = "", "", nil, 0 + run, delivered := dispatchFrame(name, frameID, payload, lastSeq, onFrame) + if run != nil { + if run.ID != id { + return nil, true, &fatalStreamError{errors.New("the event stream ended with a different run")} + } + return run, true, nil + } + if delivered { + progressed = true + } + continue + } + if strings.HasPrefix(line, ":") { + continue + } + field, value, _ := strings.Cut(line, ":") + value = strings.TrimPrefix(value, " ") + switch field { + case "event": + eventName = value + case "id": + eventID = value + case "data": + dataBytes += len(value) + if dataBytes > maxSSEFrame { + return nil, healthy(), &fatalStreamError{fmt.Errorf("the platform sent a stream frame over %d bytes", maxSSEFrame)} + } + data = append(data, value) + } + } + } +} + +// activityReader records when it last read any bytes, so the idle limit +// can tell a quiet connection from one busy with a long line. +type activityReader struct { + r io.Reader + last *atomic.Int64 +} + +func (a *activityReader) Read(p []byte) (int, error) { + n, err := a.r.Read(p) + if n > 0 { + a.last.Store(time.Now().UnixNano()) + } + return n, err +} + +// dispatchFrame hands one complete frame to the caller, and returns the run +// when the frame is `end`. Events at or below lastSeq were already delivered +// on a previous connection and are dropped, so a replay never shows twice. +func dispatchFrame(name, id, payload string, lastSeq *int64, onFrame func(Frame)) (end *Run, delivered bool) { + switch name { + case "", "run_event": + var ev Event + if json.Unmarshal([]byte(payload), &ev) != nil { + return nil, false + } + if ev.Seq == 0 { + if n, err := strconv.ParseInt(id, 10, 64); err == nil { + ev.Seq = n + } + } + // A replay of what was already shown is not progress: a server that + // ignored Last-Event-ID would otherwise keep the retry budget full. + if ev.Seq <= *lastSeq { + return nil, false + } + *lastSeq = ev.Seq + onFrame(Frame{Event: &ev}) + return nil, true + case "state": + var st StateChange + if json.Unmarshal([]byte(payload), &st) != nil { + return nil, false + } + onFrame(Frame{State: &st}) + return nil, false + case "end": + var run Run + if json.Unmarshal([]byte(payload), &run) != nil || run.ID == "" { + return nil, false + } + return &run, true + } + return nil, false +} diff --git a/internal/api/runs_test.go b/internal/api/runs_test.go new file mode 100644 index 0000000..305fc00 --- /dev/null +++ b/internal/api/runs_test.go @@ -0,0 +1,659 @@ +package api + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "io" + "net/http" + "net/http/httptest" + "strings" + "sync" + "sync/atomic" + "testing" + "time" +) + +const runID = "6f1c2a9b-0000-4000-8000-000000000001" + +func testRunsClient(t *testing.T, h http.Handler, token, key string) *RunsClient { + t.Helper() + srv := httptest.NewServer(h) + t.Cleanup(srv.Close) + c := NewRunsClient(token, key).WithBaseURL(srv.URL) + c.Backoff = func(int) time.Duration { return time.Millisecond } + c.IdleTimeout = 200 * time.Millisecond + return c +} + +func runJSON(state string) string { + return fmt.Sprintf(`{"id":%q,"object":"run","workspace":{"id":"w1","name":"develop"},"agent":"pi","state":%q,"last_seq":2,"created_at":"2026-10-10T12:00:00Z"}`, runID, state) +} + +func TestRunsClientSendsTheSessionAsACookie(t *testing.T) { + var cookie, authz string + c := testRunsClient(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + cookie, authz = r.Header.Get("Cookie"), r.Header.Get("Authorization") + _, _ = io.WriteString(w, runJSON("running")) + }), "sess", "sk-key") + if _, err := c.GetRun(context.Background(), runID); err != nil { + t.Fatal(err) + } + if cookie != "nan_session=sess" { + t.Errorf("cookie = %q", cookie) + } + if authz != "" { + t.Errorf("sent the API key alongside a session: %q", authz) + } +} + +func TestRunsClientFallsBackToTheAPIKey(t *testing.T) { + var cookie, authz string + c := testRunsClient(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + cookie, authz = r.Header.Get("Cookie"), r.Header.Get("Authorization") + _, _ = io.WriteString(w, runJSON("running")) + }), "", "sk-key") + if _, err := c.GetRun(context.Background(), runID); err != nil { + t.Fatal(err) + } + if authz != "Bearer sk-key" || cookie != "" { + t.Errorf("authz=%q cookie=%q", authz, cookie) + } +} + +func TestRunsClient401IsAnExpiredSession(t *testing.T) { + c := testRunsClient(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusUnauthorized) + _, _ = io.WriteString(w, `{"error":{"message":"invalid session","type":"auth_error","code":"invalid_api_key"}}`) + }), "stale", "") + if _, err := c.GetRun(context.Background(), runID); !errors.Is(err, ErrSessionExpired) { + t.Fatalf("got %v", err) + } +} + +func TestRunsClientParsesBothErrorEnvelopes(t *testing.T) { + for _, tc := range []struct { + body, code, msg, param string + }{ + {`{"error":{"message":"workspace is not running","type":"invalid_request_error","param":null,"code":"workspace_not_running"}}`, "workspace_not_running", "workspace is not running", ""}, + {`{"error":{"message":"bad timeout","type":"invalid_request_error","param":"timeout_seconds","code":"invalid_request"}}`, "invalid_request", "bad timeout", "timeout_seconds"}, + {`{"error":"workspaces are not enabled"}`, "", "workspaces are not enabled", ""}, + } { + c := testRunsClient(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusConflict) + _, _ = io.WriteString(w, tc.body) + }), "t", "") + _, err := c.GetRun(context.Background(), runID) + var apiErr *APIError + if !errors.As(err, &apiErr) { + t.Fatalf("%s: got %v", tc.body, err) + } + if apiErr.Status != 409 || apiErr.Code != tc.code || apiErr.Message != tc.msg || apiErr.Param != tc.param { + t.Errorf("%s: parsed %+v", tc.body, apiErr) + } + } +} + +func TestCreateRunSendsTheBodyAndTheIdempotencyKey(t *testing.T) { + var got map[string]any + var idem, method, path string + status := http.StatusCreated + c := testRunsClient(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + method, path, idem = r.Method, r.URL.Path, r.Header.Get("Idempotency-Key") + _ = json.NewDecoder(r.Body).Decode(&got) + w.WriteHeader(status) + _, _ = io.WriteString(w, runJSON("queued")) + }), "t", "") + + timeout := 600 + iso := false + run, replayed, err := c.CreateRun(context.Background(), CreateRunRequest{ + Workspace: "develop", Agent: "pi", Prompt: "review PR 42", + TimeoutSeconds: &timeout, GitIsolation: &iso, Trigger: "cli", + }, "key-1") + if err != nil { + t.Fatal(err) + } + if method != http.MethodPost || path != "/v1/runs" || idem != "key-1" { + t.Errorf("%s %s idem=%q", method, path, idem) + } + if replayed { + t.Error("a 201 reads as a replay") + } + if run.ID != runID || run.State != StateQueued { + t.Errorf("run = %+v", run) + } + if got["workspace"] != "develop" || got["agent"] != "pi" || got["prompt"] != "review PR 42" || + got["timeout_seconds"] != float64(600) || got["git_isolation"] != false || got["trigger"] != "cli" { + t.Errorf("body = %v", got) + } + for _, absent := range []string{"cwd", "model"} { + if _, ok := got[absent]; ok { + t.Errorf("sent %q although it was not set, which overrides the platform default", absent) + } + } + + status = http.StatusOK + if _, replayed, _ = c.CreateRun(context.Background(), CreateRunRequest{Workspace: "develop", Agent: "pi", Prompt: "x"}, "key-1"); !replayed { + t.Error("a 200 on create is an idempotent replay") + } +} + +func TestListRunsSendsTheFilters(t *testing.T) { + var query string + c := testRunsClient(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + query = r.URL.RawQuery + _, _ = io.WriteString(w, `{"object":"list","data":[`+runJSON("running")+`],"next_cursor":"abc"}`) + }), "t", "") + list, err := c.ListRuns(context.Background(), ListRunsParams{Workspace: "develop", State: "running", Limit: 5}) + if err != nil { + t.Fatal(err) + } + if query != "limit=5&state=running&workspace=develop" { + t.Errorf("query = %q", query) + } + if len(list.Data) != 1 || list.NextCursor == nil || *list.NextCursor != "abc" { + t.Errorf("list = %+v", list) + } +} + +func TestCancelRunTellsAcceptedFromAlreadyFinished(t *testing.T) { + status := http.StatusAccepted + c := testRunsClient(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost || r.URL.Path != "/v1/runs/"+runID+"/cancel" { + t.Errorf("%s %s", r.Method, r.URL.Path) + } + w.WriteHeader(status) + _, _ = io.WriteString(w, runJSON("running")) + }), "t", "") + if _, changed, err := c.CancelRun(context.Background(), runID); err != nil || !changed { + t.Errorf("202: changed=%v err=%v", changed, err) + } + status = http.StatusOK + if _, changed, err := c.CancelRun(context.Background(), runID); err != nil || changed { + t.Errorf("200: changed=%v err=%v", changed, err) + } +} + +func TestRunIDsAreEscapedIntoThePath(t *testing.T) { + var raw string + c := testRunsClient(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + raw = r.URL.EscapedPath() + _, _ = io.WriteString(w, runJSON("running")) + }), "t", "") + _, _ = c.GetRun(context.Background(), "../../api/keys") + if strings.Contains(raw, "/api/keys") { + t.Errorf("an id walked out of /v1/runs: %q", raw) + } +} + +func TestListWorkspacesReadsTheArray(t *testing.T) { + c := testRunsClient(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path != "/api/workspaces" { + t.Errorf("path %q", r.URL.Path) + } + _, _ = io.WriteString(w, `[{"workspace_uuid":"u1","name":"develop","status":"running","ssh_command":"ssh x"}]`) + }), "t", "") + list, err := c.ListWorkspaces(context.Background()) + if err != nil { + t.Fatal(err) + } + if len(list) != 1 || list[0].Name != "develop" || list[0].ID != "u1" { + t.Errorf("list = %+v", list) + } +} + +// --- the event stream --- + +func sseEvent(seq int64, typ, data string) string { + return fmt.Sprintf("id: %d\nevent: run_event\ndata: {\"v\":1,\"seq\":%d,\"ts\":\"t\",\"type\":%q,\"data\":%s}\n\n", seq, seq, typ, data) +} + +func sseEnd(state string) string { return "event: end\ndata: " + runJSON(state) + "\n\n" } + +func collect(frames *[]Frame, mu *sync.Mutex) func(Frame) { + return func(f Frame) { + mu.Lock() + defer mu.Unlock() + *frames = append(*frames, f) + } +} + +func TestFollowReadsEventsStatesAndEnd(t *testing.T) { + c := testRunsClient(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Header.Get("Accept") != "text/event-stream" { + t.Errorf("Accept = %q", r.Header.Get("Accept")) + } + if r.URL.Query().Get("after") != "0" { + t.Errorf("after = %q", r.URL.Query().Get("after")) + } + w.Header().Set("Content-Type", "text/event-stream") + _, _ = io.WriteString(w, ": hb\n\n") + _, _ = io.WriteString(w, "event: state\ndata: {\"state\":\"running\",\"error_code\":null}\n\n") + _, _ = io.WriteString(w, sseEvent(1, "message", `{"role":"assistant","text":"hi","final":true}`)) + // CRLF endings and a multi-line data field are still one frame. + _, _ = io.WriteString(w, "id: 2\r\nevent: run_event\r\ndata: {\"v\":1,\"seq\":2,\r\ndata: \"type\":\"log\",\"data\":{}}\r\n\r\n") + _, _ = io.WriteString(w, sseEnd("succeeded")) + }), "t", "") + + var frames []Frame + var mu sync.Mutex + run, err := c.Follow(context.Background(), runID, 0, collect(&frames, &mu)) + if err != nil { + t.Fatal(err) + } + if run.State != StateSucceeded { + t.Errorf("end run = %+v", run) + } + if len(frames) != 3 || frames[0].State == nil || frames[0].State.State != "running" || + frames[1].Event == nil || frames[1].Event.Seq != 1 || frames[2].Event == nil || frames[2].Event.Type != "log" { + t.Fatalf("frames = %+v", frames) + } +} + +// The platform closes a stream after 30 minutes and anything in between can +// drop it sooner. The reconnect has to resume after the last event seen, and +// an event the server replays anyway must not print twice. +func TestFollowReconnectsWithLastEventID(t *testing.T) { + var mu sync.Mutex + var lastEventIDs []string + calls := 0 + c := testRunsClient(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + mu.Lock() + calls++ + n := calls + lastEventIDs = append(lastEventIDs, r.Header.Get("Last-Event-ID")) + mu.Unlock() + w.Header().Set("Content-Type", "text/event-stream") + switch n { + case 1: + _, _ = io.WriteString(w, sseEvent(1, "log", `{}`)) + _, _ = io.WriteString(w, sseEvent(2, "log", `{}`)) + // and the connection closes without `end` + case 2: + w.WriteHeader(http.StatusBadGateway) + default: + _, _ = io.WriteString(w, sseEvent(2, "log", `{}`)) // a replayed duplicate + _, _ = io.WriteString(w, sseEvent(3, "log", `{}`)) + _, _ = io.WriteString(w, sseEnd("failed")) + } + }), "t", "") + + var frames []Frame + var fmu sync.Mutex + run, err := c.Follow(context.Background(), runID, 0, collect(&frames, &fmu)) + if err != nil { + t.Fatal(err) + } + if run.State != StateFailed { + t.Errorf("state = %s", run.State) + } + var seqs []int64 + for _, f := range frames { + seqs = append(seqs, f.Event.Seq) + } + if fmt.Sprint(seqs) != "[1 2 3]" { + t.Errorf("seqs = %v", seqs) + } + if fmt.Sprint(lastEventIDs) != "[ 2 2]" { + t.Errorf("Last-Event-ID per connection = %q", lastEventIDs) + } +} + +func TestFollowGivesUpAfterFiveFailuresInARow(t *testing.T) { + var mu sync.Mutex + calls := 0 + c := testRunsClient(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + mu.Lock() + calls++ + mu.Unlock() + w.WriteHeader(http.StatusServiceUnavailable) + }), "t", "") + _, err := c.Follow(context.Background(), runID, 0, func(Frame) {}) + if !errors.Is(err, ErrStreamLost) { + t.Fatalf("got %v", err) + } + if calls != 6 { + t.Errorf("connected %d times, want the first try plus 5 retries", calls) + } +} + +// A stream that is accepted and then says nothing - not even the 15 s +// heartbeat - is dead in a way the TCP connection does not notice. +func TestFollowReconnectsAfterIdleTimeout(t *testing.T) { + var mu sync.Mutex + calls := 0 + release := make(chan struct{}) + c := testRunsClient(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + mu.Lock() + calls++ + n := calls + mu.Unlock() + w.Header().Set("Content-Type", "text/event-stream") + if n == 1 { + w.(http.Flusher).Flush() + select { + case <-r.Context().Done(): + case <-release: + } + return + } + _, _ = io.WriteString(w, sseEnd("succeeded")) + }), "t", "") + defer close(release) + c.IdleTimeout = 50 * time.Millisecond + run, err := c.Follow(context.Background(), runID, 0, func(Frame) {}) + if err != nil { + t.Fatal(err) + } + if run.State != StateSucceeded || calls != 2 { + t.Errorf("state=%s calls=%d", run.State, calls) + } +} + +// The 30 s timeout every other request has would cut a healthy stream off +// mid-run, so the stream client must not have one. +func TestTheStreamHasNoClientTimeout(t *testing.T) { + c := NewRunsClient("t", "") + if c.stream.Timeout != 0 { + t.Errorf("stream timeout = %s", c.stream.Timeout) + } + if c.http.Timeout == 0 { + t.Error("the ordinary requests lost their timeout") + } + if c.StreamRetries != 5 || c.IdleTimeout != 45*time.Second { + t.Errorf("retries=%d idle=%s", c.StreamRetries, c.IdleTimeout) + } +} + +func TestFollowDoesNotRetryAuthOrNotFound(t *testing.T) { + for _, tc := range []struct { + status int + check func(error) bool + }{ + {http.StatusUnauthorized, func(err error) bool { return errors.Is(err, ErrSessionExpired) }}, + {http.StatusNotFound, func(err error) bool { var e *APIError; return errors.As(err, &e) && e.Status == 404 }}, + } { + calls := 0 + c := testRunsClient(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + calls++ + w.WriteHeader(tc.status) + }), "t", "") + _, err := c.Follow(context.Background(), runID, 0, func(Frame) {}) + if !tc.check(err) || calls != 1 { + t.Errorf("%d: err=%v calls=%d", tc.status, err, calls) + } + } +} + +func TestFollowStopsWhenTheContextIsCancelled(t *testing.T) { + c := testRunsClient(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "text/event-stream") + w.(http.Flusher).Flush() + <-r.Context().Done() + }), "t", "") + c.IdleTimeout = time.Minute + ctx, cancel := context.WithCancel(context.Background()) + go func() { time.Sleep(30 * time.Millisecond); cancel() }() + if _, err := c.Follow(ctx, runID, 0, func(Frame) {}); !errors.Is(err, context.Canceled) { + t.Errorf("got %v", err) + } +} + +// A connection that is accepted and never answered must not hang the CLI: +// the idle limit covers the wait for the headers as well. +func TestFollowTimesOutWaitingForHeaders(t *testing.T) { + c := testRunsClient(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + <-r.Context().Done() + }), "t", "") + c.IdleTimeout = 30 * time.Millisecond + c.StreamRetries = 1 + done := make(chan error, 1) + go func() { _, err := c.Follow(context.Background(), runID, 0, func(Frame) {}); done <- err }() + select { + case err := <-done: + if !errors.Is(err, ErrStreamLost) { + t.Errorf("got %v", err) + } + case <-time.After(3 * time.Second): + t.Fatal("Follow hung on a server that never sends headers") + } +} + +// Something in between answering 200 with JSON - or with a heartbeat and a +// hang-up - must count as a failure, or the client reconnects forever. +func TestFollowGivesUpOnStreamsThatNeverWork(t *testing.T) { + for name, h := range map[string]http.HandlerFunc{ + "not an event stream": func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + _, _ = io.WriteString(w, `{"data":[]}`) + }, + "heartbeat and hang up": func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "text/event-stream") + _, _ = io.WriteString(w, ": hb\n\n") + }, + } { + var mu sync.Mutex + calls := 0 + c := testRunsClient(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + mu.Lock() + calls++ + mu.Unlock() + h(w, r) + }), "t", "") + done := make(chan error, 1) + go func() { _, err := c.Follow(context.Background(), runID, 0, func(Frame) {}); done <- err }() + select { + case err := <-done: + mu.Lock() + n := calls + mu.Unlock() + if !errors.Is(err, ErrStreamLost) || n != 6 { + t.Errorf("%s: err=%v calls=%d", name, err, n) + } + case <-time.After(3 * time.Second): + t.Fatalf("%s: reconnected forever", name) + } + } +} + +// The budget is for failures in a row: a connection that delivered frames, +// or that stayed up heartbeating, starts the count again. +func TestFollowRetryBudgetResetsOnWorkingConnections(t *testing.T) { + t.Run("frames", func(t *testing.T) { + var mu sync.Mutex + calls := 0 + c := testRunsClient(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + mu.Lock() + calls++ + n := calls + mu.Unlock() + switch { + case n == 4: + w.Header().Set("Content-Type", "text/event-stream") + _, _ = io.WriteString(w, sseEvent(1, "log", `{}`)) + case n == 8: + w.Header().Set("Content-Type", "text/event-stream") + _, _ = io.WriteString(w, sseEnd("succeeded")) + default: + w.WriteHeader(http.StatusServiceUnavailable) + } + }), "t", "") + if _, err := c.Follow(context.Background(), runID, 0, func(Frame) {}); err != nil { + t.Fatalf("7 failures split by a working connection gave up: %v", err) + } + }) + t.Run("long-lived heartbeats", func(t *testing.T) { + var mu sync.Mutex + calls := 0 + c := testRunsClient(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + mu.Lock() + calls++ + n := calls + mu.Unlock() + w.Header().Set("Content-Type", "text/event-stream") + if n < 9 { + _, _ = io.WriteString(w, ": hb\n\n") + w.(http.Flusher).Flush() + time.Sleep(20 * time.Millisecond) + return + } + _, _ = io.WriteString(w, sseEnd("succeeded")) + }), "t", "") + c.HealthyAfter = 10 * time.Millisecond + if _, err := c.Follow(context.Background(), runID, 0, func(Frame) {}); err != nil { + t.Fatalf("a quiet run whose stream is recycled gave up: %v", err) + } + }) +} + +func TestFollowRetriesTooManyRequests(t *testing.T) { + var calls atomic.Int32 + c := testRunsClient(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if calls.Add(1) == 1 { + w.WriteHeader(http.StatusTooManyRequests) + return + } + w.Header().Set("Content-Type", "text/event-stream") + _, _ = io.WriteString(w, sseEnd("succeeded")) + }), "t", "") + if _, err := c.Follow(context.Background(), runID, 0, func(Frame) {}); err != nil || calls.Load() != 2 { + t.Errorf("err=%v calls=%d", err, calls.Load()) + } +} + +func TestFollowSkipsAMalformedEnd(t *testing.T) { + var calls atomic.Int32 + c := testRunsClient(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + n := calls.Add(1) + w.Header().Set("Content-Type", "text/event-stream") + if n == 1 { + _, _ = io.WriteString(w, "event: end\ndata: {not json\n\n") + return + } + _, _ = io.WriteString(w, sseEnd("failed")) + }), "t", "") + run, err := c.Follow(context.Background(), runID, 0, func(Frame) {}) + if err != nil || run.State != StateFailed { + t.Errorf("run=%v err=%v", run, err) + } +} + +func TestFollowRefusesOversizedLinesAndFrames(t *testing.T) { + for name, body := range map[string]string{ + "line": "data: " + strings.Repeat("a", maxSSELine+1) + "\n\n", + "frame": strings.Repeat("data: "+strings.Repeat("a", maxSSELine/2)+"\n", 10) + "\n", + } { + var calls atomic.Int32 + c := testRunsClient(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + calls.Add(1) + w.Header().Set("Content-Type", "text/event-stream") + _, _ = io.WriteString(w, body) + }), "t", "") + _, err := c.Follow(context.Background(), runID, 0, func(Frame) {}) + if err == nil || errors.Is(err, ErrStreamLost) || calls.Load() != 1 { + t.Errorf("%s: err=%v calls=%d (want a fatal error, no retry)", name, err, calls.Load()) + } + } +} + +// The idle limit is about a connection that has gone quiet, not about one +// slow to finish a line: an oversized line that trickles in for longer than +// IdleTimeout must still end as the fatal oversize error on the first +// connection, never as an idle reconnect. (On a slow runner the 1 MiB line of +// the test above took longer than the idle limit to arrive, and was retried.) +func TestFollowOversizedLineArrivingSlowlyIsNotIdle(t *testing.T) { + var calls atomic.Int32 + c := testRunsClient(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + calls.Add(1) + w.Header().Set("Content-Type", "text/event-stream") + _, _ = io.WriteString(w, "data: ") + chunk := strings.Repeat("a", 128<<10) + for sent := 0; sent <= maxSSELine; sent += len(chunk) { + if _, err := io.WriteString(w, chunk); err != nil { + return + } + w.(http.Flusher).Flush() + select { + case <-r.Context().Done(): + return + case <-time.After(20 * time.Millisecond): + } + } + }), "t", "") + c.IdleTimeout = 60 * time.Millisecond + _, err := c.Follow(context.Background(), runID, 0, func(Frame) {}) + if err == nil || errors.Is(err, ErrStreamLost) || !strings.Contains(err.Error(), "stream line over") || calls.Load() != 1 { + t.Errorf("err=%v calls=%d (want the fatal oversize error, no retry)", err, calls.Load()) + } +} + +// /v1/runs never redirects, and following one would carry the credential to +// wherever it points. +func TestRunsClientDoesNotFollowRedirects(t *testing.T) { + hit := false + target := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { hit = true })) + defer target.Close() + c := testRunsClient(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + http.Redirect(w, r, target.URL+r.URL.Path, http.StatusFound) + }), "secret-session", "") + _, _ = c.GetRun(context.Background(), runID) + _, _ = c.Follow(context.Background(), runID, 0, func(Frame) {}) + if hit { + t.Error("followed a redirect with the session attached") + } +} + +func TestTerminal(t *testing.T) { + for state, want := range map[string]bool{"queued": false, "starting": false, "running": false, "succeeded": true, "failed": true, "cancelled": true, "timed_out": true} { + if Terminal(state) != want { + t.Errorf("Terminal(%s) != %v", state, want) + } + } +} + +// A server that ignores Last-Event-ID and replays old events before hanging +// up has shown nothing new: that is not a working connection. +func TestFollowReplaysAloneDoNotRefillTheBudget(t *testing.T) { + var calls atomic.Int32 + c := testRunsClient(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + calls.Add(1) + w.Header().Set("Content-Type", "text/event-stream") + _, _ = io.WriteString(w, sseEvent(1, "log", `{}`)) + }), "t", "") + done := make(chan error, 1) + go func() { _, err := c.Follow(context.Background(), runID, 0, func(Frame) {}); done <- err }() + select { + case err := <-done: + // the first connection delivers seq 1 and closes (one failure, budget + // fresh), then five that only replay it use the rest + if !errors.Is(err, ErrStreamLost) || calls.Load() != 6 { + t.Errorf("err=%v calls=%d", err, calls.Load()) + } + case <-time.After(3 * time.Second): + t.Fatal("reconnected forever on replays") + } +} + +func TestFollowRefusesAnEndForAnotherRun(t *testing.T) { + c := testRunsClient(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "text/event-stream") + _, _ = io.WriteString(w, sseEnd("succeeded")) + }), "t", "") + if _, err := c.Follow(context.Background(), "00000000-0000-4000-8000-000000000000", 0, func(Frame) {}); err == nil { + t.Error("took the end of a different run as this one's") + } +} + +func TestOnRetryHearsEachReconnect(t *testing.T) { + c := testRunsClient(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusBadGateway) + }), "t", "") + var attempts []int + c.OnRetry = func(attempt, max int, err error) { attempts = append(attempts, attempt) } + _, _ = c.Follow(context.Background(), runID, 0, func(Frame) {}) + if fmt.Sprint(attempts) != "[1 2 3 4 5]" { + t.Errorf("attempts = %v", attempts) + } +} diff --git a/internal/runs/render.go b/internal/runs/render.go new file mode 100644 index 0000000..c90594f --- /dev/null +++ b/internal/runs/render.go @@ -0,0 +1,335 @@ +// Package runs turns the events of a run into lines on a terminal. +// +// Everything a run produces comes out of the member's workspace, where the +// agent - and anything that got into its context - can write whatever bytes +// it likes. Printed raw, an escape sequence in that text can move the cursor, +// rewrite lines already on screen, set the window title, or on some terminals +// write to the clipboard. So no guest string reaches the terminal without +// going through Sanitize. +package runs + +import ( + "encoding/json" + "fmt" + "io" + "strings" + "unicode/utf8" + + "github.com/nxssie/nan-cli/internal/api" +) + +// Sanitize removes ANSI escape sequences and control characters from text +// that came from a run, keeping newlines and tabs. Invalid UTF-8 becomes +// U+FFFD, carriage returns are dropped (a bare one rewinds the line), and so +// are the bidirectional overrides that make text read differently from what +// it is. +func Sanitize(s string) string { + if !strings.ContainsFunc(s, unsafeRune) && utf8.ValidString(s) { + return s + } + var b strings.Builder + b.Grow(len(s)) + for i := 0; i < len(s); { + r, size := utf8.DecodeRuneInString(s[i:]) + if r == utf8.RuneError && size <= 1 { + b.WriteRune(utf8.RuneError) + i++ + continue + } + if r == 0x1b { + i += escapeLen(s[i:]) + continue + } + i += size + if unsafeRune(r) { + continue + } + b.WriteRune(r) + } + return b.String() +} + +// SanitizeLine is Sanitize for a value that has to stay on one line, such as +// a cell of a table: line breaks and tabs become spaces. +func SanitizeLine(s string) string { + s = Sanitize(s) + return strings.Map(func(r rune) rune { + if r == '\n' || r == '\t' { + return ' ' + } + return r + }, s) +} + +func unsafeRune(r rune) bool { + switch { + case r == '\n' || r == '\t': + return false + case r < 0x20 || r == 0x7f: + return true + case r >= 0x80 && r <= 0x9f: + return true + case r >= 0x202a && r <= 0x202e, r >= 0x2066 && r <= 0x2069, r == 0x200e, r == 0x200f, r == 0x061c: + return true + } + return false +} + +// escapeLen is how many bytes the escape sequence at the start of s takes, +// ESC included. An unterminated sequence takes the rest of the string: what +// follows an open OSC is the sequence's payload, not text. +func escapeLen(s string) int { + if len(s) < 2 { + return len(s) + } + switch s[1] { + case '[': // CSI: parameters and intermediates, then a final byte 0x40-0x7e + for i := 2; i < len(s); i++ { + if s[i] >= 0x40 && s[i] <= 0x7e { + return i + 1 + } + } + return len(s) + case ']', 'P', 'X', '^', '_': // OSC, DCS, SOS, PM, APC: up to BEL or ST + for i := 2; i < len(s); i++ { + if s[i] == 0x07 { + return i + 1 + } + if s[i] == 0x1b && i+1 < len(s) && s[i+1] == '\\' { + return i + 2 + } + } + return len(s) + default: // a two-byte sequence: ESC and one character + _, size := utf8.DecodeRuneInString(s[1:]) + return 1 + size + } +} + +// Truncate shortens s to at most n runes, marking the cut. +func Truncate(s string, n int) string { + if n < 1 { + return "" + } + if utf8.RuneCountInString(s) <= n { + return s + } + runes := []rune(s) + return string(runes[:n-1]) + "…" +} + +// Renderer prints a run's events as text. It remembers whether the last +// thing printed left the cursor mid-line, because assistant text arrives in +// chunks and the next event must not be glued onto the end of one. +type Renderer struct { + Out io.Writer + midLine bool +} + +func (r *Renderer) newline() { + if r.midLine { + fmt.Fprintln(r.Out) + r.midLine = false + } +} + +func (r *Renderer) line(format string, args ...any) { + r.newline() + fmt.Fprintf(r.Out, format+"\n", args...) +} + +// prefixed prints text one line at a time, each behind the same label, so a +// multi-line log entry stays attributable. +func (r *Renderer) prefixed(label, text string) { + r.newline() + text = strings.TrimRight(Sanitize(text), "\n") + for _, l := range strings.Split(text, "\n") { + fmt.Fprintf(r.Out, "%s │ %s\n", label, l) + } +} + +// Event prints one event. Event types this version does not know are skipped +// rather than dumped: a newer platform may send them, and the raw payload is +// still there for `nan runs logs --json`. +func (r *Renderer) Event(ev api.Event) { + switch ev.Type { + case "started": + var d struct { + Agent string `json:"agent"` + AgentVersion string `json:"agent_version"` + Model string `json:"model"` + Cwd string `json:"cwd"` + Branch string `json:"branch"` + } + _ = json.Unmarshal(ev.Data, &d) + parts := []string{"started " + SanitizeLine(strings.TrimSpace(d.Agent+" "+d.AgentVersion))} + if d.Model != "" { + parts = append(parts, "model "+SanitizeLine(d.Model)) + } + if d.Cwd != "" { + parts = append(parts, "in "+SanitizeLine(d.Cwd)) + } + if d.Branch != "" { + parts = append(parts, "on branch "+SanitizeLine(d.Branch)) + } + r.line("── %s", strings.Join(parts, " · ")) + case "log": + var d struct { + Stream string `json:"stream"` + Text string `json:"text"` + } + _ = json.Unmarshal(ev.Data, &d) + label := "log" + switch d.Stream { + case "agent_stdout": + label = "stdout" + case "agent_stderr": + label = "stderr" + case "nan-run": + label = "nan-run" + } + r.prefixed(label, d.Text) + case "message": + var d struct { + Text string `json:"text"` + Final bool `json:"final"` + } + _ = json.Unmarshal(ev.Data, &d) + text := Sanitize(d.Text) + if text != "" { + fmt.Fprint(r.Out, text) + r.midLine = !strings.HasSuffix(text, "\n") + } + if d.Final { + r.newline() + } + case "tool_call": + var d struct { + Name string `json:"name"` + ArgsPreview string `json:"args_preview"` + } + _ = json.Unmarshal(ev.Data, &d) + r.line("→ %s %s", SanitizeLine(d.Name), Truncate(SanitizeLine(d.ArgsPreview), 160)) + case "tool_result": + var d struct { + Name string `json:"name"` + IsError bool `json:"is_error"` + OutputPreview string `json:"output_preview"` + } + _ = json.Unmarshal(ev.Data, &d) + status := "ok" + if d.IsError { + status = "error" + } + preview := Truncate(SanitizeLine(strings.TrimSpace(d.OutputPreview)), 160) + r.line("← %s %s %s", SanitizeLine(d.Name), status, preview) + case "artifact": + var d struct { + Path string `json:"path"` + Bytes int64 `json:"bytes"` + } + _ = json.Unmarshal(ev.Data, &d) + r.line("── artifact %s (%d bytes)", SanitizeLine(d.Path), d.Bytes) + case "error": + var d struct { + Code string `json:"code"` + Message string `json:"message"` + } + _ = json.Unmarshal(ev.Data, &d) + r.line("── error %s: %s", SanitizeLine(d.Code), SanitizeLine(d.Message)) + case "truncated": + var d struct { + Reason string `json:"reason"` + } + _ = json.Unmarshal(ev.Data, &d) + r.line("── output truncated (%s): the run went on, its log did not", SanitizeLine(d.Reason)) + case "finished": + // The end of the stream carries the authoritative outcome; Finish + // prints it, so this would only say it twice. + r.newline() + } +} + +// Flush ends a line left open by assistant text. +func (r *Renderer) Flush() { r.newline() } + +// ExitCode is the process exit code for a run, as `nan run` documents it. +// A run that has not finished yet has none; ok is false. +func ExitCode(run *api.Run) (code int, ok bool) { + switch run.State { + case api.StateSucceeded: + return 0, true + case api.StateTimedOut: + return 2, true + case api.StateCancelled: + return 3, true + case api.StateFailed: + if run.ErrorCode != nil { + switch *run.ErrorCode { + case "config_error", "agent_not_installed", "no_inference_key": + return 4, true + } + } + return 1, true + } + return 0, false +} + +// Outcome is a one-line description of how a run ended, for stderr. +func Outcome(run *api.Run) string { + var b strings.Builder + b.WriteString("run " + SanitizeLine(run.ID) + " " + strings.ReplaceAll(SanitizeLine(run.State), "_", " ")) + if run.ErrorCode != nil && *run.ErrorCode != "" { + b.WriteString(" (" + SanitizeLine(*run.ErrorCode) + ")") + } + if run.Result != nil { + if run.Result.Branch != "" { + b.WriteString(" · branch " + SanitizeLine(run.Result.Branch)) + } + if run.Result.ChangedFiles > 0 { + fmt.Fprintf(&b, " · %d files changed", run.Result.ChangedFiles) + } + } + return b.String() +} + +// jsonUnsafe is what encoding/json leaves raw and a terminal may act on. +func jsonUnsafe(r rune) bool { return r == 0x7f || (r >= 0x80 && unsafeRune(r)) } + +// SafeJSON marshals v for --json output. encoding/json already escapes C0 +// controls, ESC included; it leaves C1 controls and the bidi overrides as raw +// UTF-8, which some terminals still act on when the output is read on one, +// and it passes invalid UTF-8 in a json.RawMessage through untouched. This +// writes all of those as \u escapes, which every JSON parser reads back. +func SafeJSON(v any, indent bool) ([]byte, error) { + var out []byte + var err error + if indent { + out, err = json.MarshalIndent(v, "", " ") + } else { + out, err = json.Marshal(v) + } + if err != nil { + return nil, err + } + if utf8.Valid(out) && !strings.ContainsFunc(string(out), jsonUnsafe) { + return out, nil + } + // Invalid UTF-8 can only be inside a string here - json.RawMessage + // carries a guest's bytes through verbatim - and a lone 0x9b is a + // C1 CSI to a terminal not in UTF-8 mode. It goes out as U+FFFD. + var b strings.Builder + for i := 0; i < len(out); { + r, size := utf8.DecodeRune(out[i:]) + i += size + switch { + case r == utf8.RuneError && size == 1: + b.WriteString(`\ufffd`) + case jsonUnsafe(r): + fmt.Fprintf(&b, `\u%04x`, r) + default: + b.WriteRune(r) + } + } + return []byte(b.String()), nil +} diff --git a/internal/runs/render_test.go b/internal/runs/render_test.go new file mode 100644 index 0000000..785aa39 --- /dev/null +++ b/internal/runs/render_test.go @@ -0,0 +1,165 @@ +package runs + +import ( + "bytes" + "encoding/json" + "strings" + "testing" + + "github.com/nxssie/nan-cli/internal/api" +) + +func TestSanitizeStripsEscapesAndControls(t *testing.T) { + for _, tc := range []struct{ name, in, want string }{ + {"plain text untouched", "hello\n\tworld", "hello\n\tworld"}, + {"colour", "\x1b[31mred\x1b[0m", "red"}, + {"cursor movement", "a\x1b[2Ab\x1b[Kc", "abc"}, + {"window title (OSC, BEL)", "x\x1b]0;pwned\x07y", "xy"}, + {"clipboard write (OSC 52, ST)", "x\x1b]52;c;ZXZpbA==\x1b\\y", "xy"}, + {"hyperlink", "\x1b]8;;http://evil\x1b\\click\x1b]8;;\x1b\\", "click"}, + {"DCS", "a\x1bPq#0;2;0;0;0\x1b\\b", "ab"}, + {"unterminated OSC eats the rest", "ok\x1b]0;title", "ok"}, + {"two-byte escape", "a\x1bcb", "ab"}, + {"lone ESC at the end", "a\x1b", "a"}, + {"carriage return rewrites the line", "safe\rEVIL", "safeEVIL"}, + {"backspace and bell", "ab\x08\x07c", "abc"}, + {"NUL and DEL", "a\x00b\x7fc", "abc"}, + {"C1 CSI", "a\u009b31mb", "a31mb"}, + {"bidi override", "a‮b⁦c", "abc"}, + {"invalid UTF-8", "a\xffb", "a�b"}, + {"non-ASCII text survives", "¡Hola, señor! 日本 🚀", "¡Hola, señor! 日本 🚀"}, + } { + if got := Sanitize(tc.in); got != tc.want { + t.Errorf("%s: Sanitize(%q) = %q, want %q", tc.name, tc.in, got, tc.want) + } + } +} + +func TestSanitizeLineKeepsOneLine(t *testing.T) { + if got := SanitizeLine("a\nb\tc\x1b[1m"); got != "a b c" { + t.Errorf("got %q", got) + } +} + +func ev(typ string, data any) api.Event { + raw, _ := json.Marshal(data) + return api.Event{V: 1, Type: typ, Data: raw} +} + +// Every field of every event type is guest text, so every one of them goes +// through the sanitiser - not just the assistant's message. +func TestRendererNeverPrintsAnEscape(t *testing.T) { + const bad = "\x1b]0;pwned\x07\x1b[2J\r" + var out bytes.Buffer + r := &Renderer{Out: &out} + for _, e := range []api.Event{ + ev("started", map[string]any{"agent": "pi" + bad, "agent_version": bad, "model": bad, "cwd": bad, "branch": bad}), + ev("log", map[string]any{"stream": "agent_stderr", "text": "line" + bad}), + ev("message", map[string]any{"role": "assistant", "text": "hi" + bad, "final": true}), + ev("tool_call", map[string]any{"id": "1", "name": "bash" + bad, "args_preview": bad}), + ev("tool_result", map[string]any{"id": "1", "name": bad, "is_error": true, "output_preview": bad}), + ev("artifact", map[string]any{"path": "artifacts/r.md" + bad, "bytes": 3}), + ev("error", map[string]any{"code": bad, "message": bad}), + ev("truncated", map[string]any{"reason": bad}), + } { + r.Event(e) + } + if strings.ContainsAny(out.String(), "\x1b\x07\r") { + t.Errorf("control bytes reached the terminal: %q", out.String()) + } + for _, want := range []string{"started pi", "stderr │ line", "hi\n", "→ bash", "← ", "error", "artifact artifacts/r.md (3 bytes)"} { + if !strings.Contains(out.String(), want) { + t.Errorf("missing %q in %q", want, out.String()) + } + } +} + +func TestRendererJoinsMessageChunksAndBreaksBeforeTheNextEvent(t *testing.T) { + var out bytes.Buffer + r := &Renderer{Out: &out} + r.Event(ev("message", map[string]any{"text": "Hel"})) + r.Event(ev("message", map[string]any{"text": "lo"})) + r.Event(ev("tool_call", map[string]any{"name": "read", "args_preview": "{}"})) + r.Event(ev("message", map[string]any{"text": "done", "final": true})) + r.Event(ev("log", map[string]any{"stream": "nan-run", "text": "a\nb\n"})) + want := "Hello\n→ read {}\ndone\nnan-run │ a\nnan-run │ b\n" + if out.String() != want { + t.Errorf("got %q\nwant %q", out.String(), want) + } +} + +func TestExitCodeFollowsTheTable(t *testing.T) { + code := func(s string) *string { return &s } + for _, tc := range []struct { + state string + err *string + want int + ok bool + }{ + {"succeeded", nil, 0, true}, + {"failed", code("agent_failed"), 1, true}, + {"failed", nil, 1, true}, + {"timed_out", code("timed_out"), 2, true}, + {"cancelled", code("cancelled"), 3, true}, + {"failed", code("config_error"), 4, true}, + {"failed", code("agent_not_installed"), 4, true}, + {"failed", code("no_inference_key"), 4, true}, + {"running", nil, 0, false}, + {"queued", nil, 0, false}, + } { + got, ok := ExitCode(&api.Run{State: tc.state, ErrorCode: tc.err}) + if got != tc.want || ok != tc.ok { + t.Errorf("%s/%v: got %d,%v want %d,%v", tc.state, tc.err, got, ok, tc.want, tc.ok) + } + } +} + +func TestSafeJSONEscapesWhatTerminalsActOn(t *testing.T) { + out, err := SafeJSON(map[string]string{"t": "a\x1b[31m\u009b‮b"}, false) + if err != nil { + t.Fatal(err) + } + if strings.ContainsAny(string(out), "\x1b\u009b‮") { + t.Errorf("raw control in JSON: %q", out) + } + var back map[string]string + if err := json.Unmarshal(out, &back); err != nil || back["t"] != "a\x1b[31m\u009b‮b" { + t.Errorf("did not round-trip: %v %q", err, back["t"]) + } +} + +func TestOutcomeIsOneSanitisedLine(t *testing.T) { + ec := "agent_failed" + got := Outcome(&api.Run{ID: "r1", State: "timed_out", ErrorCode: &ec, Result: &api.RunResult{Branch: "nan-run/x\x1b[2J", ChangedFiles: 2}}) + if got != "run r1 timed out (agent_failed) · branch nan-run/x · 2 files changed" { + t.Errorf("got %q", got) + } +} + +// encoding/json carries a json.RawMessage's bytes through verbatim, invalid +// UTF-8 included, and a lone 0x9b is a C1 CSI to an 8-bit terminal. +func TestSafeJSONNeutralisesInvalidUTF8InRawMessages(t *testing.T) { + var ev api.Event + if err := json.Unmarshal([]byte("{\"seq\":1,\"type\":\"log\",\"data\":{\"text\":\"a\x9b31mRED\"}}"), &ev); err != nil { + t.Fatal(err) + } + out, err := SafeJSON(ev, false) + if err != nil { + t.Fatal(err) + } + if bytes.IndexByte(out, 0x9b) >= 0 || !json.Valid(out) { + t.Errorf("out = %q", out) + } + if !strings.Contains(string(out), `a\ufffd31mRED`) { + t.Errorf("out = %q", out) + } + if got, _ := SafeJSON(map[string]string{"t": "a\x7fb"}, false); strings.ContainsRune(string(got), 0x7f) { + t.Errorf("DEL left raw: %q", got) + } +} + +func TestTruncate(t *testing.T) { + if Truncate("hello", 3) != "he…" || Truncate("hi", 3) != "hi" || Truncate("x", 0) != "" { + t.Error("Truncate") + } +} diff --git a/internal/tui/tui.go b/internal/tui/tui.go index 0f9beeb..fca689f 100644 --- a/internal/tui/tui.go +++ b/internal/tui/tui.go @@ -3549,7 +3549,7 @@ func (m model) renderSetup(l layout) string { // ── about renderer ─────────────────────────────────────────────────────────── -const Version = "0.1.23" +const Version = "0.1.24" func renderAbout(l layout) string { var b strings.Builder