14 files changed,
+994,
-57
+19,
-5
1@@ -287,8 +287,10 @@ Rules
2 - it runs off the end of the happy path → `success`;
3 - a result says `next: "stop"` → `success`, or `failure` if that result is
4 a failure (the way for an action to end a run early on purpose);
5- - an action's `then: stop` → `success` only if the cursor is past the last
6- step, otherwise `failure` ("stopped after rollback, before record");
7+ - an action's `then: stop` → `failure` if that action is off the happy
8+ path ("diverted: stopped after rollback"), even when the last step had
9+ already started; for a happy-path step, `success` only if it was the
10+ last one;
11 - a failure with no `next` → `failure`; a contract or engine error →
12 `failure`; the loop guard → `failure`;
13 - dismissing a declared gate reached by default routing → `cancelled`;
14@@ -317,7 +319,7 @@ ci run <wf> --until plan # stop before an action: safe rehearsal
15 ci run <wf> --stub health-check=fixtures/degraded.json # fake a result, test routing
16 ci run <wf> --detach # exit at gates, as on the runner
17 ci action <name> --no-follow # run one action, print result + what routing would do
18-ci resume <run> [--choose rollback] # answer a gate / continue a paused run
19+ci resume <run> [--choose rollback|N] [--dismiss] [--expire] # answer a paused run's gate
20 ci action rollback --run <run> # manual action against an earlier run's state
21 ci view <run> # render events.ndjson
22 ci runs # list local runs ($CI_RUNS_DIR, default ~/.local/state/ci/runs)
23@@ -389,6 +391,13 @@ Repos
24 compacts the mirrors. Mirrors have `gc.auto=0`, so objects a live
25 checkout borrows aren't pruned under it.
26
27+Manual actions against a run: `ci action --run <run> <action>` (only
28+`manual: true` actions, as in the appview) starts a **child run** that shares
29+the parent's checkouts and `$CI_STATE` (and gets `$CI_PARENT_RUN`). It runs
30+the action, follows only what its result explicitly routes to, and ends with
31+the action's own status, so a rollback is recorded as its own run ("rollback
32+done") instead of rewriting the deploy run's history. Declared gates apply.
33+
34 Segments: a gate ends the process. The engine runs actions until it reaches a
35 gate, writes a `gate` event, saves `state.json` and exits with a "paused"
36 status. Locally it prompts inline instead (unless `--detach`). On the runner
37@@ -396,8 +405,13 @@ the daemon publishes the gate and later starts `ci resume <run> --choose X` in
38 a new unit. Nothing runs while a gate waits, and runner restarts never lose a
39 run.
40
41-Concurrency: `flock` on a lock directory (local state dir, or
42-`/var/lib/ci/locks` on the runner). Optional later: ask the runner for the lock
43+Concurrency: `flock` on `<runs>/.locks/<group>.lock` (the runner passes
44+`/var/lib/ci/locks`). A run takes a group's lock when it first runs an action
45+in the group and holds it until it ends or pauses at a gate, so a run waiting
46+for a person never blocks others. `supersede: true` cancels older runs of the
47+same repo and workflow waiting at a gate in that group when a newer run
48+reaches one. Gates past `expiresAt` expire on `ci resume` (any answer) or
49+`ci resume --expire` (the runner calls it on a timer). Optional later: ask the runner for the lock
50 when reachable, so a laptop deploy and a CI deploy can't overlap.
51
52 ## 10. Runner daemon (CI host)
+75,
-20
1@@ -160,11 +160,13 @@ func cmdAction(args []string) int { return runCommon("action", args) }
2
3 func runCommon(name string, args []string) int {
4 f := newRunFlags(name)
5- noFollow := false
6+ noFollow, detach, parentRun := false, false, ""
7+ f.fs.BoolVar(&detach, "detach", false, "pause at gates instead of prompting (as on the runner); continue with ci resume")
8 usage := "ci run [flags] <workflow>"
9 if name == "action" {
10 f.fs.BoolVar(&noFollow, "no-follow", false, "run only this action; report what routing would do next")
11- usage = "ci action [flags] <workflow> <action>"
12+ f.fs.StringVar(&parentRun, "run", "", "run a manual action against an earlier `run` (its checkouts and $CI_STATE)")
13+ usage = "ci action [flags] <workflow> <action>\n ci action --run <run> [flags] <action>"
14 }
15 f.fs.Usage = func() {
16 fmt.Fprintf(os.Stderr, "usage: %s\n", usage)
17@@ -175,7 +177,7 @@ func runCommon(name string, args []string) int {
18 return exitUsage
19 }
20 want := 1
21- if name == "action" {
22+ if name == "action" && parentRun == "" {
23 want = 2
24 }
25 if len(pos) != want {
26@@ -187,31 +189,46 @@ func runCommon(name string, args []string) int {
27 fmt.Fprintln(os.Stderr, "ci:", err)
28 return exitUsage
29 }
30- m, root, err := loadWorkflow(pos[0])
31- if err != nil {
32- fmt.Fprintln(os.Stderr, "ci:", err)
33- return exitFailure
34- }
35- for a := range f.stubs {
36- if _, ok := m.Actions[a]; !ok {
37- fmt.Fprintf(os.Stderr, "ci: --stub %s: no such action in %s\n", a, m.Name)
38- return exitUsage
39- }
40- }
41 if f.rev != "" && !f.clean {
42 fmt.Fprintln(os.Stderr, "ci: --rev needs --clean")
43 return exitUsage
44 }
45 id := tid.New()
46 dir := engine.Dir(filepath.Join(runsDir(), id))
47- meta, err := prepareLocal(m, root, dir, f)
48+
49+ var (
50+ m *manifest.Manifest
51+ meta engine.Meta
52+ from string
53+ )
54+ if parentRun != "" {
55+ m, meta, err = prepareChild(parentRun, id, pos[0])
56+ from = pos[0]
57+ } else {
58+ var root string
59+ if m, root, err = loadWorkflow(pos[0]); err == nil {
60+ meta, err = prepareLocal(m, root, dir, f)
61+ if err != nil {
62+ _ = os.RemoveAll(dir.Path())
63+ }
64+ }
65+ if name == "action" {
66+ from = pos[1]
67+ }
68+ }
69 if err != nil {
70- _ = os.RemoveAll(dir.Path())
71 fmt.Fprintln(os.Stderr, "ci:", err)
72 return exitFailure
73 }
74+ for a := range f.stubs {
75+ if _, ok := m.Actions[a]; !ok {
76+ fmt.Fprintf(os.Stderr, "ci: --stub %s: no such action in %s\n", a, m.Name)
77+ return exitUsage
78+ }
79+ }
80 e, err := engine.Create(m, dir, meta, ex, engine.Options{
81 Stubs: f.stubs, Until: f.until, NoFollow: noFollow, MaxTransitions: f.maxTrans,
82+ Gates: terminalGates, Detach: detach,
83 })
84 if err != nil {
85 fmt.Fprintln(os.Stderr, "ci:", err)
86@@ -231,8 +248,8 @@ func runCommon(name string, args []string) int {
87 ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
88 defer stop()
89 var st *engine.State
90- if name == "action" {
91- st, err = e.RunFrom(ctx, pos[1])
92+ if from != "" {
93+ st, err = e.RunFrom(ctx, from)
94 } else {
95 st, err = e.Run(ctx)
96 }
97@@ -240,8 +257,43 @@ func runCommon(name string, args []string) int {
98 fmt.Fprintln(os.Stderr, "ci:", err)
99 return exitFailure
100 }
101- fmt.Fprintf(os.Stderr, "run %s: %s\n", id, dir)
102- return exitCode(st.Status)
103+ return finished(st, dir)
104+}
105+
106+// prepareChild sets up a manual action run against an earlier run: the
107+// earlier run's manifest (the version that ran), checkouts and $CI_STATE.
108+// Only actions marked manual can be run this way, as in the appview.
109+func prepareChild(parent, id, action string) (*manifest.Manifest, engine.Meta, error) {
110+ pdir, err := runDir(parent)
111+ if err != nil {
112+ return nil, engine.Meta{}, err
113+ }
114+ pst, err := engine.Load(pdir)
115+ if err != nil {
116+ return nil, engine.Meta{}, err
117+ }
118+ raw, err := os.ReadFile(pdir.ManifestFile())
119+ if err != nil {
120+ return nil, engine.Meta{}, err
121+ }
122+ m, err := manifest.Parse(pst.Meta.Workflow, raw)
123+ if err != nil {
124+ return nil, engine.Meta{}, err
125+ }
126+ a, ok := m.Actions[action]
127+ if !ok {
128+ return nil, engine.Meta{}, fmt.Errorf("no action %q in %s", action, m.Name)
129+ }
130+ if !a.Manual {
131+ return nil, engine.Meta{}, fmt.Errorf("%s is not a manual action (manual: true) in %s", action, m.Name)
132+ }
133+ meta := pst.Meta
134+ root := pdir.Path()
135+ if pst.Meta.Parent != "" {
136+ root = pst.Meta.Parent // a manual run of a manual run shares the original's state
137+ }
138+ meta.Run, meta.Parent, meta.Kind, meta.Host = id, root, "manual", "local"
139+ return m, meta, nil
140 }
141
142 // prepareLocal lays out a local run's checkouts (DESIGN.md §9 Repos): the
143@@ -477,6 +529,9 @@ func cmdRuns(args []string) int {
144 } else if st.Gate != nil {
145 detail = "waiting at gate: " + st.Gate.Prompt
146 }
147+ if st.Meta.Parent != "" {
148+ detail += " (manual, for " + filepath.Base(st.Meta.Parent) + ")"
149+ }
150 fmt.Printf("%s %-9s %-12s %-10s %s\n", filepath.Base(string(d)), st.Status, st.Meta.Repo, st.Meta.Workflow, detail)
151 }
152 return exitSuccess
+236,
-0
1@@ -0,0 +1,236 @@
2+package main
3+
4+import (
5+ "bufio"
6+ "context"
7+ "encoding/json"
8+ "flag"
9+ "fmt"
10+ "os"
11+ "os/signal"
12+ "os/user"
13+ "path/filepath"
14+ "strconv"
15+ "strings"
16+ "syscall"
17+
18+ "git.kilimanjaro.io/ci/internal/engine"
19+ "git.kilimanjaro.io/ci/internal/lex"
20+ "git.kilimanjaro.io/ci/internal/render"
21+)
22+
23+func init() {
24+ commands["resume"] = command{"answer a paused run's gate and continue it", cmdResume}
25+}
26+
27+// actor names whoever answers a gate from this terminal.
28+func actor() string {
29+ name := os.Getenv("USER")
30+ if u, err := user.Current(); err == nil {
31+ name = u.Username
32+ }
33+ host, _ := os.Hostname()
34+ return "local:" + name + "@" + host
35+}
36+
37+func isTerminal(f *os.File) bool {
38+ fi, err := f.Stat()
39+ return err == nil && fi.Mode()&os.ModeCharDevice != 0
40+}
41+
42+// terminalGates prompts on the terminal. Without one (a script, an agent),
43+// every gate pauses the run, to be answered with `ci resume` or the appview.
44+var terminalGates = engine.GateFunc(func(ctx context.Context, e *engine.Engine, g *engine.PendingGate) (engine.Choice, error) {
45+ if !isTerminal(os.Stdin) || !isTerminal(os.Stdout) {
46+ return engine.Choice{}, engine.ErrPause
47+ }
48+ return promptGate(e, g)
49+})
50+
51+func promptGate(e *engine.Engine, g *engine.PendingGate) (engine.Choice, error) {
52+ w := os.Stdout
53+ fmt.Fprintln(w)
54+ fmt.Fprintf(w, "⏸ %s\n", orDefault(g.Prompt, "Choose the next step"))
55+ if g.Summary != "" {
56+ for _, l := range strings.Split(strings.TrimRight(g.Summary, "\n"), "\n") {
57+ fmt.Fprintln(w, " "+l)
58+ }
59+ } else {
60+ printEarlierSummaries(e)
61+ }
62+ fmt.Fprintf(w, " expires %s\n", g.ExpiresAt)
63+ for i, o := range g.Options {
64+ label := orDefault(o.Label, e.M.Label(o.Action))
65+ extra := ""
66+ if o.Prompt != "" && len(g.Options) > 1 {
67+ extra = " — " + o.Prompt
68+ }
69+ if e.M.Actions[o.Action].Style == "danger" {
70+ label = "!! " + label
71+ }
72+ fmt.Fprintf(w, " [%d] %s%s\n", i+1, label, extra)
73+ }
74+ fmt.Fprintln(w, " [d] dismiss (end the run) [p] pause (answer later with ci resume)")
75+ in := bufio.NewReader(os.Stdin)
76+ for {
77+ fmt.Fprint(w, " > ")
78+ line, err := in.ReadString('\n')
79+ if err != nil {
80+ return engine.Choice{}, engine.ErrPause
81+ }
82+ switch a := strings.TrimSpace(strings.ToLower(line)); a {
83+ case "d":
84+ return engine.Choice{Dismiss: true, Actor: actor()}, nil
85+ case "p":
86+ return engine.Choice{}, engine.ErrPause
87+ case "":
88+ default:
89+ if n, err := strconv.Atoi(a); err == nil && n >= 1 && n <= len(g.Options) {
90+ return engine.Choice{Option: n - 1, Actor: actor()}, nil
91+ }
92+ }
93+ }
94+}
95+
96+// printEarlierSummaries shows what the run's actions reported so far: for a
97+// declared gate (Deploy?), that is the plan's diff.
98+func printEarlierSummaries(e *engine.Engine) {
99+ for _, x := range e.State().History {
100+ b, err := os.ReadFile(filepath.Join(e.Dir.Path(), x.Result))
101+ if err != nil {
102+ continue
103+ }
104+ var r lex.Result
105+ if json.Unmarshal(b, &r) != nil || r.Summary == "" {
106+ continue
107+ }
108+ fmt.Fprintf(os.Stdout, " %s:\n", x.Action)
109+ for _, l := range strings.Split(strings.TrimRight(r.Summary, "\n"), "\n") {
110+ fmt.Fprintln(os.Stdout, " "+l)
111+ }
112+ }
113+}
114+
115+func orDefault(s, d string) string {
116+ if s == "" {
117+ return d
118+ }
119+ return s
120+}
121+
122+func cmdResume(args []string) int {
123+ fs := flag.NewFlagSet("ci resume", flag.ContinueOnError)
124+ choose := fs.String("choose", "", "answer with this option: an action name or its number")
125+ dismiss := fs.Bool("dismiss", false, "dismiss the gate (end the run)")
126+ expire := fs.Bool("expire", false, "expire the gate (only once past its deadline)")
127+ detach := fs.Bool("detach", false, "pause at the next gate instead of prompting")
128+ shell := fs.String("shell", "devenv", "how actions run: devenv or bash")
129+ quiet := fs.Bool("q", false, "don't render events")
130+ fs.Usage = func() {
131+ fmt.Fprintln(os.Stderr, "usage: ci resume [flags] <run-id|run-dir|last>")
132+ fs.PrintDefaults()
133+ }
134+ f := &runFlags{fs: fs}
135+ pos, err := f.parse(args)
136+ if err != nil || len(pos) != 1 {
137+ fs.Usage()
138+ return exitUsage
139+ }
140+ f.shell = *shell
141+ ex, err := f.executor()
142+ if err != nil {
143+ fmt.Fprintln(os.Stderr, "ci:", err)
144+ return exitUsage
145+ }
146+ dir, err := runDir(pos[0])
147+ if err != nil {
148+ fmt.Fprintln(os.Stderr, "ci:", err)
149+ return exitFailure
150+ }
151+ e, err := engine.Open(dir, ex, engine.Options{Gates: terminalGates, Detach: *detach})
152+ if err != nil {
153+ fmt.Fprintln(os.Stderr, "ci:", err)
154+ return exitFailure
155+ }
156+ defer e.Log.Close()
157+ g := e.State().Gate
158+ if e.State().Status != engine.StatusPaused || g == nil {
159+ fmt.Fprintf(os.Stderr, "ci: run %s is %s, not waiting at a gate\n", dir.Base(), e.State().Status)
160+ return exitFailure
161+ }
162+
163+ var c engine.Choice
164+ switch {
165+ case *dismiss:
166+ c = engine.Choice{Dismiss: true, Actor: actor()}
167+ case *expire:
168+ c = engine.Choice{Expire: true, Actor: actor()}
169+ case *choose != "":
170+ i, err := optionIndex(g, *choose)
171+ if err != nil {
172+ fmt.Fprintln(os.Stderr, "ci:", err)
173+ return exitUsage
174+ }
175+ c = engine.Choice{Option: i, Actor: actor()}
176+ default:
177+ if !isTerminal(os.Stdin) || !isTerminal(os.Stdout) {
178+ fmt.Fprintf(os.Stderr, "ci: run %s waits at: %s\n", dir.Base(), orDefault(g.Prompt, "a choice"))
179+ for i, o := range g.Options {
180+ fmt.Fprintf(os.Stderr, " %d %s\n", i+1, o.Action)
181+ }
182+ fmt.Fprintln(os.Stderr, "answer with --choose <action|number> or --dismiss")
183+ return exitUsage
184+ }
185+ if c, err = promptGate(e, g); err != nil {
186+ return exitPaused
187+ }
188+ }
189+ if !*quiet {
190+ e.Log.AddSink(render.New(os.Stdout).Event)
191+ }
192+ ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
193+ defer stop()
194+ st, err := e.Resume(ctx, c)
195+ if err != nil {
196+ fmt.Fprintln(os.Stderr, "ci:", err)
197+ return exitFailure
198+ }
199+ return finished(st, dir)
200+}
201+
202+func optionIndex(g *engine.PendingGate, s string) (int, error) {
203+ if n, err := strconv.Atoi(s); err == nil {
204+ if n >= 1 && n <= len(g.Options) {
205+ return n - 1, nil
206+ }
207+ return 0, fmt.Errorf("no option %d (1-%d)", n, len(g.Options))
208+ }
209+ idx := -1
210+ for i, o := range g.Options {
211+ if o.Action == s {
212+ if idx >= 0 {
213+ return 0, fmt.Errorf("%s is offered more than once; choose by number", s)
214+ }
215+ idx = i
216+ }
217+ }
218+ if idx < 0 {
219+ var names []string
220+ for _, o := range g.Options {
221+ names = append(names, o.Action)
222+ }
223+ return 0, fmt.Errorf("%s is not an option here (options: %s)", s, strings.Join(names, ", "))
224+ }
225+ return idx, nil
226+}
227+
228+// finished reports how a run (segment) ended and returns the exit code.
229+func finished(st *engine.State, dir engine.Dir) int {
230+ if st.Status == engine.StatusPaused && st.Gate != nil {
231+ fmt.Fprintf(os.Stderr, "run %s paused: %s\n answer: ci resume %s --choose <action|number> | --dismiss\n",
232+ dir.Base(), orDefault(st.Gate.Prompt, "choose the next step"), dir.Base())
233+ } else {
234+ fmt.Fprintf(os.Stderr, "run %s: %s\n", dir.Base(), dir)
235+ }
236+ return exitCode(st.Status)
237+}
+24,
-3
1@@ -26,6 +26,13 @@ type Options struct {
2 NoFollow bool
3 // MaxTransitions overrides DefaultMaxTransitions.
4 MaxTransitions int
5+ // Gates decides gates as they are reached; nil pauses at every gate.
6+ Gates Gatekeeper
7+ // Detach pauses at every gate even with a Gatekeeper (the runner's
8+ // mode; locally `ci run --detach`).
9+ Detach bool
10+ // LocksDir holds concurrency locks; default <runs>/.locks.
11+ LocksDir string
12 }
13
14 // Engine runs one workflow run.
15@@ -39,6 +46,7 @@ type Engine struct {
16
17 router *Router
18 st *State
19+ locks map[string]*os.File
20 }
21
22 // Create starts a new run in dir: it writes the manifest copy, state.json and
23@@ -60,7 +68,7 @@ func Create(m *manifest.Manifest, dir Dir, meta Meta, ex Executor, opts Options)
24 t := now()
25 e := &Engine{
26 M: m, Dir: dir, Exec: ex, Log: log, BaseEnv: osEnviron(), Opts: opts,
27- router: &Router{M: m},
28+ router: &Router{M: m, Detached: meta.Parent != ""},
29 st: &State{Version: StateVersion, Meta: meta, Status: StatusRunning, Created: t},
30 }
31 for _, c := range meta.Checkouts {
32@@ -121,7 +129,14 @@ func (e *Engine) drive(ctx context.Context, d Decision) (*State, error) {
33 case d.End != nil:
34 return e.finish(*d.End, e.lastAction())
35 case d.Gate != nil:
36- return e.finish(End{StatusFailure, lex.ReasonError, "reached a gate (" + optionNames(d.Gate) + "); gates are not implemented yet"}, e.lastAction())
37+ next, paused, err := e.reachGate(ctx, d.Gate)
38+ if err != nil {
39+ return e.finish(End{StatusFailure, lex.ReasonError, err.Error()}, e.lastAction())
40+ }
41+ if paused {
42+ return e.st, nil
43+ }
44+ d = next
45 case d.Run != nil:
46 p := *d.Run
47 if e.st.Seq >= e.maxTransitions() {
48@@ -175,6 +190,7 @@ func (e *Engine) emit(ev lex.RunEvent) {
49
50 // finish ends the run.
51 func (e *Engine) finish(end End, from string) (*State, error) {
52+ e.releaseLocks()
53 e.transition(lex.Transition{From: from, Reason: end.Reason, Detail: end.Detail})
54 e.st.End = &end
55 e.st.Status = end.Status
56@@ -217,6 +233,11 @@ func (e *Engine) execute(ctx context.Context, p Pending) (lex.Result, string, er
57 if err := e.st.Save(e.Dir); err != nil {
58 return lex.Result{}, "", err
59 }
60+ if a.Concurrency != nil {
61+ if err := e.acquire(ctx, a.Concurrency.Group); err != nil {
62+ return lex.Result{}, "", errors.Join(err, e.recordExec(x, StatusCancelled, -1))
63+ }
64+ }
65 e.emit(lex.RunEvent{Kind: lex.KindStart, Action: p.Action, Attempt: n, Command: a.Command})
66
67 code, runErr := 0, error(nil)
68@@ -252,7 +273,7 @@ func (e *Engine) execute(ctx context.Context, p Pending) (lex.Result, string, er
69 res = ContractFailure(err)
70 }
71 }
72- if err := mergeOutputs(e.Dir.Outputs(), res.Outputs); err != nil {
73+ if err := mergeOutputs(filepath.Join(e.stateDir(), "outputs.json"), res.Outputs); err != nil {
74 res = ContractFailure(fmt.Errorf("outputs: %w", err))
75 }
76 exit := code
+0,
-17
1@@ -2,7 +2,6 @@ package engine
2
3 import (
4 "context"
5- "encoding/json"
6 "os"
7 "path/filepath"
8 "slices"
9@@ -328,19 +327,3 @@ actions: { a: { command: "sleep 30" }, b: { command: "true" } }
10 t.Fatalf("status %s", st.Status)
11 }
12 }
13-
14-func TestGateNotYetImplemented(t *testing.T) {
15- f := newFixture(t, `
16-engine: devenv
17-steps: [a]
18-actions: { a: { command: "true", gate: true } }
19-`)
20- st, _ := f.run("")
21- if st.Status != StatusFailure || len(st.History) != 0 {
22- t.Fatalf("%s %v", st.Status, executed(st))
23- }
24- b, _ := json.Marshal(st.End)
25- if !strings.Contains(string(b), "not implemented") {
26- t.Errorf("%s", b)
27- }
28-}
+12,
-1
1@@ -32,7 +32,8 @@ func (e *Engine) actionEnv(p Pending, n int, resultFile, inputsFile, triggerFile
2 set("CI_ACTION", p.Action)
3 set("CI_ATTEMPT", strconv.Itoa(n))
4 set("CI_RESULT", resultFile)
5- set("CI_STATE", e.Dir.State())
6+ set("CI_STATE", e.stateDir())
7+ set("CI_PARENT_RUN", m.Parent)
8 set("CI_ARTIFACTS", e.Dir.Artifacts())
9 set("CI_INPUTS", inputsFile)
10 set("CI_TRIGGER_RESULT", triggerFile)
11@@ -76,5 +77,15 @@ func (e *Engine) actionEnv(p Pending, n int, resultFile, inputsFile, triggerFile
12 return out
13 }
14
15+// stateDir is $CI_STATE: the run's own, or for a manual action run against
16+// an earlier run (ci action --run), that run's, so rollback sees what deploy
17+// recorded.
18+func (e *Engine) stateDir() string {
19+ if p := e.st.Meta.Parent; p != "" {
20+ return Dir(p).State()
21+ }
22+ return e.Dir.State()
23+}
24+
25 // osEnviron is os.Environ, replaceable in tests.
26 var osEnviron = os.Environ
+165,
-0
1@@ -0,0 +1,165 @@
2+package engine
3+
4+import (
5+ "context"
6+ "errors"
7+ "fmt"
8+ "os"
9+ "time"
10+
11+ "git.kilimanjaro.io/ci/internal/lex"
12+ "git.kilimanjaro.io/ci/internal/manifest"
13+ "git.kilimanjaro.io/ci/internal/runlog"
14+)
15+
16+// ErrPause tells the engine to leave a gate for later: the run is saved as
17+// paused and the process can exit.
18+var ErrPause = errors.New("paused at gate")
19+
20+// Choice answers a gate.
21+type Choice struct {
22+ Option int // index into the gate's options, when not Dismiss/Expire
23+ Dismiss bool // end the run here
24+ Expire bool // the gate timed out
25+ Actor string
26+}
27+
28+// Gatekeeper decides gates as they are reached. Returning ErrPause (or
29+// having no Gatekeeper) pauses the run; it continues with Resume.
30+type Gatekeeper interface {
31+ Decide(ctx context.Context, e *Engine, g *PendingGate) (Choice, error)
32+}
33+
34+// GateFunc adapts a function to Gatekeeper.
35+type GateFunc func(ctx context.Context, e *Engine, g *PendingGate) (Choice, error)
36+
37+func (f GateFunc) Decide(ctx context.Context, e *Engine, g *PendingGate) (Choice, error) {
38+ return f(ctx, e, g)
39+}
40+
41+// reachGate records g, supersedes older runs waiting at the same
42+// concurrency group, and asks the Gatekeeper. It returns the next decision,
43+// or paused=true when the run should stop here.
44+func (e *Engine) reachGate(ctx context.Context, g *PendingGate) (Decision, bool, error) {
45+ e.st.Gate = g
46+ e.st.Status = StatusPaused
47+ e.emit(lex.RunEvent{Kind: lex.KindGate, Gate: e.gateRecord(g, lex.GateWaiting, Choice{})})
48+ if err := e.st.Save(e.Dir); err != nil {
49+ return Decision{}, false, err
50+ }
51+ e.supersedeOlder(g)
52+
53+ var c Choice
54+ err := ErrPause
55+ if e.Opts.Gates != nil && !e.Opts.Detach {
56+ c, err = e.Opts.Gates.Decide(ctx, e, g)
57+ }
58+ if errors.Is(err, ErrPause) {
59+ e.releaseLocks()
60+ e.emit(lex.RunEvent{Kind: lex.KindRun, Status: StatusPaused, Content: "waiting: " + gateTitle(g)})
61+ return Decision{}, true, e.st.Save(e.Dir)
62+ }
63+ if err != nil {
64+ return Decision{}, false, err
65+ }
66+ d, err := e.answer(g, c)
67+ return d, false, err
68+}
69+
70+// answer records a decision on g and returns what follows.
71+func (e *Engine) answer(g *PendingGate, c Choice) (Decision, error) {
72+ var d Decision
73+ state := lex.GateApproved
74+ switch {
75+ case c.Expire:
76+ state, d = lex.GateExpired, e.router.Expire(g)
77+ case c.Dismiss:
78+ state, d = lex.GateDismissed, e.router.Dismiss(g)
79+ default:
80+ var err error
81+ if d, err = e.router.Choose(g, c.Option); err != nil {
82+ return Decision{}, err
83+ }
84+ }
85+ e.emit(lex.RunEvent{Kind: lex.KindGate, Gate: e.gateRecord(g, state, c)})
86+ e.st.Gate = nil
87+ e.st.Status = StatusRunning
88+ return d, e.st.Save(e.Dir)
89+}
90+
91+func (e *Engine) gateRecord(g *PendingGate, state string, c Choice) *lex.GateRecord {
92+ r := &lex.GateRecord{
93+ Pipeline: e.st.Meta.Pipeline, Run: e.st.Meta.Run, Workflow: e.st.Meta.Workflow, Seq: e.st.Seq,
94+ From: g.From, Prompt: g.Prompt, Summary: g.Summary, Options: g.Options, State: state,
95+ ExpiresAt: g.ExpiresAt, CreatedAt: g.CreatedAt, Actor: c.Actor,
96+ }
97+ if state != lex.GateWaiting {
98+ r.DecidedAt = now()
99+ }
100+ if state == lex.GateApproved && c.Option < len(g.Options) {
101+ r.Choice = g.Options[c.Option].Action
102+ }
103+ return r
104+}
105+
106+func gateTitle(g *PendingGate) string {
107+ if g.Prompt != "" {
108+ return g.Prompt
109+ }
110+ return "choose: " + optionNames(g)
111+}
112+
113+// Expired reports whether the gate's deadline has passed.
114+func (g *PendingGate) Expired(at time.Time) bool {
115+ t, err := time.Parse(time.RFC3339Nano, g.ExpiresAt)
116+ return err == nil && at.After(t)
117+}
118+
119+// Open reopens an existing run (to resume it).
120+func Open(dir Dir, ex Executor, opts Options) (*Engine, error) {
121+ st, err := Load(dir)
122+ if err != nil {
123+ return nil, err
124+ }
125+ raw, err := os.ReadFile(dir.ManifestFile())
126+ if err != nil {
127+ return nil, err
128+ }
129+ m, err := manifest.Parse(st.Meta.Workflow, raw)
130+ if err != nil {
131+ return nil, fmt.Errorf("%s: %w", dir.ManifestFile(), err)
132+ }
133+ log, err := runlog.Open(dir.Events(), st.Meta.Run)
134+ if err != nil {
135+ return nil, err
136+ }
137+ return &Engine{
138+ M: m, Dir: dir, Exec: ex, Log: log, BaseEnv: osEnviron(), Opts: opts,
139+ router: &Router{M: m, Detached: st.Meta.Parent != ""},
140+ st: st,
141+ }, nil
142+}
143+
144+// Resume answers the gate a paused run is waiting at and continues it. A
145+// gate past its deadline expires whatever the choice.
146+func (e *Engine) Resume(ctx context.Context, c Choice) (*State, error) {
147+ g := e.st.Gate
148+ if e.st.Status != StatusPaused || g == nil {
149+ return nil, fmt.Errorf("run %s is %s, not waiting at a gate", e.st.Meta.Run, e.st.Status)
150+ }
151+ if !c.Expire && g.Expired(time.Now()) {
152+ c = Choice{Expire: true, Actor: c.Actor}
153+ }
154+ if c.Expire && !g.Expired(time.Now()) {
155+ return nil, fmt.Errorf("the gate expires at %s, not yet", g.ExpiresAt)
156+ }
157+ if !c.Dismiss && !c.Expire && (c.Option < 0 || c.Option >= len(g.Options)) {
158+ return nil, fmt.Errorf("no option %d", c.Option+1)
159+ }
160+ e.emit(lex.RunEvent{Kind: lex.KindRun, Status: StatusRunning, Content: "resumed"})
161+ d, err := e.answer(g, c)
162+ if err != nil {
163+ return nil, err
164+ }
165+ return e.drive(ctx, d)
166+}
+284,
-0
1@@ -0,0 +1,284 @@
2+package engine
3+
4+import (
5+ "context"
6+ "os"
7+ "path/filepath"
8+ "slices"
9+ "strings"
10+ "syscall"
11+ "testing"
12+ "time"
13+
14+ "git.kilimanjaro.io/ci/internal/lex"
15+)
16+
17+// create starts a run of f's manifest in runs/<id> (sharing f.root).
18+func (f *fixture) create(runs, id string, meta Meta) *Engine {
19+ f.t.Helper()
20+ if meta.Run == "" {
21+ meta = Meta{Run: id, Workflow: "w", Repo: "repo", Kind: "manual", Host: "local", Root: f.root}
22+ }
23+ e, err := Create(f.m, Dir(filepath.Join(runs, id)), meta, Bash{}, f.opts)
24+ if err != nil {
25+ f.t.Fatal(err)
26+ }
27+ return e
28+}
29+
30+func reopen(t *testing.T, d Dir, opts Options) *Engine {
31+ t.Helper()
32+ e, err := Open(d, Bash{}, opts)
33+ if err != nil {
34+ t.Fatal(err)
35+ }
36+ return e
37+}
38+
39+func runE(t *testing.T, e *Engine) *State {
40+ t.Helper()
41+ st, err := e.Run(context.Background())
42+ if err != nil {
43+ t.Fatal(err)
44+ }
45+ return st
46+}
47+
48+func approve(i int) Gatekeeper {
49+ return GateFunc(func(context.Context, *Engine, *PendingGate) (Choice, error) {
50+ return Choice{Option: i, Actor: "test"}, nil
51+ })
52+}
53+
54+const gatedWF = `
55+engine: devenv
56+steps: [plan, deploy, health]
57+actions:
58+ plan: { command: "true" }
59+ deploy: { command: "echo deployed > $CI_STATE/prev", gate: { prompt: "Deploy?" }, concurrency: { group: fleet, supersede: true } }
60+ health:
61+ command: |
62+ n=$(cat $CI_STATE/n 2>/dev/null || echo 0); echo $((n+1)) > $CI_STATE/n
63+ if [ "$n" = 0 ]; then
64+ printf '%s' '{"version":1,"status":"failure","summary":"503","next":[{"action":"rollback"},{"action":"health","label":"Check again"}]}' > $CI_RESULT
65+ fi
66+ rollback: { command: "cat $CI_STATE/prev", manual: true, style: danger, gate: { prompt: "Roll back?" } }
67+`
68+
69+func TestPauseAndResume(t *testing.T) {
70+ f := newFixture(t, gatedWF)
71+ runs := t.TempDir()
72+ e := f.create(runs, "a", Meta{})
73+ st, err := e.Run(context.Background())
74+ e.Log.Close()
75+ if err != nil {
76+ t.Fatal(err)
77+ }
78+ if st.Status != StatusPaused || st.Gate == nil || st.Gate.Prompt != "Deploy?" || !slices.Equal(executed(st), []string{"plan"}) {
79+ t.Fatalf("not paused at deploy: %s %v %+v", st.Status, executed(st), st.Gate)
80+ }
81+
82+ // a new process reopens the run and approves
83+ e = reopen(t, Dir(filepath.Join(runs, "a")), Options{})
84+ st, err = e.Resume(context.Background(), Choice{Option: 0, Actor: "local:me"})
85+ if err != nil {
86+ t.Fatal(err)
87+ }
88+ // health fails the first time and offers a choice: paused again
89+ if st.Status != StatusPaused || len(st.Gate.Options) != 2 || st.Gate.From != "health" {
90+ t.Fatalf("want choice gate after health: %s %+v", st.Status, st.Gate)
91+ }
92+ e.Log.Close()
93+
94+ // "Check again" reruns health, which passes now
95+ e = reopen(t, Dir(filepath.Join(runs, "a")), Options{})
96+ st, err = e.Resume(context.Background(), Choice{Option: 1})
97+ e.Log.Close()
98+ if err != nil {
99+ t.Fatal(err)
100+ }
101+ if st.Status != StatusSuccess || !slices.Equal(executed(st), []string{"plan", "deploy", "health", "health"}) {
102+ t.Fatalf("%s %v %+v", st.Status, executed(st), st.End)
103+ }
104+
105+ // gate events record waiting and approved, with the actor
106+ b, _ := os.ReadFile(filepath.Join(runs, "a", "events.ndjson"))
107+ if !strings.Contains(string(b), `"state":"approved","choice":"deploy","actor":"local:me"`) {
108+ t.Errorf("gate decision not recorded:\n%s", b)
109+ }
110+
111+ // resuming a finished run is an error
112+ e = reopen(t, Dir(filepath.Join(runs, "a")), Options{})
113+ if _, err := e.Resume(context.Background(), Choice{}); err == nil {
114+ t.Error("resumed a finished run")
115+ }
116+ e.Log.Close()
117+}
118+
119+func TestGatekeeperInline(t *testing.T) {
120+ f := newFixture(t, gatedWF)
121+ f.opts.Gates = approve(0) // deploy, then rollback at the health choice
122+ e := f.create(t.TempDir(), "a", Meta{})
123+ st, err := e.Run(context.Background())
124+ e.Log.Close()
125+ if err != nil {
126+ t.Fatal(err)
127+ }
128+ // rollback is off-path: the run ends as a failure after it
129+ if st.Status != StatusFailure || !slices.Equal(executed(st), []string{"plan", "deploy", "health", "rollback"}) {
130+ t.Fatalf("%s %v %+v", st.Status, executed(st), st.End)
131+ }
132+}
133+
134+func TestDetachPausesEvenWithGatekeeper(t *testing.T) {
135+ f := newFixture(t, gatedWF)
136+ f.opts.Gates = approve(0)
137+ f.opts.Detach = true
138+ e := f.create(t.TempDir(), "a", Meta{})
139+ st, _ := e.Run(context.Background())
140+ e.Log.Close()
141+ if st.Status != StatusPaused {
142+ t.Fatalf("status %s", st.Status)
143+ }
144+}
145+
146+func TestDismissAndExpire(t *testing.T) {
147+ f := newFixture(t, gatedWF)
148+ runs := t.TempDir()
149+ e := f.create(runs, "a", Meta{})
150+ runE(t, e)
151+ e.Log.Close()
152+ e = reopen(t, Dir(filepath.Join(runs, "a")), Options{})
153+ if _, err := e.Resume(context.Background(), Choice{Expire: true}); err == nil {
154+ t.Error("expired a gate before its deadline")
155+ }
156+ st, err := e.Resume(context.Background(), Choice{Dismiss: true, Actor: "me"})
157+ e.Log.Close()
158+ if err != nil || st.Status != StatusCancelled || st.End.Reason != lex.ReasonDismiss {
159+ t.Fatalf("dismiss: %v %+v", err, st)
160+ }
161+
162+ // a gate past its deadline expires whatever the choice
163+ f = newFixture(t, strings.Replace(gatedWF, `prompt: "Deploy?"`, `prompt: "Deploy?", timeout: 1ms`, 1))
164+ e = f.create(runs, "b", Meta{})
165+ runE(t, e)
166+ e.Log.Close()
167+ time.Sleep(5 * time.Millisecond)
168+ e = reopen(t, Dir(filepath.Join(runs, "b")), Options{})
169+ st, err = e.Resume(context.Background(), Choice{Option: 0})
170+ e.Log.Close()
171+ if err != nil || st.Status != StatusCancelled || st.End.Reason != lex.ReasonExpired || len(st.History) != 1 {
172+ t.Fatalf("expire: %v %+v", err, st)
173+ }
174+}
175+
176+func TestSupersede(t *testing.T) {
177+ f := newFixture(t, gatedWF)
178+ runs := t.TempDir()
179+ old := f.create(runs, "a", Meta{})
180+ runE(t, old)
181+ old.Log.Close()
182+ time.Sleep(2 * time.Millisecond) // Created must differ
183+ // another workflow waiting in the same group is not touched
184+ other := f.create(runs, "c", Meta{Run: "c", Workflow: "other", Repo: "repo", Kind: "manual", Host: "local", Root: f.root})
185+ runE(t, other)
186+ other.Log.Close()
187+ time.Sleep(2 * time.Millisecond)
188+ newer := f.create(runs, "b", Meta{})
189+ st, _ := newer.Run(context.Background())
190+ newer.Log.Close()
191+ if st.Status != StatusPaused {
192+ t.Fatalf("newer: %s", st.Status)
193+ }
194+ if s, _ := Load(Dir(filepath.Join(runs, "a"))); s.Status != StatusCancelled || !strings.Contains(s.End.Detail, "superseded by run b") {
195+ t.Errorf("older run: %+v", s.End)
196+ }
197+ if s, _ := Load(Dir(filepath.Join(runs, "c"))); s.Status != StatusPaused {
198+ t.Errorf("other workflow: %s", s.Status)
199+ }
200+}
201+
202+func TestConcurrencyLockWaits(t *testing.T) {
203+ f := newFixture(t, wf("deploy", "deploy: true"))
204+ f.m.Actions["deploy"] = func() lex.Action {
205+ a := f.m.Actions["deploy"]
206+ a.Concurrency = &lex.Concurrency{Group: "fleet"}
207+ return a
208+ }()
209+ runs := t.TempDir()
210+ locks := filepath.Join(runs, ".locks")
211+ if err := os.MkdirAll(locks, 0o755); err != nil {
212+ t.Fatal(err)
213+ }
214+ held, err := os.OpenFile(filepath.Join(locks, "fleet.lock"), os.O_CREATE|os.O_RDWR, 0o644)
215+ if err != nil {
216+ t.Fatal(err)
217+ }
218+ if _, err := held.WriteString("other-run\n"); err != nil {
219+ t.Fatal(err)
220+ }
221+ if err := syscall.Flock(int(held.Fd()), syscall.LOCK_EX); err != nil {
222+ t.Fatal(err)
223+ }
224+ e := f.create(runs, "a", Meta{})
225+ defer e.Log.Close()
226+ done := make(chan *State)
227+ go func() { st, _ := e.Run(context.Background()); done <- st }()
228+ time.Sleep(300 * time.Millisecond)
229+ select {
230+ case st := <-done:
231+ t.Fatalf("ran while the lock was held: %+v", st)
232+ default:
233+ }
234+ held.Close() // releases the lock
235+ st := <-done
236+ if st.Status != StatusSuccess {
237+ t.Fatalf("%+v", st.End)
238+ }
239+ b, _ := os.ReadFile(filepath.Join(runs, "a", "events.ndjson"))
240+ if !strings.Contains(string(b), "waiting for concurrency group fleet (held by run other-run)") {
241+ t.Errorf("no waiting message:\n%s", b)
242+ }
243+}
244+
245+func TestManualChildRun(t *testing.T) {
246+ f := newFixture(t, gatedWF)
247+ f.opts.Gates = approve(0)
248+ runs := t.TempDir()
249+ parent := f.create(runs, "a", Meta{})
250+ parent.Opts.Gates = GateFunc(func(_ context.Context, _ *Engine, g *PendingGate) (Choice, error) {
251+ if g.From == "health" {
252+ return Choice{Option: 1}, nil // check again
253+ }
254+ return Choice{Option: 0}, nil
255+ })
256+ pst, _ := parent.Run(context.Background())
257+ parent.Log.Close()
258+ if pst.Status != StatusSuccess {
259+ t.Fatalf("parent %+v", pst.End)
260+ }
261+
262+ // later: roll back against the finished run
263+ meta := pst.Meta
264+ meta.Run, meta.Parent = "b", filepath.Join(runs, "a")
265+ child, err := Create(f.m, Dir(filepath.Join(runs, "b")), meta, Bash{}, Options{Gates: approve(0)})
266+ if err != nil {
267+ t.Fatal(err)
268+ }
269+ st, err := child.RunFrom(context.Background(), "rollback")
270+ child.Log.Close()
271+ if err != nil {
272+ t.Fatal(err)
273+ }
274+ if st.Status != StatusSuccess || !slices.Equal(executed(st), []string{"rollback"}) {
275+ t.Fatalf("child %s %v %+v", st.Status, executed(st), st.End)
276+ }
277+ // it saw the parent's state; the parent is unchanged
278+ b, _ := os.ReadFile(filepath.Join(runs, "b", "events.ndjson"))
279+ if !strings.Contains(string(b), `"content":"deployed"`) {
280+ t.Errorf("child did not read the parent's $CI_STATE:\n%s", b)
281+ }
282+ if again, _ := Load(Dir(filepath.Join(runs, "a"))); again.Status != StatusSuccess || len(again.History) != len(pst.History) {
283+ t.Errorf("parent changed: %+v", again)
284+ }
285+}
+128,
-0
1@@ -0,0 +1,128 @@
2+package engine
3+
4+import (
5+ "context"
6+ "errors"
7+ "fmt"
8+ "os"
9+ "path/filepath"
10+ "slices"
11+ "strings"
12+ "syscall"
13+ "time"
14+
15+ "git.kilimanjaro.io/ci/internal/lex"
16+)
17+
18+// Concurrency groups are flock(2) locks on <LocksDir>/<group>.lock. A run
19+// takes a group's lock when it first runs an action in that group and holds
20+// it until it ends or pauses at a gate (a run waiting for a person must not
21+// block others), taking it again when it next runs an action in the group.
22+
23+func (e *Engine) locksDir() string {
24+ if e.Opts.LocksDir != "" {
25+ return e.Opts.LocksDir
26+ }
27+ return filepath.Join(filepath.Dir(e.Dir.Path()), ".locks")
28+}
29+
30+// acquire blocks until this run holds group's lock, or ctx ends.
31+func (e *Engine) acquire(ctx context.Context, group string) error {
32+ if _, held := e.locks[group]; held {
33+ return nil
34+ }
35+ if err := os.MkdirAll(e.locksDir(), 0o755); err != nil {
36+ return err
37+ }
38+ f, err := os.OpenFile(filepath.Join(e.locksDir(), group+".lock"), os.O_CREATE|os.O_RDWR, 0o644)
39+ if err != nil {
40+ return err
41+ }
42+ told := false
43+ for {
44+ err := syscall.Flock(int(f.Fd()), syscall.LOCK_EX|syscall.LOCK_NB)
45+ if err == nil {
46+ break
47+ }
48+ if !errors.Is(err, syscall.EWOULDBLOCK) {
49+ f.Close()
50+ return fmt.Errorf("lock %s: %w", group, err)
51+ }
52+ if !told {
53+ holder, _ := os.ReadFile(f.Name())
54+ e.emit(lex.RunEvent{Kind: lex.KindLog, Stream: "stderr",
55+ Content: fmt.Sprintf("ci: waiting for concurrency group %s (held by run %s)", group, strings.TrimSpace(string(holder)))})
56+ told = true
57+ }
58+ select {
59+ case <-ctx.Done():
60+ f.Close()
61+ return ctx.Err()
62+ case <-time.After(time.Second):
63+ }
64+ }
65+ if err := f.Truncate(0); err == nil {
66+ _, _ = f.WriteAt([]byte(e.st.Meta.Run+"\n"), 0)
67+ }
68+ if e.locks == nil {
69+ e.locks = map[string]*os.File{}
70+ }
71+ e.locks[group] = f
72+ return nil
73+}
74+
75+func (e *Engine) releaseLocks() {
76+ for g, f := range e.locks {
77+ _ = syscall.Flock(int(f.Fd()), syscall.LOCK_UN)
78+ f.Close()
79+ delete(e.locks, g)
80+ }
81+}
82+
83+// supersedeGroups returns the concurrency groups of g's options that ask for
84+// superseding.
85+func (e *Engine) supersedeGroups(g *PendingGate) []string {
86+ var out []string
87+ for _, o := range g.Options {
88+ if c := e.M.Actions[o.Action].Concurrency; c != nil && c.Supersede && !slices.Contains(out, c.Group) {
89+ out = append(out, c.Group)
90+ }
91+ }
92+ return out
93+}
94+
95+// supersedeOlder cancels older runs of the same repo and workflow that are
96+// waiting at a gate in one of g's superseding concurrency groups: only the
97+// newest deploy is worth approving.
98+func (e *Engine) supersedeOlder(g *PendingGate) {
99+ groups := e.supersedeGroups(g)
100+ if len(groups) == 0 {
101+ return
102+ }
103+ runs := filepath.Dir(e.Dir.Path())
104+ ents, err := os.ReadDir(runs)
105+ if err != nil {
106+ return
107+ }
108+ for _, ent := range ents {
109+ d := Dir(filepath.Join(runs, ent.Name()))
110+ if d == e.Dir || !ent.IsDir() {
111+ continue
112+ }
113+ st, err := Load(d)
114+ if err != nil || st.Status != StatusPaused || st.Gate == nil ||
115+ st.Meta.Repo != e.st.Meta.Repo || st.Meta.Workflow != e.st.Meta.Workflow || st.Created >= e.st.Created {
116+ continue
117+ }
118+ other, err := Open(d, nil, Options{})
119+ if err != nil {
120+ continue
121+ }
122+ if shared := other.supersedeGroups(st.Gate); slices.ContainsFunc(shared, func(s string) bool { return slices.Contains(groups, s) }) {
123+ other.emit(lex.RunEvent{Kind: lex.KindGate, Gate: other.gateRecord(st.Gate, lex.GateSuperseded, Choice{Actor: "run:" + e.st.Meta.Run})})
124+ _, _ = other.finish(End{StatusCancelled, lex.ReasonSuper, "superseded by run " + e.st.Meta.Run}, "")
125+ e.emit(lex.RunEvent{Kind: lex.KindLog, Stream: "stderr", Content: "ci: superseded run " + st.Meta.Run})
126+ }
127+ other.Log.Close()
128+ }
129+}
+29,
-9
1@@ -75,6 +75,10 @@ func (d Decision) String() string {
2 type Router struct {
3 M *manifest.Manifest
4 Now func() time.Time
5+ // Detached is set for a manual action run against an earlier run: it
6+ // runs the action (and whatever its result routes to) and ends with
7+ // that action's status instead of resuming the happy path.
8+ Detached bool
9 }
10
11 func (r *Router) now() time.Time {
12@@ -101,7 +105,11 @@ func (r *Router) Start() Decision {
13 // Begin starts a run at an arbitrary action (ci action), as if a person had
14 // chosen it: a declared gate still applies.
15 func (r *Router) Begin(action string) Decision {
16- return r.reach(Pending{Action: action, Then: r.defaultThen(action), Reason: lex.ReasonManual})
17+ then := r.defaultThen(action)
18+ if r.Detached {
19+ then = lex.ThenStop
20+ }
21+ return r.reach(Pending{Action: action, Then: then, Reason: lex.ReasonManual})
22 }
23
24 // reach routes to p without a result asking for it, applying the action's
25@@ -157,15 +165,15 @@ func (r *Router) After(st *State, finished Pending, res lex.Result, resultFile s
26 opts[i] = o
27 }
28 if r.needsGate(opts, res.Gated) {
29- prompt := ""
30+ prompt, _, _ := strings.Cut(strings.TrimSpace(res.Summary), "\n")
31 if len(opts) == 1 {
32- prompt = opts[0].Prompt
33- if prompt == "" {
34- if g := r.M.Actions[opts[0].Action].Gate; g != nil && g.Prompt != "" {
35- prompt = g.Prompt
36- } else {
37- prompt = fmt.Sprintf("Run %s?", r.M.Label(opts[0].Action))
38- }
39+ switch g := r.M.Actions[opts[0].Action].Gate; {
40+ case opts[0].Prompt != "":
41+ prompt = opts[0].Prompt
42+ case g != nil && g.Prompt != "":
43+ prompt = g.Prompt
44+ default:
45+ prompt = fmt.Sprintf("Run %s?", r.M.Label(opts[0].Action))
46 }
47 }
48 return Decision{Gate: r.gate(PendingGate{
49@@ -193,6 +201,18 @@ func (r *Router) After(st *State, finished Pending, res lex.Result, resultFile s
50 // continueOrStop applies finished's `then` after it succeeded.
51 func (r *Router) continueOrStop(st *State, finished Pending) Decision {
52 if finished.Then == lex.ThenStop {
53+ if r.Detached {
54+ return Decision{End: &End{StatusSuccess, lex.ReasonStop, finished.Action + " done"}}
55+ }
56+ if _, onPath := r.M.IsStep(finished.Action); !onPath {
57+ // an off-path action ending the run means the happy path was
58+ // diverted (a rollback), even if its last step had started
59+ detail := "diverted: stopped after " + finished.Action
60+ if st.Cursor < len(r.M.Steps) {
61+ detail += ", before " + r.M.Steps[st.Cursor]
62+ }
63+ return Decision{End: &End{StatusFailure, lex.ReasonStop, detail}}
64+ }
65 if st.Cursor >= len(r.M.Steps) {
66 return Decision{End: &End{StatusSuccess, lex.ReasonStop, "stopped after " + finished.Action + " at the end of the happy path"}}
67 }
+15,
-0
1@@ -216,6 +216,21 @@ func TestDismiss(t *testing.T) {
2 mustEnd(t, r.Expire(declared), StatusCancelled)
3 }
4
5+func TestRollbackAfterLastStepIsFailure(t *testing.T) {
6+ // health-check is the last step here; its failure routes to rollback
7+ m := newRouter(t).M
8+ m.Steps = []string{"test", "health-check"}
9+ r := &Router{M: m}
10+ st := &State{Cursor: 2}
11+ p := mustRun(t, r.After(st, Pending{Action: "health-check", Then: "continue"}, res("failure", `"notify"`), "r"), "notify", "stop")
12+ e := mustEnd(t, r.After(st, p, res("success", ""), "r2"), StatusFailure)
13+ if !strings.Contains(e.Detail, "diverted") {
14+ t.Errorf("detail %q", e.Detail)
15+ }
16+ // a happy-path step that stops at the end is a success
17+ mustEnd(t, r.After(st, Pending{Action: "health-check", Then: "stop"}, res("success", ""), "r3"), StatusSuccess)
18+}
19+
20 func TestExplicitStop(t *testing.T) {
21 r := newRouter(t)
22 mustEnd(t, r.After(&State{}, Pending{Action: "test"}, res("success", `"stop"`), "r"), StatusSuccess)
+4,
-1
1@@ -67,7 +67,10 @@ type Meta struct {
2 Host string `json:"host"` // local | runner
3 Root string `json:"root"` // main checkout
4 Dirty bool `json:"dirty,omitempty"` // local working copy with changes
5- // Checkouts are the extra repos (manifest repos:), by name.
6+ // Parent is the run directory a manual action run belongs to (ci action
7+ // --run); it shares that run's checkouts and $CI_STATE.
8+ Parent string `json:"parent,omitempty"`
9+ // Checkouts are the repos of the run, main first.
10 Checkouts []Checkout `json:"checkouts,omitempty"`
11 }
12
+1,
-0
1@@ -213,6 +213,7 @@ const (
2 ReasonExpired = "expired"
3 ReasonUntil = "until"
4 ReasonCancel = "cancelled"
5+ ReasonSuper = "superseded"
6 )
7
8 // Transition is run.rtw.ci.transition.
1@@ -56,7 +56,8 @@
2 "limit",
3 "expired",
4 "until",
5- "cancelled"
6+ "cancelled",
7+ "superseded"
8 ]
9 },
10 "detail": {