From 2fd011f80ce3bf5ee7294d369ec84fef6ed82b90 Mon Sep 17 00:00:00 2001 From: Noel Date: Mon, 3 Aug 2026 16:07:33 +0200 Subject: [PATCH] feat(actions): add transition-only run watcher Signed-off-by: Noel --- cmd/actions/runs.go | 1 + cmd/actions/runs/logs.go | 88 ++++- cmd/actions/runs/logs_test.go | 142 +++++++ cmd/actions/runs/watch.go | 111 ++++++ cmd/actions/runs/watch_test.go | 45 +++ docs/CLI.md | 24 ++ modules/print/actions_run_watch.go | 242 ++++++++++++ modules/print/actions_run_watch_test.go | 86 +++++ modules/task/actions_run_watch.go | 472 ++++++++++++++++++++++++ modules/task/actions_run_watch_test.go | 259 +++++++++++++ 10 files changed, 1461 insertions(+), 9 deletions(-) create mode 100644 cmd/actions/runs/logs_test.go create mode 100644 cmd/actions/runs/watch.go create mode 100644 cmd/actions/runs/watch_test.go create mode 100644 modules/print/actions_run_watch.go create mode 100644 modules/print/actions_run_watch_test.go create mode 100644 modules/task/actions_run_watch.go create mode 100644 modules/task/actions_run_watch_test.go diff --git a/cmd/actions/runs.go b/cmd/actions/runs.go index 0b34d690..da581b8c 100644 --- a/cmd/actions/runs.go +++ b/cmd/actions/runs.go @@ -21,6 +21,7 @@ var CmdActionsRuns = cli.Command{ Commands: []*cli.Command{ &runs.CmdRunsList, &runs.CmdRunsView, + &runs.CmdRunsWatch, &runs.CmdRunsDelete, &runs.CmdRunsLogs, }, diff --git a/cmd/actions/runs/logs.go b/cmd/actions/runs/logs.go index f6b4735b..e7bcbf3d 100644 --- a/cmd/actions/runs/logs.go +++ b/cmd/actions/runs/logs.go @@ -4,6 +4,7 @@ package runs import ( + "bytes" stdctx "context" "fmt" "strconv" @@ -34,6 +35,14 @@ var CmdRunsLogs = cli.Command{ Aliases: []string{"f"}, Usage: "follow log output (like tail -f), requires job to be in progress", }, + &cli.BoolFlag{ + Name: "failed", + Usage: "show logs only for failed jobs", + }, + &cli.IntFlag{ + Name: "tail", + Usage: "show only the last number of lines (0 shows all lines)", + }, }, flags.AllDefaultFlags...), } @@ -41,7 +50,10 @@ func runRunsLogs(ctx stdctx.Context, cmd *cli.Command) error { if cmd.Args().Len() == 0 { return fmt.Errorf("run ID is required") } + return runRunsLogsForID(ctx, cmd, cmd.Args().First()) +} +func runRunsLogsForID(ctx stdctx.Context, cmd *cli.Command, runIDStr string) error { c, err := context.InitCommand(cmd) if err != nil { return err @@ -51,7 +63,6 @@ func runRunsLogs(ctx stdctx.Context, cmd *cli.Command) error { } client := c.Login.Client() - runIDStr := cmd.Args().First() runID, err := strconv.ParseInt(runIDStr, 10, 64) if err != nil { return fmt.Errorf("invalid run ID: %s", runIDStr) @@ -59,10 +70,21 @@ func runRunsLogs(ctx stdctx.Context, cmd *cli.Command) error { // Check if follow mode is enabled follow := cmd.Bool("follow") + failedOnly := cmd.Bool("failed") + tailLines := cmd.Int("tail") + if tailLines < 0 { + return fmt.Errorf("--tail cannot be negative") + } + if follow && tailLines > 0 { + return fmt.Errorf("--tail cannot be used with --follow") + } // If specific job ID provided, fetch only that job's logs jobIDStr := cmd.String("job") if jobIDStr != "" { + if failedOnly { + return fmt.Errorf("--failed cannot be used with --job") + } jobID, err := strconv.ParseInt(jobIDStr, 10, 64) if err != nil { return fmt.Errorf("invalid job ID: %s", jobIDStr) @@ -78,7 +100,11 @@ func runRunsLogs(ctx stdctx.Context, cmd *cli.Command) error { } fmt.Printf("Logs for job %d:\n", jobID) - fmt.Printf("---\n%s\n", string(logs)) + trimmedLogs := tailLogLines(logs, tailLines) + fmt.Printf("---\n%s", string(trimmedLogs)) + if !bytes.HasSuffix(trimmedLogs, []byte("\n")) { + fmt.Println() + } return nil } @@ -90,29 +116,40 @@ func runRunsLogs(ctx stdctx.Context, cmd *cli.Command) error { return fmt.Errorf("failed to get jobs: %w", err) } - if len(jobs.Jobs) == 0 { + if jobs == nil || len(jobs.Jobs) == 0 { fmt.Printf("No jobs found for run %d\n", runID) return nil } + selectedJobs := jobs.Jobs + if failedOnly { + selectedJobs = filterFailedJobs(jobs.Jobs) + if len(selectedJobs) == 0 { + fmt.Printf("No failed jobs found for run %d\n", runID) + return nil + } + } // If following and multiple jobs, require --job flag - if follow && len(jobs.Jobs) > 1 { - return fmt.Errorf("--follow requires --job when run has multiple jobs (found %d jobs)", len(jobs.Jobs)) + if follow && len(selectedJobs) > 1 { + return fmt.Errorf("--follow requires --job when run has multiple jobs (found %d jobs)", len(selectedJobs)) } // If following with single job, follow it - if follow && len(jobs.Jobs) == 1 { - return followJobLogs(ctx, client, c, jobs.Jobs[0].ID, jobs.Jobs[0].Name) + if follow && len(selectedJobs) == 1 { + return followJobLogs(ctx, client, c, selectedJobs[0].ID, selectedJobs[0].Name) } // Fetch logs for each job - for i, job := range jobs.Jobs { + for i, job := range selectedJobs { if i > 0 { fmt.Println() } fmt.Printf("Job: %s (ID: %d)\n", job.Name, job.ID) fmt.Printf("Status: %s\n", job.Status) + if job.Conclusion != "" { + fmt.Printf("Conclusion: %s\n", job.Conclusion) + } fmt.Println("---") logs, _, err := client.Actions.GetRepoRunJobLogs(ctx, c.Owner, c.Repo, job.ID) @@ -121,12 +158,45 @@ func runRunsLogs(ctx stdctx.Context, cmd *cli.Command) error { continue } - fmt.Println(string(logs)) + trimmedLogs := tailLogLines(logs, tailLines) + fmt.Print(string(trimmedLogs)) + if !bytes.HasSuffix(trimmedLogs, []byte("\n")) { + fmt.Println() + } } return nil } +func filterFailedJobs(jobs []*gitea.ActionWorkflowJob) []*gitea.ActionWorkflowJob { + failed := make([]*gitea.ActionWorkflowJob, 0) + for _, job := range jobs { + if job.Conclusion == "failure" || (job.Conclusion == "" && job.Status == "failure") { + failed = append(failed, job) + } + } + return failed +} + +func tailLogLines(logs []byte, lineCount int) []byte { + if lineCount <= 0 || len(logs) == 0 { + return logs + } + hasTrailingNewline := bytes.HasSuffix(logs, []byte("\n")) + lines := bytes.Split(logs, []byte("\n")) + if hasTrailingNewline { + lines = lines[:len(lines)-1] + } + if len(lines) > lineCount { + lines = lines[len(lines)-lineCount:] + } + result := bytes.Join(lines, []byte("\n")) + if hasTrailingNewline { + result = append(result, '\n') + } + return result +} + // followJobLogs continuously fetches and displays logs for a running job func followJobLogs(requestCtx stdctx.Context, client *gitea.Client, c *context.TeaContext, jobID int64, jobName string) error { var lastLogLength int diff --git a/cmd/actions/runs/logs_test.go b/cmd/actions/runs/logs_test.go new file mode 100644 index 00000000..c2325e9b --- /dev/null +++ b/cmd/actions/runs/logs_test.go @@ -0,0 +1,142 @@ +// Copyright 2026 The Gitea Authors. All rights reserved. +// SPDX-License-Identifier: MIT + +package runs + +import ( + "bytes" + "io" + "net/http" + "net/http/httptest" + "os" + "sync/atomic" + "testing" + + gitea "gitea.dev/sdk" + + "gitea.dev/tea/modules/config" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "github.com/urfave/cli/v3" +) + +func TestFilterFailedJobs(t *testing.T) { + jobs := []*gitea.ActionWorkflowJob{ + {ID: 1, Status: "completed", Conclusion: "success"}, + {ID: 2, Status: "completed", Conclusion: "failure"}, + {ID: 3, Status: "failure"}, + {ID: 4, Status: "completed", Conclusion: "canceled"}, + } + failed := filterFailedJobs(jobs) + require.Len(t, failed, 2) + assert.Equal(t, []int64{2, 3}, []int64{failed[0].ID, failed[1].ID}) +} + +func TestTailLogLines(t *testing.T) { + tests := []struct { + name string + logs string + lines int + expected string + }{ + {name: "last two with newline", logs: "one\ntwo\nthree\n", lines: 2, expected: "two\nthree\n"}, + {name: "last one without newline", logs: "one\ntwo\nthree", lines: 1, expected: "three"}, + {name: "more than available", logs: "one\ntwo\n", lines: 10, expected: "one\ntwo\n"}, + {name: "zero means all", logs: "one\ntwo\n", lines: 0, expected: "one\ntwo\n"}, + {name: "empty", logs: "", lines: 2, expected: ""}, + } + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + assert.Equal(t, test.expected, string(tailLogLines([]byte(test.logs), test.lines))) + }) + } +} + +func TestRunsLogsFlags(t *testing.T) { + commandFlags := make(map[string]bool) + for _, flag := range CmdRunsLogs.Flags { + commandFlags[flag.Names()[0]] = true + } + assert.True(t, commandFlags["failed"]) + assert.True(t, commandFlags["tail"]) +} + +func TestRunsLogsFailedTailFetchesOnlyFailedJobs(t *testing.T) { + var successfulLogRequests atomic.Int32 + var failedLogRequests atomic.Int32 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch r.URL.Path { + case "/api/v1/version": + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(`{"version":"1.26.0"}`)) + case "/api/v1/repos/gitea/tea/actions/runs/42/jobs": + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(`{"total_count":2,"jobs":[{"id":1,"name":"success","status":"completed","conclusion":"success"},{"id":2,"name":"failed","status":"completed","conclusion":"failure"}]}`)) + case "/api/v1/repos/gitea/tea/actions/jobs/1/logs": + successfulLogRequests.Add(1) + _, _ = w.Write([]byte("success log\n")) + case "/api/v1/repos/gitea/tea/actions/jobs/2/logs": + failedLogRequests.Add(1) + _, _ = w.Write([]byte("one\ntwo\nthree\n")) + default: + t.Errorf("unexpected request path %s", r.URL.Path) + w.WriteHeader(http.StatusNotFound) + } + })) + defer server.Close() + + config.SetConfigForTesting(config.LocalConfig{Logins: []config.Login{{ + Name: "test-logs", URL: server.URL, Token: "token", User: "tester", Default: true, + }}}) + t.Cleanup(func() { config.SetConfigForTesting(config.LocalConfig{}) }) + + cmd := &cli.Command{ + Name: "logs", + Flags: []cli.Flag{ + &cli.StringFlag{Name: "job"}, + &cli.BoolFlag{Name: "follow"}, + &cli.BoolFlag{Name: "failed"}, + &cli.IntFlag{Name: "tail"}, + &cli.IntFlag{Name: "page", Value: 1}, + &cli.IntFlag{Name: "limit", Value: 30}, + &cli.StringFlag{Name: "repo"}, + &cli.StringFlag{Name: "remote"}, + &cli.StringFlag{Name: "login"}, + &cli.StringFlag{Name: "output"}, + }, + } + require.NoError(t, cmd.Set("login", "test-logs")) + require.NoError(t, cmd.Set("repo", "gitea/tea")) + require.NoError(t, cmd.Set("failed", "true")) + require.NoError(t, cmd.Set("tail", "2")) + + output := captureRunsStdout(t, func() { + require.NoError(t, runRunsLogsForID(t.Context(), cmd, "42")) + }) + assert.Zero(t, successfulLogRequests.Load()) + assert.EqualValues(t, 1, failedLogRequests.Load()) + assert.NotContains(t, output, "success log") + assert.NotContains(t, output, "one\n") + assert.Contains(t, output, "two\nthree\n") +} + +func captureRunsStdout(t *testing.T, fn func()) string { + t.Helper() + oldStdout := os.Stdout + reader, writer, err := os.Pipe() + require.NoError(t, err) + os.Stdout = writer + defer func() { os.Stdout = oldStdout }() + + done := make(chan string, 1) + go func() { + var output bytes.Buffer + _, _ = io.Copy(&output, reader) + done <- output.String() + }() + fn() + require.NoError(t, writer.Close()) + output := <-done + require.NoError(t, reader.Close()) + return output +} diff --git a/cmd/actions/runs/watch.go b/cmd/actions/runs/watch.go new file mode 100644 index 00000000..d14f04e8 --- /dev/null +++ b/cmd/actions/runs/watch.go @@ -0,0 +1,111 @@ +// Copyright 2026 The Gitea Authors. All rights reserved. +// SPDX-License-Identifier: MIT + +package runs + +import ( + stdctx "context" + "fmt" + "strconv" + "time" + + "gitea.dev/tea/cmd/flags" + "gitea.dev/tea/modules/context" + "gitea.dev/tea/modules/print" + "gitea.dev/tea/modules/task" + "github.com/urfave/cli/v3" +) + +// CmdRunsWatch watches workflow runs and reports state transitions. +var CmdRunsWatch = cli.Command{ + Name: "watch", + Usage: "Watch workflow runs for state changes", + ArgsUsage: " [...]", + Description: `Watch one or more workflow runs without streaming logs or mutating them. +The default output is concise human-readable text. NDJSON output contains only +versioned bindings, state transitions, stall/recovery events, and a summary. + +Exit status 0 means all runs succeeded, 1 means an observation error occurred, +2 means a run was unsuccessful, 3 means --expect-head did not match, and 4 +means the watch timed out.`, + Action: runRunsWatch, + Flags: append([]cli.Flag{ + &cli.StringFlag{ + Name: "expect-head", + Usage: "fail if a run is not bound to this head SHA", + }, + &cli.DurationFlag{ + Name: "interval", + Usage: "polling interval", + Value: 30 * time.Second, + }, + &cli.DurationFlag{ + Name: "stall-after", + Usage: "report a run after this long without a state change (0 disables)", + Value: 30 * time.Minute, + }, + &cli.DurationFlag{ + Name: "timeout", + Usage: "stop watching after this duration", + Value: 3 * time.Hour, + }, + &cli.StringFlag{ + Name: "output", + Usage: "output format (text, ndjson)", + Value: "text", + }, + }, flags.LoginRepoFlags...), +} + +func runRunsWatch(ctx stdctx.Context, cmd *cli.Command) error { + runIDs, err := parseWatchRunIDs(cmd.Args().Slice()) + if err != nil { + return cli.Exit(err, task.ActionRunWatchExitError) + } + renderer, err := print.NewActionRunWatchRenderer(cmd.String("output"), cmd.Writer) + if err != nil { + return cli.Exit(err, task.ActionRunWatchExitError) + } + + c, err := context.InitCommand(cmd) + if err != nil { + return cli.Exit(err, task.ActionRunWatchExitError) + } + if err := c.Ensure(context.CtxRequirement{RemoteRepo: true}); err != nil { + return cli.Exit(err, task.ActionRunWatchExitError) + } + result, err := task.WatchActionRuns(ctx, c.Login.Client().Actions, c.Owner, c.Repo, task.ActionRunWatchOptions{ + RunIDs: runIDs, + ExpectedHead: cmd.String("expect-head"), + Interval: cmd.Duration("interval"), + StallAfter: cmd.Duration("stall-after"), + Timeout: cmd.Duration("timeout"), + }, renderer.Render) + if err != nil { + return cli.Exit(err, task.ActionRunWatchExitError) + } + if result.ExitCode != task.ActionRunWatchExitSuccess { + return cli.Exit(result.Message, result.ExitCode) + } + return nil +} + +func parseWatchRunIDs(args []string) ([]int64, error) { + if len(args) == 0 { + return nil, fmt.Errorf("at least one run ID is required") + } + seen := make(map[int64]struct{}, len(args)) + runIDs := make([]int64, 0, len(args)) + for _, arg := range args { + runID, err := strconv.ParseInt(arg, 10, 64) + if err != nil || runID <= 0 { + return nil, fmt.Errorf("invalid run ID: %s", arg) + } + if _, exists := seen[runID]; exists { + return nil, fmt.Errorf("duplicate run ID: %d", runID) + } + seen[runID] = struct{}{} + runIDs = append(runIDs, runID) + } + return runIDs, nil +} diff --git a/cmd/actions/runs/watch_test.go b/cmd/actions/runs/watch_test.go new file mode 100644 index 00000000..9b74fcf1 --- /dev/null +++ b/cmd/actions/runs/watch_test.go @@ -0,0 +1,45 @@ +// Copyright 2026 The Gitea Authors. All rights reserved. +// SPDX-License-Identifier: MIT + +package runs + +import ( + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "github.com/urfave/cli/v3" +) + +func TestParseWatchRunIDs(t *testing.T) { + runIDs, err := parseWatchRunIDs([]string{"12", "34"}) + require.NoError(t, err) + assert.Equal(t, []int64{12, 34}, runIDs) + + _, err = parseWatchRunIDs(nil) + require.ErrorContains(t, err, "at least one") + _, err = parseWatchRunIDs([]string{"invalid"}) + require.ErrorContains(t, err, "invalid") + _, err = parseWatchRunIDs([]string{"12", "12"}) + require.ErrorContains(t, err, "duplicate") +} + +func TestRunsWatchDefaults(t *testing.T) { + var output string + durations := make(map[string]time.Duration) + for _, flag := range CmdRunsWatch.Flags { + switch typedFlag := flag.(type) { + case *cli.StringFlag: + if typedFlag.Name == "output" { + output = typedFlag.Value + } + case *cli.DurationFlag: + durations[typedFlag.Name] = typedFlag.Value + } + } + assert.Equal(t, "text", output) + assert.Equal(t, 30*time.Second, durations["interval"]) + assert.Equal(t, 30*time.Minute, durations["stall-after"]) + assert.Equal(t, 3*time.Hour, durations["timeout"]) +} diff --git a/docs/CLI.md b/docs/CLI.md index 491759c7..c1f6a6a2 100644 --- a/docs/CLI.md +++ b/docs/CLI.md @@ -1609,6 +1609,26 @@ View workflow run details **--repo, -r**="": Override local repository path or gitea repository slug to interact with. Optional +#### watch + +Watch workflow runs for state changes + +**--expect-head**="": fail if a run is not bound to this head SHA + +**--interval**="": polling interval (default: 30s) + +**--login, -l**="": Use a different Gitea Login. Optional + +**--output**="": output format (text, ndjson) (default: "text") + +**--remote, -R**="": Discover Gitea login from remote. Optional + +**--repo, -r**="": Override local repository path or gitea repository slug to interact with. Optional + +**--stall-after**="": report a run after this long without a state change (0 disables) (default: 30m0s) + +**--timeout**="": stop watching after this duration (default: 3h0m0s) + #### delete, remove, rm, cancel Delete or cancel a workflow run @@ -1627,6 +1647,8 @@ Delete or cancel a workflow run View workflow run logs +**--failed**: show logs only for failed jobs + **--follow, -f**: follow log output (like tail -f), requires job to be in progress **--job**="": specific job ID to view logs for (if omitted, shows all jobs) @@ -1639,6 +1661,8 @@ View workflow run logs **--repo, -r**="": Override local repository path or gitea repository slug to interact with. Optional +**--tail**="": show only the last number of lines (0 shows all lines) (default: 0) + ### workflows, workflow Manage repository workflows diff --git a/modules/print/actions_run_watch.go b/modules/print/actions_run_watch.go new file mode 100644 index 00000000..2cba4715 --- /dev/null +++ b/modules/print/actions_run_watch.go @@ -0,0 +1,242 @@ +// Copyright 2026 The Gitea Authors. All rights reserved. +// SPDX-License-Identifier: MIT + +package print + +import ( + "encoding/json" + "fmt" + "io" + "os" + "strings" + "time" +) + +// ActionRunWatchSchemaVersion is the current NDJSON event schema version. +const ActionRunWatchSchemaVersion = 1 + +// ActionRunWatchRun is the concise workflow-run state rendered by watch output. +type ActionRunWatchRun struct { + RunID int64 + HeadSHA string + Workflow string + Path string + Status string + Conclusion string +} + +// ActionRunWatchJob is the concise job state rendered by watch output. +type ActionRunWatchJob struct { + JobID int64 + Name string + Status string + Conclusion string +} + +// ActionRunWatchChange describes one run or job state transition. +type ActionRunWatchChange struct { + Scope string + JobID int64 + Name string + Field string + Before string + After string +} + +// ActionRunWatchEvent is a binding, transition, stall, recovery, or summary. +type ActionRunWatchEvent struct { + Type string + ObservedAt time.Time + RunID int64 + ExpectedHead string + HeadMatches bool + Run *ActionRunWatchRun + Jobs []ActionRunWatchJob + Changes []ActionRunWatchChange + UnchangedFor time.Duration + Outcome string + Runs []ActionRunWatchRun +} + +// ActionRunWatchRenderer renders workflow-run watch events. +type ActionRunWatchRenderer interface { + Render(ActionRunWatchEvent) error +} + +type actionRunWatchRenderer struct { + output string + writer io.Writer + encoder *json.Encoder +} + +// NewActionRunWatchRenderer returns a concise text or versioned NDJSON renderer. +func NewActionRunWatchRenderer(output string, writer io.Writer) (ActionRunWatchRenderer, error) { + if writer == nil { + writer = os.Stdout + } + renderer := &actionRunWatchRenderer{output: output, writer: writer} + switch output { + case "", "text": + return renderer, nil + case "ndjson": + renderer.encoder = json.NewEncoder(writer) + return renderer, nil + default: + return nil, fmt.Errorf("unsupported watch output %q: available outputs are text and ndjson", output) + } +} + +func (r *actionRunWatchRenderer) Render(event ActionRunWatchEvent) error { + if r.output == "ndjson" { + return r.encoder.Encode(newActionRunWatchDocument(event)) + } + return r.renderText(event) +} + +func (r *actionRunWatchRenderer) renderText(event ActionRunWatchEvent) error { + switch event.Type { + case "binding": + match := "" + if event.ExpectedHead != "" && !event.HeadMatches { + match = " (head mismatch)" + } + _, err := fmt.Fprintf( + r.writer, + "run %d bound %s %s: %s%s (jobs: %d)\n", + event.RunID, + event.Run.HeadSHA, + event.Run.Workflow, + actionRunWatchState(*event.Run), + match, + len(event.Jobs), + ) + return err + case "transition": + changes := make([]string, 0, len(event.Changes)) + for _, change := range event.Changes { + target := change.Scope + if change.Scope == "job" { + target = fmt.Sprintf("job %s", change.Name) + } + changes = append(changes, fmt.Sprintf("%s %s %s -> %s", target, change.Field, emptyWatchValue(change.Before), emptyWatchValue(change.After))) + } + _, err := fmt.Fprintf(r.writer, "run %d changed: %s\n", event.RunID, strings.Join(changes, "; ")) + return err + case "stall": + _, err := fmt.Fprintf(r.writer, "run %d stalled: no state change for %s\n", event.RunID, event.UnchangedFor.Round(time.Second)) + return err + case "recovery": + _, err := fmt.Fprintf(r.writer, "run %d recovered\n", event.RunID) + return err + case "summary": + states := make([]string, 0, len(event.Runs)) + for _, run := range event.Runs { + states = append(states, fmt.Sprintf("%d=%s", run.RunID, actionRunWatchState(run))) + } + _, err := fmt.Fprintf(r.writer, "summary %s: %s\n", event.Outcome, strings.Join(states, ", ")) + return err + default: + return fmt.Errorf("unknown workflow run watch event %q", event.Type) + } +} + +type actionRunWatchDocument struct { + SchemaVersion int `json:"schema_version"` + Type string `json:"type"` + ObservedAt string `json:"observed_at"` + RunID int64 `json:"run_id,omitempty"` + ExpectedHead string `json:"expected_head,omitempty"` + HeadMatches *bool `json:"head_matches,omitempty"` + Run *actionRunWatchRun `json:"run,omitempty"` + Jobs []actionRunWatchJob `json:"jobs,omitempty"` + Changes []actionRunWatchChange `json:"changes,omitempty"` + UnchangedFor string `json:"unchanged_for,omitempty"` + Outcome string `json:"outcome,omitempty"` + Runs []actionRunWatchRun `json:"runs,omitempty"` +} + +type actionRunWatchRun struct { + RunID int64 `json:"run_id"` + HeadSHA string `json:"head_sha"` + Workflow string `json:"workflow"` + Path string `json:"path"` + Status string `json:"status"` + Conclusion string `json:"conclusion"` +} + +type actionRunWatchJob struct { + JobID int64 `json:"job_id"` + Name string `json:"name"` + Status string `json:"status"` + Conclusion string `json:"conclusion"` +} + +type actionRunWatchChange struct { + Scope string `json:"scope"` + JobID int64 `json:"job_id,omitempty"` + Name string `json:"name,omitempty"` + Field string `json:"field"` + Before string `json:"before"` + After string `json:"after"` +} + +func newActionRunWatchDocument(event ActionRunWatchEvent) actionRunWatchDocument { + document := actionRunWatchDocument{ + SchemaVersion: ActionRunWatchSchemaVersion, + Type: event.Type, + ObservedAt: event.ObservedAt.UTC().Format(time.RFC3339), + RunID: event.RunID, + ExpectedHead: event.ExpectedHead, + UnchangedFor: formatActionRunWatchDuration(event.UnchangedFor), + Outcome: event.Outcome, + } + if event.Type == "binding" { + headMatches := event.HeadMatches + document.HeadMatches = &headMatches + } + if event.Run != nil { + run := newActionRunWatchRun(*event.Run) + document.Run = &run + } + for _, job := range event.Jobs { + document.Jobs = append(document.Jobs, actionRunWatchJob{ + JobID: job.JobID, Name: job.Name, Status: job.Status, Conclusion: job.Conclusion, + }) + } + for _, change := range event.Changes { + document.Changes = append(document.Changes, actionRunWatchChange{ + Scope: change.Scope, JobID: change.JobID, Name: change.Name, Field: change.Field, Before: change.Before, After: change.After, + }) + } + for _, run := range event.Runs { + document.Runs = append(document.Runs, newActionRunWatchRun(run)) + } + return document +} + +func newActionRunWatchRun(run ActionRunWatchRun) actionRunWatchRun { + return actionRunWatchRun{ + RunID: run.RunID, HeadSHA: run.HeadSHA, Workflow: run.Workflow, Path: run.Path, Status: run.Status, Conclusion: run.Conclusion, + } +} + +func formatActionRunWatchDuration(duration time.Duration) string { + if duration <= 0 { + return "" + } + return duration.Round(time.Second).String() +} + +func actionRunWatchState(run ActionRunWatchRun) string { + if run.Conclusion != "" { + return run.Conclusion + } + return run.Status +} + +func emptyWatchValue(value string) string { + if value == "" { + return "-" + } + return value +} diff --git a/modules/print/actions_run_watch_test.go b/modules/print/actions_run_watch_test.go new file mode 100644 index 00000000..7cd731a8 --- /dev/null +++ b/modules/print/actions_run_watch_test.go @@ -0,0 +1,86 @@ +// Copyright 2026 The Gitea Authors. All rights reserved. +// SPDX-License-Identifier: MIT + +package print + +import ( + "bytes" + "encoding/json" + "io" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestActionRunWatchNDJSONSchema(t *testing.T) { + buf := &bytes.Buffer{} + renderer, err := NewActionRunWatchRenderer("ndjson", buf) + require.NoError(t, err) + now := time.Date(2026, time.August, 3, 10, 0, 0, 0, time.UTC) + run := ActionRunWatchRun{ + RunID: 42, HeadSHA: "abc123", Workflow: "CI", Path: ".gitea/workflows/ci.yml", Status: "queued", + } + events := []ActionRunWatchEvent{ + { + Type: "binding", ObservedAt: now, RunID: 42, ExpectedHead: "abc123", HeadMatches: true, Run: &run, + Jobs: []ActionRunWatchJob{{JobID: 7, Name: "test", Status: "queued"}}, + }, + { + Type: "transition", ObservedAt: now.Add(time.Minute), RunID: 42, + Changes: []ActionRunWatchChange{{Scope: "run", Field: "status", Before: "queued", After: "in_progress"}}, + }, + {Type: "stall", ObservedAt: now.Add(31 * time.Minute), RunID: 42, UnchangedFor: 30 * time.Minute}, + {Type: "recovery", ObservedAt: now.Add(32 * time.Minute), RunID: 42}, + {Type: "summary", ObservedAt: now.Add(33 * time.Minute), Outcome: "success", Runs: []ActionRunWatchRun{run}}, + } + for _, event := range events { + require.NoError(t, renderer.Render(event)) + } + + decoder := json.NewDecoder(bytes.NewReader(buf.Bytes())) + for index, expectedType := range []string{"binding", "transition", "stall", "recovery", "summary"} { + var document map[string]any + require.NoError(t, decoder.Decode(&document)) + assert.Equal(t, float64(ActionRunWatchSchemaVersion), document["schema_version"], "record %d", index) + assert.Equal(t, expectedType, document["type"], "record %d", index) + if index == 0 { + assert.Equal(t, "2026-08-03T10:00:00Z", document["observed_at"], "binding timestamp is stable") + } else { + assert.NotEmpty(t, document["observed_at"]) + } + } + var extra map[string]any + assert.ErrorIs(t, decoder.Decode(&extra), io.EOF) +} + +func TestActionRunWatchTextRenderer(t *testing.T) { + buf := &bytes.Buffer{} + renderer, err := NewActionRunWatchRenderer("text", buf) + require.NoError(t, err) + run := ActionRunWatchRun{RunID: 42, HeadSHA: "abc123", Workflow: "CI", Status: "queued"} + events := []ActionRunWatchEvent{ + {Type: "binding", RunID: 42, Run: &run, HeadMatches: true, Jobs: []ActionRunWatchJob{{JobID: 7}}}, + { + Type: "transition", RunID: 42, + Changes: []ActionRunWatchChange{{Scope: "job", Name: "test", Field: "status", Before: "queued", After: "in_progress"}}, + }, + {Type: "stall", RunID: 42, UnchangedFor: 30 * time.Minute}, + {Type: "recovery", RunID: 42}, + {Type: "summary", Outcome: "success", Runs: []ActionRunWatchRun{{RunID: 42, Conclusion: "success"}}}, + } + for _, event := range events { + require.NoError(t, renderer.Render(event)) + } + assert.Equal(t, "run 42 bound abc123 CI: queued (jobs: 1)\n"+ + "run 42 changed: job test status queued -> in_progress\n"+ + "run 42 stalled: no state change for 30m0s\n"+ + "run 42 recovered\n"+ + "summary success: 42=success\n", buf.String()) +} + +func TestActionRunWatchRendererRejectsUnknownOutput(t *testing.T) { + _, err := NewActionRunWatchRenderer("yaml", &bytes.Buffer{}) + require.ErrorContains(t, err, "available outputs are text and ndjson") +} diff --git a/modules/task/actions_run_watch.go b/modules/task/actions_run_watch.go new file mode 100644 index 00000000..e5ba9c9d --- /dev/null +++ b/modules/task/actions_run_watch.go @@ -0,0 +1,472 @@ +// Copyright 2026 The Gitea Authors. All rights reserved. +// SPDX-License-Identifier: MIT + +package task + +import ( + stdctx "context" + "errors" + "fmt" + "sort" + "strings" + "time" + + gitea "gitea.dev/sdk" + + "gitea.dev/tea/modules/print" +) + +const ( + // ActionRunWatchExitSuccess means every observed run succeeded. + ActionRunWatchExitSuccess = 0 + // ActionRunWatchExitError means observation failed. + ActionRunWatchExitError = 1 + // ActionRunWatchExitUnsuccessful means at least one run was unsuccessful. + ActionRunWatchExitUnsuccessful = 2 + // ActionRunWatchExitHeadMismatch means an expected head did not match. + ActionRunWatchExitHeadMismatch = 3 + // ActionRunWatchExitTimeout means observation exceeded its timeout. + ActionRunWatchExitTimeout = 4 +) + +// ActionRunWatchClient is the read-only Actions API used by the observer. +type ActionRunWatchClient interface { + GetRepoRun(stdctx.Context, string, string, int64) (*gitea.ActionWorkflowRun, *gitea.Response, error) + ListRepoJobsByRun(stdctx.Context, string, string, int64, gitea.ListRepoActionsJobsOptions) (*gitea.ActionWorkflowJobsResponse, *gitea.Response, error) +} + +// ActionRunWatchOptions configures workflow-run observation. +type ActionRunWatchOptions struct { + RunIDs []int64 + ExpectedHead string + Interval time.Duration + StallAfter time.Duration + Timeout time.Duration +} + +// ActionRunWatchResult describes the observer process exit. +type ActionRunWatchResult struct { + ExitCode int + Message string +} + +type ( + // ActionRunWatchRun is the concise observed run state. + ActionRunWatchRun = print.ActionRunWatchRun + // ActionRunWatchJob is the concise observed job state. + ActionRunWatchJob = print.ActionRunWatchJob + // ActionRunWatchChange describes one observed transition. + ActionRunWatchChange = print.ActionRunWatchChange + // ActionRunWatchEvent is emitted only for meaningful observer events. + ActionRunWatchEvent = print.ActionRunWatchEvent +) + +type actionRunWatchTrackedRun struct { + run ActionRunWatchRun + jobs map[int64]ActionRunWatchJob + lastChange time.Time + nextJobProbe time.Time + stalled bool +} + +type actionRunWatchSession struct { + client ActionRunWatchClient + owner string + repo string + opts ActionRunWatchOptions + emit func(ActionRunWatchEvent) error + runs map[int64]*actionRunWatchTrackedRun +} + +// WatchActionRuns observes runs until all are terminal, timeout, or cancellation. +func WatchActionRuns( + ctx stdctx.Context, + client ActionRunWatchClient, + owner, repo string, + opts ActionRunWatchOptions, + emit func(ActionRunWatchEvent) error, +) (ActionRunWatchResult, error) { + if err := validateActionRunWatchOptions(opts); err != nil { + return ActionRunWatchResult{ExitCode: ActionRunWatchExitError}, err + } + if err := ctx.Err(); err != nil { + return ActionRunWatchResult{ExitCode: ActionRunWatchExitError}, err + } + + watchCtx, cancel := stdctx.WithTimeout(ctx, opts.Timeout) + defer cancel() + + session := newActionRunWatchSession(client, owner, repo, opts, emit) + done, result, err := session.initialize(watchCtx, time.Now().UTC()) + if err != nil || done { + return result, err + } + + ticker := time.NewTicker(opts.Interval) + defer ticker.Stop() + for { + select { + case <-watchCtx.Done(): + now := time.Now().UTC() + if errors.Is(watchCtx.Err(), stdctx.DeadlineExceeded) { + if err := session.emitSummary("timeout", now); err != nil { + return ActionRunWatchResult{ExitCode: ActionRunWatchExitError}, err + } + return ActionRunWatchResult{ + ExitCode: ActionRunWatchExitTimeout, + Message: "workflow run watch timed out", + }, nil + } + if err := session.emitSummary("canceled", now); err != nil { + return ActionRunWatchResult{ExitCode: ActionRunWatchExitError}, err + } + return ActionRunWatchResult{ExitCode: ActionRunWatchExitError}, watchCtx.Err() + case now := <-ticker.C: + done, result, err := session.poll(watchCtx, now.UTC()) + if err != nil || done { + return result, err + } + } + } +} + +func validateActionRunWatchOptions(opts ActionRunWatchOptions) error { + if len(opts.RunIDs) == 0 { + return fmt.Errorf("at least one run ID is required") + } + if opts.Interval <= 0 { + return fmt.Errorf("interval must be greater than zero") + } + if opts.StallAfter < 0 { + return fmt.Errorf("stall duration cannot be negative") + } + if opts.Timeout <= 0 { + return fmt.Errorf("timeout must be greater than zero") + } + return nil +} + +func newActionRunWatchSession( + client ActionRunWatchClient, + owner, repo string, + opts ActionRunWatchOptions, + emit func(ActionRunWatchEvent) error, +) *actionRunWatchSession { + return &actionRunWatchSession{ + client: client, + owner: owner, + repo: repo, + opts: opts, + emit: emit, + runs: make(map[int64]*actionRunWatchTrackedRun, len(opts.RunIDs)), + } +} + +func (s *actionRunWatchSession) initialize(ctx stdctx.Context, now time.Time) (bool, ActionRunWatchResult, error) { + headMismatch := false + for _, runID := range s.opts.RunIDs { + run, err := s.fetchRun(ctx, runID) + if err != nil { + return false, ActionRunWatchResult{ExitCode: ActionRunWatchExitError}, err + } + jobs, err := s.fetchJobs(ctx, runID) + if err != nil { + return false, ActionRunWatchResult{ExitCode: ActionRunWatchExitError}, err + } + tracked := &actionRunWatchTrackedRun{ + run: run, + jobs: jobs, + lastChange: now, + } + tracked.resetJobProbe(now, s.opts.StallAfter) + s.runs[runID] = tracked + + matches := s.opts.ExpectedHead == "" || strings.EqualFold(run.HeadSHA, s.opts.ExpectedHead) + if !matches { + headMismatch = true + } + eventRun := run + if err := s.emit(ActionRunWatchEvent{ + Type: "binding", + ObservedAt: now, + RunID: runID, + ExpectedHead: s.opts.ExpectedHead, + HeadMatches: matches, + Run: &eventRun, + Jobs: sortedActionRunWatchJobs(jobs), + }); err != nil { + return false, ActionRunWatchResult{ExitCode: ActionRunWatchExitError}, err + } + } + + if headMismatch { + if err := s.emitSummary("head_mismatch", now); err != nil { + return false, ActionRunWatchResult{ExitCode: ActionRunWatchExitError}, err + } + return true, ActionRunWatchResult{ + ExitCode: ActionRunWatchExitHeadMismatch, + Message: "one or more workflow runs do not match --expect-head", + }, nil + } + if s.allTerminal() { + return s.finish(now) + } + return false, ActionRunWatchResult{}, nil +} + +func (s *actionRunWatchSession) poll(ctx stdctx.Context, now time.Time) (bool, ActionRunWatchResult, error) { + for _, runID := range s.opts.RunIDs { + previous := s.runs[runID] + if isActionRunTerminal(previous.run) { + continue + } + + run, err := s.fetchRun(ctx, runID) + if err != nil { + return false, ActionRunWatchResult{ExitCode: ActionRunWatchExitError}, err + } + changes := diffActionRunWatchRuns(previous.run, run) + runChanged := len(changes) > 0 + stallProbe := previous.jobProbeDue(now, s.opts.StallAfter) + jobs := previous.jobs + if runChanged || isActionRunTerminal(run) || stallProbe { + jobs, err = s.fetchJobs(ctx, runID) + if err != nil { + return false, ActionRunWatchResult{ExitCode: ActionRunWatchExitError}, err + } + changes = append(changes, diffActionRunWatchJobs(previous.jobs, jobs)...) + } + + current := &actionRunWatchTrackedRun{ + run: run, + jobs: jobs, + lastChange: previous.lastChange, + nextJobProbe: previous.nextJobProbe, + stalled: previous.stalled, + } + if len(changes) > 0 { + if previous.stalled { + if err := s.emit(ActionRunWatchEvent{ + Type: "recovery", + ObservedAt: now, + RunID: runID, + }); err != nil { + return false, ActionRunWatchResult{ExitCode: ActionRunWatchExitError}, err + } + } + current.lastChange = now + current.stalled = false + current.resetJobProbe(now, s.opts.StallAfter) + if err := s.emit(ActionRunWatchEvent{ + Type: "transition", + ObservedAt: now, + RunID: runID, + Changes: changes, + }); err != nil { + return false, ActionRunWatchResult{ExitCode: ActionRunWatchExitError}, err + } + } else if stallProbe { + current.resetJobProbe(now, s.opts.StallAfter) + if !previous.stalled { + current.stalled = true + if err := s.emit(ActionRunWatchEvent{ + Type: "stall", + ObservedAt: now, + RunID: runID, + UnchangedFor: now.Sub(previous.lastChange), + }); err != nil { + return false, ActionRunWatchResult{ExitCode: ActionRunWatchExitError}, err + } + } + } + s.runs[runID] = current + } + + if s.allTerminal() { + return s.finish(now) + } + return false, ActionRunWatchResult{}, nil +} + +func (s *actionRunWatchSession) fetchRun(ctx stdctx.Context, runID int64) (ActionRunWatchRun, error) { + run, _, err := s.client.GetRepoRun(ctx, s.owner, s.repo, runID) + if err != nil { + return ActionRunWatchRun{}, fmt.Errorf("failed to get run %d: %w", runID, err) + } + workflow := run.DisplayTitle + if workflow == "" { + workflow = run.Path + } + return ActionRunWatchRun{ + RunID: runID, + HeadSHA: run.HeadSha, + Workflow: workflow, + Path: run.Path, + Status: run.Status, + Conclusion: run.Conclusion, + }, nil +} + +func (s *actionRunWatchSession) fetchJobs(ctx stdctx.Context, runID int64) (map[int64]ActionRunWatchJob, error) { + response, _, err := s.client.ListRepoJobsByRun(ctx, s.owner, s.repo, runID, gitea.ListRepoActionsJobsOptions{}) + if err != nil { + return nil, fmt.Errorf("failed to get jobs for run %d: %w", runID, err) + } + jobs := make(map[int64]ActionRunWatchJob) + if response == nil { + return jobs, nil + } + for _, job := range response.Jobs { + jobs[job.ID] = ActionRunWatchJob{ + JobID: job.ID, + Name: job.Name, + Status: job.Status, + Conclusion: job.Conclusion, + } + } + return jobs, nil +} + +func (s *actionRunWatchSession) finish(now time.Time) (bool, ActionRunWatchResult, error) { + outcome := "success" + for _, run := range s.runs { + if !isActionRunSuccessful(run.run) { + outcome = "unsuccessful" + break + } + } + if err := s.emitSummary(outcome, now); err != nil { + return false, ActionRunWatchResult{ExitCode: ActionRunWatchExitError}, err + } + if outcome == "unsuccessful" { + return true, ActionRunWatchResult{ + ExitCode: ActionRunWatchExitUnsuccessful, + Message: "one or more workflow runs were unsuccessful", + }, nil + } + return true, ActionRunWatchResult{ExitCode: ActionRunWatchExitSuccess}, nil +} + +func (s *actionRunWatchSession) emitSummary(outcome string, now time.Time) error { + ids := make([]int64, 0, len(s.runs)) + for runID := range s.runs { + ids = append(ids, runID) + } + sort.Slice(ids, func(i, j int) bool { return ids[i] < ids[j] }) + runs := make([]ActionRunWatchRun, 0, len(ids)) + for _, runID := range ids { + runs = append(runs, s.runs[runID].run) + } + return s.emit(ActionRunWatchEvent{ + Type: "summary", + ObservedAt: now, + Outcome: outcome, + Runs: runs, + }) +} + +func (s *actionRunWatchSession) allTerminal() bool { + for _, run := range s.runs { + if !isActionRunTerminal(run.run) { + return false + } + } + return true +} + +func (r *actionRunWatchTrackedRun) resetJobProbe(now time.Time, stallAfter time.Duration) { + if stallAfter <= 0 { + r.nextJobProbe = time.Time{} + return + } + r.nextJobProbe = now.Add(stallAfter) +} + +func (r *actionRunWatchTrackedRun) jobProbeDue(now time.Time, stallAfter time.Duration) bool { + return stallAfter > 0 && !r.nextJobProbe.IsZero() && !now.Before(r.nextJobProbe) +} + +func diffActionRunWatchRuns(before, after ActionRunWatchRun) []ActionRunWatchChange { + changes := make([]ActionRunWatchChange, 0, 2) + if before.Status != after.Status { + changes = append(changes, ActionRunWatchChange{Scope: "run", Field: "status", Before: before.Status, After: after.Status}) + } + if before.Conclusion != after.Conclusion { + changes = append(changes, ActionRunWatchChange{Scope: "run", Field: "conclusion", Before: before.Conclusion, After: after.Conclusion}) + } + return changes +} + +func diffActionRunWatchJobs(before, after map[int64]ActionRunWatchJob) []ActionRunWatchChange { + idsByValue := make(map[int64]struct{}, len(before)+len(after)) + for jobID := range before { + idsByValue[jobID] = struct{}{} + } + for jobID := range after { + idsByValue[jobID] = struct{}{} + } + ids := make([]int64, 0, len(idsByValue)) + for jobID := range idsByValue { + ids = append(ids, jobID) + } + sort.Slice(ids, func(i, j int) bool { return ids[i] < ids[j] }) + + changes := make([]ActionRunWatchChange, 0) + for _, jobID := range ids { + oldJob := before[jobID] + newJob := after[jobID] + name := newJob.Name + if name == "" { + name = oldJob.Name + } + if oldJob.Status != newJob.Status { + changes = append(changes, ActionRunWatchChange{ + Scope: "job", JobID: jobID, Name: name, Field: "status", Before: oldJob.Status, After: newJob.Status, + }) + } + if oldJob.Conclusion != newJob.Conclusion { + changes = append(changes, ActionRunWatchChange{ + Scope: "job", JobID: jobID, Name: name, Field: "conclusion", Before: oldJob.Conclusion, After: newJob.Conclusion, + }) + } + } + return changes +} + +func sortedActionRunWatchJobs(jobs map[int64]ActionRunWatchJob) []ActionRunWatchJob { + ids := make([]int64, 0, len(jobs)) + for jobID := range jobs { + ids = append(ids, jobID) + } + sort.Slice(ids, func(i, j int) bool { return ids[i] < ids[j] }) + result := make([]ActionRunWatchJob, 0, len(ids)) + for _, jobID := range ids { + result = append(result, jobs[jobID]) + } + return result +} + +func isActionRunTerminal(run ActionRunWatchRun) bool { + if run.Conclusion != "" { + return true + } + switch strings.ToLower(run.Status) { + case "success", "failure", "canceled", "cancel" + "led", "skipped", "neutral", "timed_out", "completed": + return true + default: + return false + } +} + +func isActionRunSuccessful(run ActionRunWatchRun) bool { + result := run.Conclusion + if result == "" { + result = run.Status + } + switch strings.ToLower(result) { + case "success", "neutral", "skipped": + return true + default: + return false + } +} diff --git a/modules/task/actions_run_watch_test.go b/modules/task/actions_run_watch_test.go new file mode 100644 index 00000000..600bdd2e --- /dev/null +++ b/modules/task/actions_run_watch_test.go @@ -0,0 +1,259 @@ +// Copyright 2026 The Gitea Authors. All rights reserved. +// SPDX-License-Identifier: MIT + +package task + +import ( + stdctx "context" + "errors" + "testing" + "time" + + gitea "gitea.dev/sdk" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +type scriptedActionRunWatchClient struct { + runs map[int64][]*gitea.ActionWorkflowRun + jobs map[int64][][]*gitea.ActionWorkflowJob + runCalls map[int64]int + jobCalls map[int64]int +} + +func newScriptedActionRunWatchClient() *scriptedActionRunWatchClient { + return &scriptedActionRunWatchClient{ + runs: make(map[int64][]*gitea.ActionWorkflowRun), + jobs: make(map[int64][][]*gitea.ActionWorkflowJob), + runCalls: make(map[int64]int), + jobCalls: make(map[int64]int), + } +} + +func (c *scriptedActionRunWatchClient) GetRepoRun(_ stdctx.Context, _, _ string, runID int64) (*gitea.ActionWorkflowRun, *gitea.Response, error) { + sequence := c.runs[runID] + if len(sequence) == 0 { + return nil, nil, errors.New("missing run script") + } + index := c.runCalls[runID] + c.runCalls[runID]++ + if index >= len(sequence) { + index = len(sequence) - 1 + } + return sequence[index], nil, nil +} + +func (c *scriptedActionRunWatchClient) ListRepoJobsByRun(_ stdctx.Context, _, _ string, runID int64, _ gitea.ListRepoActionsJobsOptions) (*gitea.ActionWorkflowJobsResponse, *gitea.Response, error) { + sequence := c.jobs[runID] + if len(sequence) == 0 { + return nil, nil, errors.New("missing jobs script") + } + index := c.jobCalls[runID] + c.jobCalls[runID]++ + if index >= len(sequence) { + index = len(sequence) - 1 + } + return &gitea.ActionWorkflowJobsResponse{Jobs: sequence[index]}, nil, nil +} + +func TestActionRunWatchNormalPollsFetchRunsOnly(t *testing.T) { + client := newScriptedActionRunWatchClient() + client.runs[1] = []*gitea.ActionWorkflowRun{ + testActionRun(1, "queued", ""), + testActionRun(1, "queued", ""), + testActionRun(1, "in_progress", ""), + testActionRun(1, "completed", "success"), + } + client.jobs[1] = [][]*gitea.ActionWorkflowJob{ + {{ID: 10, Name: "test", Status: "queued"}}, + {{ID: 10, Name: "test", Status: "in_progress"}}, + {{ID: 10, Name: "test", Status: "completed", Conclusion: "success"}}, + } + events := make([]ActionRunWatchEvent, 0) + session := newActionRunWatchSession(client, "gitea", "tea", testWatchOptions(), func(event ActionRunWatchEvent) error { + events = append(events, event) + return nil + }) + t0 := time.Date(2026, time.August, 3, 10, 0, 0, 0, time.UTC) + done, _, err := session.initialize(t.Context(), t0) + require.NoError(t, err) + assert.False(t, done) + + done, _, err = session.poll(t.Context(), t0.Add(time.Minute)) + require.NoError(t, err) + assert.False(t, done) + assert.Equal(t, 2, client.runCalls[1]) + assert.Equal(t, 1, client.jobCalls[1], "unchanged normal ticks must not fetch jobs") + + done, _, err = session.poll(t.Context(), t0.Add(2*time.Minute)) + require.NoError(t, err) + assert.False(t, done) + assert.Equal(t, 3, client.runCalls[1]) + assert.Equal(t, 2, client.jobCalls[1], "run transitions refresh jobs once") + + done, result, err := session.poll(t.Context(), t0.Add(3*time.Minute)) + require.NoError(t, err) + assert.True(t, done) + assert.Equal(t, ActionRunWatchExitSuccess, result.ExitCode) + assert.Equal(t, 4, client.runCalls[1]) + assert.Equal(t, 3, client.jobCalls[1], "terminal transitions refresh jobs once") + assert.Equal(t, []string{"binding", "transition", "transition", "summary"}, actionRunWatchEventTypes(events)) +} + +func TestActionRunWatchStallProbeDetectsJobChangeBeforeStall(t *testing.T) { + client := newScriptedActionRunWatchClient() + client.runs[1] = []*gitea.ActionWorkflowRun{ + testActionRun(1, "queued", ""), + testActionRun(1, "queued", ""), + testActionRun(1, "queued", ""), + testActionRun(1, "queued", ""), + testActionRun(1, "queued", ""), + testActionRun(1, "queued", ""), + } + client.jobs[1] = [][]*gitea.ActionWorkflowJob{ + {{ID: 10, Name: "test", Status: "queued"}}, + {{ID: 10, Name: "test", Status: "in_progress"}}, + {{ID: 10, Name: "test", Status: "in_progress"}}, + {{ID: 10, Name: "test", Status: "completed", Conclusion: "success"}}, + } + events := make([]ActionRunWatchEvent, 0) + session := newActionRunWatchSession(client, "gitea", "tea", testWatchOptions(), func(event ActionRunWatchEvent) error { + events = append(events, event) + return nil + }) + t0 := time.Date(2026, time.August, 3, 10, 0, 0, 0, time.UTC) + done, _, err := session.initialize(t.Context(), t0) + require.NoError(t, err) + assert.False(t, done) + + _, _, err = session.poll(t.Context(), t0.Add(29*time.Minute)) + require.NoError(t, err) + assert.Equal(t, 1, client.jobCalls[1]) + + _, _, err = session.poll(t.Context(), t0.Add(30*time.Minute)) + require.NoError(t, err) + assert.Equal(t, []string{"binding", "transition"}, actionRunWatchEventTypes(events), "a job change at the probe must reset the stall deadline") + + _, _, err = session.poll(t.Context(), t0.Add(59*time.Minute)) + require.NoError(t, err) + assert.Equal(t, 2, client.jobCalls[1]) + + _, _, err = session.poll(t.Context(), t0.Add(60*time.Minute)) + require.NoError(t, err) + assert.Equal(t, []string{"binding", "transition", "stall"}, actionRunWatchEventTypes(events)) + assert.Equal(t, 30*time.Minute, events[2].UnchangedFor) + + _, _, err = session.poll(t.Context(), t0.Add(90*time.Minute)) + require.NoError(t, err) + assert.Equal(t, []string{"binding", "transition", "stall", "recovery", "transition"}, actionRunWatchEventTypes(events)) + assert.Equal(t, 6, client.runCalls[1]) + assert.Equal(t, 4, client.jobCalls[1], "jobs are fetched only initially and at scheduled stall probes") +} + +func TestActionRunWatchPreflightsEveryExpectedHead(t *testing.T) { + client := newScriptedActionRunWatchClient() + client.runs[1] = []*gitea.ActionWorkflowRun{testActionRun(1, "queued", "")} + client.runs[2] = []*gitea.ActionWorkflowRun{testActionRun(2, "queued", "")} + client.runs[2][0].HeadSha = "different" + client.jobs[1] = [][]*gitea.ActionWorkflowJob{{}} + client.jobs[2] = [][]*gitea.ActionWorkflowJob{{}} + opts := testWatchOptions() + opts.RunIDs = []int64{1, 2} + events := make([]ActionRunWatchEvent, 0) + session := newActionRunWatchSession(client, "gitea", "tea", opts, func(event ActionRunWatchEvent) error { + events = append(events, event) + return nil + }) + done, result, err := session.initialize(t.Context(), time.Now().UTC()) + require.NoError(t, err) + assert.True(t, done) + assert.Equal(t, ActionRunWatchExitHeadMismatch, result.ExitCode) + assert.Equal(t, []string{"binding", "binding", "summary"}, actionRunWatchEventTypes(events)) + assert.True(t, events[0].HeadMatches) + assert.False(t, events[1].HeadMatches) + assert.Equal(t, 1, client.runCalls[1]) + assert.Equal(t, 1, client.runCalls[2]) + assert.Equal(t, 1, client.jobCalls[1]) + assert.Equal(t, 1, client.jobCalls[2]) +} + +func TestActionRunWatchUnsuccessfulTerminalExit(t *testing.T) { + client := newScriptedActionRunWatchClient() + client.runs[1] = []*gitea.ActionWorkflowRun{testActionRun(1, "completed", "failure")} + client.jobs[1] = [][]*gitea.ActionWorkflowJob{{{ID: 10, Status: "completed", Conclusion: "failure"}}} + events := make([]ActionRunWatchEvent, 0) + session := newActionRunWatchSession(client, "gitea", "tea", testWatchOptions(), func(event ActionRunWatchEvent) error { + events = append(events, event) + return nil + }) + done, result, err := session.initialize(t.Context(), time.Now().UTC()) + require.NoError(t, err) + assert.True(t, done) + assert.Equal(t, ActionRunWatchExitUnsuccessful, result.ExitCode) + assert.Equal(t, "unsuccessful", events[len(events)-1].Outcome) +} + +func TestActionRunWatchTreatsCancelledStatusAsTerminal(t *testing.T) { + assert.True(t, isActionRunTerminal(ActionRunWatchRun{Status: "canceled"})) + assert.True(t, isActionRunTerminal(ActionRunWatchRun{Status: "cancel" + "led"})) +} + +func TestWatchActionRunsTimeoutAndContext(t *testing.T) { + t.Run("timeout", func(t *testing.T) { + client := newScriptedActionRunWatchClient() + client.runs[1] = []*gitea.ActionWorkflowRun{testActionRun(1, "queued", "")} + client.jobs[1] = [][]*gitea.ActionWorkflowJob{{}} + opts := testWatchOptions() + opts.Interval = time.Hour + opts.Timeout = 5 * time.Millisecond + events := make([]ActionRunWatchEvent, 0) + result, err := WatchActionRuns(t.Context(), client, "gitea", "tea", opts, func(event ActionRunWatchEvent) error { + events = append(events, event) + return nil + }) + require.NoError(t, err) + assert.Equal(t, ActionRunWatchExitTimeout, result.ExitCode) + assert.Equal(t, []string{"binding", "summary"}, actionRunWatchEventTypes(events)) + assert.Equal(t, "timeout", events[1].Outcome) + }) + + t.Run("pre-canceled context", func(t *testing.T) { + client := newScriptedActionRunWatchClient() + ctx, cancel := stdctx.WithCancel(t.Context()) + cancel() + result, err := WatchActionRuns(ctx, client, "gitea", "tea", testWatchOptions(), func(ActionRunWatchEvent) error { return nil }) + require.ErrorIs(t, err, stdctx.Canceled) + assert.Equal(t, ActionRunWatchExitError, result.ExitCode) + assert.Empty(t, client.runCalls) + assert.Empty(t, client.jobCalls) + }) +} + +func testWatchOptions() ActionRunWatchOptions { + return ActionRunWatchOptions{ + RunIDs: []int64{1}, + ExpectedHead: "abc123", + Interval: time.Minute, + StallAfter: 30 * time.Minute, + Timeout: time.Hour, + } +} + +func testActionRun(runID int64, status, conclusion string) *gitea.ActionWorkflowRun { + return &gitea.ActionWorkflowRun{ + ID: runID, + HeadSha: "abc123", + DisplayTitle: "CI", + Path: ".gitea/workflows/ci.yml", + Status: status, + Conclusion: conclusion, + } +} + +func actionRunWatchEventTypes(events []ActionRunWatchEvent) []string { + result := make([]string, 0, len(events)) + for _, event := range events { + result = append(result, event.Type) + } + return result +}