test(ci): cover triggers, cancel, re-run, matrix, restart and the merge rule

End-to-end tests with a fake runner over the real handlers: filters
(branches, paths, tags, more than three tags, skip directives, invalid
workflow files, deleted refs), cancel through WatchJob with the server's
enforcement, lost runner, re-run (all jobs and failed jobs only), matrix
with fail-fast and always(), unsupported features, queue timeout, runner
registration rules, restart of the store with running and queued jobs,
repository deletion cleanup, events on the repository topic, merges with
and without checks. Unit tests for the merge rule, check combining, labels,
step ranges, job ids, output limits and token shapes.

commit: 29932f1f99e3b70be413919982d38d52644eb66a
author: brice · 10d ago · 2026-10-12 18:23 UTC
parent: 5fbf954
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

8 files · +771 -17

M internal/adapters/rpc/ci.go +6 -1


@@ -132,7 +132,11 @@ func (s runnerService) Heartbeat(ctx context.Context, req connect.Request[pb.H
132	132		}
133	133
134	134		func (s runnerService) WatchJob(ctx context.Context, req connect.Request[pb.WatchJobRequest], stream *connect.ServerStream[pb.WatchJobResponse]) error {
135		-	err := s.runners.WatchJob(ctx, secretFrom(ctx), req.Msg.GetJobId(), func(sig ports.JobSignal) error {
	135	+	// A client's call returns when the response begins: send the headers now,
	136	+	// not with the first message, which may be 30 seconds away. Connect
	137	+	// flushes them for a nil message; a failure shows on the first real send.
	138	+	ready := func() { _ = stream.Send(nil) }
	139	+	err := s.runners.WatchJob(ctx, secretFrom(ctx), req.Msg.GetJobId(), ready, func(sig ports.JobSignal) error {
136	140		switch {
137	141		case sig.Cancel != nil:
138	142		return stream.Send(&pb.WatchJobResponse{Msg: &pb.WatchJobResponse_Cancel{Cancel: &pb.CancelRequest{

@@ -233,6 +237,7 @@ func (s runService) GetJobLog(ctx context.Context, req connect.Request[pb.GetJ
233	237		err = s.runs.GetJobLog(ctx, caller(ctx), name, ports.GetJobLogParams{
234	238		JobID: req.Msg.GetJobId(), Attempt: req.Msg.GetAttempt(), Offset: req.Msg.GetOffset(), Follow: req.Msg.GetFollow(),
235	239		Client: clientFromContext(ctx),
	240	+	Ready: func() { _ = stream.Send(nil) },
236	241		}, func(b ports.LogBatch) error {
237	242		out := &pb.GetJobLogResponse{Offset: b.Offset, Lines: make([]*pb.JobLogLine, 0, len(b.Lines))}
238	243		for _, l := range b.Lines {

 }
 func (s *runnerService) WatchJob(ctx context.Context, req *connect.Request[pb.WatchJobRequest], stream *connect.ServerStream[pb.WatchJobResponse]) error {
-	err := s.runners.WatchJob(ctx, secretFrom(ctx), req.Msg.GetJobId(), func(sig ports.JobSignal) error {
+	// A client's call returns when the response begins: send the headers now,
+	// not with the first message, which may be 30 seconds away. Connect
+	// flushes them for a nil message; a failure shows on the first real send.
+	ready := func() { _ = stream.Send(nil) }
+	err := s.runners.WatchJob(ctx, secretFrom(ctx), req.Msg.GetJobId(), ready, func(sig ports.JobSignal) error {
 		switch {
 		case sig.Cancel != nil:
 			return stream.Send(&pb.WatchJobResponse{Msg: &pb.WatchJobResponse_Cancel{Cancel: &pb.CancelRequest{
 	err = s.runs.GetJobLog(ctx, caller(ctx), name, ports.GetJobLogParams{
 		JobID: req.Msg.GetJobId(), Attempt: req.Msg.GetAttempt(), Offset: req.Msg.GetOffset(), Follow: req.Msg.GetFollow(),
 		Client: clientFromContext(ctx),
+		Ready:  func() { _ = stream.Send(nil) },
 	}, func(b ports.LogBatch) error {
 		out := &pb.GetJobLogResponse{Offset: b.Offset, Lines: make([]*pb.JobLogLine, 0, len(b.Lines))}
 		for _, l := range b.Lines {

M internal/app/ci_e2e_test.go +8 -4


@@ -82,16 +82,20 @@ func (c ciHarness) registerRunner(name string, labels ...string) fakeRunner {
82	82		Name: name, Labels: labels, Version: "devel", Isolation: pb.RunnerIsolation_RUNNER_ISOLATION_VM, Arch: "x64", Os: "linux",
83	83		}))
84	84		require.NoError(c.t, err)
85		-	return &fakeRunner{c: c, id: res.Msg.GetRunnerId(), api: c.api(res.Msg.GetRunnerToken())}
	85	+	return &fakeRunner{c: c, id: res.Msg.GetRunnerId(), secret: res.Msg.GetRunnerToken(), api: c.api(res.Msg.GetRunnerToken())}
86	86		}
87	87
88	88		// fakeRunner is the runner side of the protocol, as a test.
89	89		type fakeRunner struct {
90		-	c *ciHarness
91		-	id string
92		-	api *clients
	90	+	c *ciHarness
	91	+	id string
	92	+	secret string
	93	+	api *clients
93	94		}
94	95
	96	+	// runnerSecret is the runner token of a fake runner.
	97	+	func (c ciHarness) runnerSecret(r fakeRunner) string { return r.secret }
	98	+
95	99		// fetch polls once for a job.
96	100		func (r fakeRunner) fetch(pollID string) (pb.Job, error) {
97	101		res, err := r.api.runner.FetchJob(r.c.ctx, connect.NewRequest(&pb.FetchJobRequest{PollId: pollID}))

 		Name: name, Labels: labels, Version: "devel", Isolation: pb.RunnerIsolation_RUNNER_ISOLATION_VM, Arch: "x64", Os: "linux",
 	}))
 	require.NoError(c.t, err)
-	return &fakeRunner{c: c, id: res.Msg.GetRunnerId(), api: c.api(res.Msg.GetRunnerToken())}
+	return &fakeRunner{c: c, id: res.Msg.GetRunnerId(), secret: res.Msg.GetRunnerToken(), api: c.api(res.Msg.GetRunnerToken())}
 }
 // fakeRunner is the runner side of the protocol, as a test.
 type fakeRunner struct {
-	c   *ciHarness
+	c      *ciHarness
-	id  string
+	id     string
-	api *clients
+	secret string
+	api    *clients
 }
+// runnerSecret is the runner token of a fake runner.
+func (c *ciHarness) runnerSecret(r *fakeRunner) string { return r.secret }
 // fetch polls once for a job.
 func (r *fakeRunner) fetch(pollID string) (*pb.Job, error) {
 	res, err := r.api.runner.FetchJob(r.c.ctx, connect.NewRequest(&pb.FetchJobRequest{PollId: pollID}))

A internal/app/ci_flow_test.go +558 -0

@@ -0,0 +1,558 @@
	1	+	// SPDX-License-Identifier: AGPL-3.0-or-later
	2	+
	3	+	package app
	4	+
	5	+	import (
	6	+	"context"
	7	+	"fmt"
	8	+	"net/http/httptest"
	9	+	"os"
	10	+	"path/filepath"
	11	+	"strings"
	12	+	"testing"
	13	+	"time"
	14	+
	15	+	"connectrpc.com/connect"
	16	+	"github.com/stretchr/testify/assert"
	17	+	"github.com/stretchr/testify/require"
	18	+
	19	+	pb "bnema.dev/ermite-api/gen/ermite"
	20	+	"bnema.dev/ermite/internal/domain"
	21	+	"bnema.dev/ermite/internal/ports"
	22	+	"bnema.dev/ermite/internal/usecase"
	23	+	)
	24	+
	25	+	const oneJob = `on:
	26	+	push:
	27	+	%s
	28	+	jobs:
	29	+	j:
	30	+	runs-on: ubuntu-latest
	31	+	steps:
	32	+	- run: echo hi
	33	+	`
	34	+
	35	+	func workflowWith(filter string) string { return fmt.Sprintf(oneJob, filter) }
	36	+
	37	+	// AC3: filters, tags, skip directives, invalid files and deletions.
	38	+	func TestCITriggerRules(t *testing.T) {
	39	+	c := newCIHarness(t, nil)
	40	+	c.registerRunner("alpha", "ubuntu-latest")
	41	+	count := func() int { return len(c.runs()) }
	42	+	settle := func() { time.Sleep(300 * time.Millisecond) } // a push that runs nothing leaves nothing to wait for
	43	+
	44	+	// A workflow with a branch filter does not run for another branch.
	45	+	c.push("main", map[string]string{".github/workflows/only-main.yml": workflowWith("branches: [main]"), "src/a.go": "package a\n"}, "")
	46	+	c.waitFor("main runs", func() bool { return count() == 1 })
	47	+	c.push("other", map[string]string{"x.txt": "x\n"}, "")
	48	+	settle()
	49	+	assert.Equal(t, 1, count(), "branches: [main] does not run for other")
	50	+
	51	+	// A paths filter looks at the files of the push.
	52	+	c.push("main", map[string]string{".github/workflows/paths.yml": workflowWith("paths: ['docs/**']")}, "")
	53	+	c.waitFor("only-main runs again for the new commit", func() bool { return count() == 2 })
	54	+	c.push("main", map[string]string{"docs/readme.md": "doc\n"}, "")
	55	+	c.waitFor("both run for docs", func() bool { return count() == 4 })
	56	+	c.push("main", map[string]string{"src/b.go": "package b\n"}, "")
	57	+	c.waitFor("only-main runs for src", func() bool { return count() == 5 })
	58	+	settle()
	59	+	assert.Equal(t, 5, count(), "the paths filter kept paths.yml out")
	60	+
	61	+	// [skip ci] on the head commit.
	62	+	c.push("main", map[string]string{"src/c.go": "package c\n"}, "wip [skip ci]")
	63	+	settle()
	64	+	assert.Equal(t, 5, count())
	65	+
	66	+	// An invalid workflow file is a failed run without a job.
	67	+	c.push("main", map[string]string{".github/workflows/broken.yml": "on: push\njobs:\n j:\n steps: [\n"}, "")
	68	+	c.waitFor("the invalid file gives a run", func() bool {
	69	+	for _, r := range c.runs() {
	70	+	if r.GetWorkflowPath() == ".github/workflows/broken.yml" {
	71	+	return true
	72	+	}
	73	+	}
	74	+	return false
	75	+	})
	76	+	var broken *pb.Run
	77	+	for _, r := range c.runs() {
	78	+	if r.GetWorkflowPath() == ".github/workflows/broken.yml" {
	79	+	broken = r
	80	+	}
	81	+	}
	82	+	assert.Equal(t, pb.CiStatus_CI_STATUS_COMPLETED, broken.GetStatus())
	83	+	assert.Equal(t, pb.CiConclusion_CI_CONCLUSION_FAILURE, broken.GetConclusion())
	84	+	assert.NotEmpty(t, broken.GetError())
	85	+	assert.Empty(t, c.getRun(broken.GetId()).GetJobs(), "an invalid file has no job")
	86	+	sha := strings.TrimSpace(c.mustGit(c.wt, "rev-parse", "HEAD"))
	87	+	var failed []string
	88	+	for _, ck := range c.checks(sha) {
	89	+	if ck.GetConclusion() == pb.CiConclusion_CI_CONCLUSION_FAILURE {
	90	+	failed = append(failed, ck.GetName())
	91	+	}
	92	+	}
	93	+	assert.Equal(t, []string{".github/workflows/broken.yml"}, failed, "and a failed check on the commit")
	94	+
	95	+	// A push never fails for CI: the push above succeeded (mustGit). A deletion
	96	+	// runs nothing.
	97	+	before := count()
	98	+	c.mustGit(c.wt, "push", "-q", c.urlAuth(adminToken, "/proj.git"), ":other")
	99	+	settle()
	100	+	assert.Equal(t, before, count())
	101	+	}
	102	+
	103	+	// AC3: tags run tag-filtered workflows; more than 3 tags in one push run none.
	104	+	func TestCITags(t *testing.T) {
	105	+	c := newCIHarness(t, nil)
	106	+	c.registerRunner("alpha", "ubuntu-latest")
	107	+	c.push("main", map[string]string{".github/workflows/rel.yml": workflowWith("tags: ['v*']")}, "")
	108	+	settle := func() { time.Sleep(300 * time.Millisecond) }
	109	+	settle()
	110	+	assert.Empty(t, c.runs(), "a tags-only workflow does not run for a branch")
	111	+
	112	+	c.mustGit(c.wt, "tag", "v1")
	113	+	c.mustGit(c.wt, "push", "-q", c.urlAuth(adminToken, "/proj.git"), "v1")
	114	+	c.waitFor("a tag runs", func() bool { return len(c.runs()) == 1 })
	115	+	assert.Equal(t, "refs/tags/v1", c.runs()[0].GetRef())
	116	+
	117	+	for _, tag := range []string{"v2", "v3", "v4", "v5"} {
	118	+	c.mustGit(c.wt, "tag", tag)
	119	+	}
	120	+	c.mustGit(c.wt, "push", "-q", c.urlAuth(adminToken, "/proj.git"), "v2", "v3", "v4", "v5")
	121	+	settle()
	122	+	assert.Len(t, c.runs(), 1, "more than 3 tags in one push run nothing")
	123	+	}
	124	+
	125	+	// AC6: cancelling a run asks the runner to stop through WatchJob; a runner that
	126	+	// does not answer is ended by the server.
	127	+	func TestCICancelReachesTheRunnerAndIsEnforced(t *testing.T) {
	128	+	c := newCIHarness(t, func(cfg *usecase.CIConfig) {
	129	+	cfg.CancelGrace, cfg.Tick, cfg.PollTimeout = 400time.Millisecond, 20time.Millisecond, 100*time.Millisecond
	130	+	})
	131	+	runner := c.registerRunner("alpha", "ubuntu-latest")
	132	+	c.push("main", map[string]string{".github/workflows/ci.yml": ciWorkflow}, "")
	133	+	job := runner.await()
	134	+	api := runner.jobAPI(job)
	135	+	_, err := api.runner.Heartbeat(c.ctx, connect.NewRequest(&pb.HeartbeatRequest{JobId: job.GetJobId()}))
	136	+	require.NoError(t, err)
	137	+
	138	+	watch, err := api.runner.WatchJob(c.ctx, connect.NewRequest(&pb.WatchJobRequest{JobId: job.GetJobId()}))
	139	+	require.NoError(t, err)
	140	+	run := c.runs()[0]
	141	+
	142	+	// A reader without ci:write cannot cancel.
	143	+	reader := c.api(c.token(scopes(domain.ScopeRepoRead)))
	144	+	_, err = reader.run.CancelRun(c.ctx, connect.NewRequest(&pb.CancelRunRequest{Repo: "proj", RunId: run.GetId()}))
	145	+	assert.Equal(t, connect.CodePermissionDenied, ciErrCode(err))
	146	+
	147	+	_, err = c.root.run.CancelRun(c.ctx, connect.NewRequest(&pb.CancelRunRequest{Repo: "proj", RunId: run.GetId()}))
	148	+	require.NoError(t, err)
	149	+	// Cancelling again changes nothing.
	150	+	_, err = c.root.run.CancelRun(c.ctx, connect.NewRequest(&pb.CancelRunRequest{Repo: "proj", RunId: run.GetId()}))
	151	+	require.NoError(t, err)
	152	+
	153	+	var cancel *pb.CancelRequest
	154	+	for watch.Receive() {
	155	+	if m := watch.Msg().GetCancel(); m != nil {
	156	+	cancel = m
	157	+	break
	158	+	}
	159	+	}
	160	+	require.NotNil(t, cancel, "the runner is told")
	161	+	assert.Equal(t, pb.CancelReason_CANCEL_REASON_USER, cancel.GetReason())
	162	+
	163	+	// The queued job that needs it never ran: cancelled at once. The running
	164	+	// one is ended by the server after the grace period.
	165	+	c.waitFor("the server ends the job", func() bool {
	166	+	return c.runs()[0].GetStatus() == pb.CiStatus_CI_STATUS_COMPLETED
	167	+	})
	168	+	detail := c.getRun(run.GetId())
	169	+	assert.Equal(t, pb.CiConclusion_CI_CONCLUSION_CANCELLED, detail.GetRun().GetConclusion())
	170	+	for _, j := range detail.GetJobs() {
	171	+	assert.Equal(t, pb.CiConclusion_CI_CONCLUSION_CANCELLED, j.GetConclusion(), j.GetKey())
	172	+	}
	173	+	assert.Contains(t, detail.GetJobs()[0].GetError(), "did not answer")
	174	+
	175	+	// The runner's late CompleteJob returns the server's result.
	176	+	res, err := api.runner.CompleteJob(c.ctx, connect.NewRequest(&pb.CompleteJobRequest{
	177	+	JobId: job.GetJobId(), Conclusion: pb.CiConclusion_CI_CONCLUSION_CANCELLED,
	178	+	}))
	179	+	require.NoError(t, err)
	180	+	assert.False(t, res.Msg.GetApplied())
	181	+
	182	+	// A completed run cannot be cancelled.
	183	+	_, err = c.root.run.CancelRun(c.ctx, connect.NewRequest(&pb.CancelRunRequest{Repo: "proj", RunId: run.GetId()}))
	184	+	assert.Equal(t, connect.CodeFailedPrecondition, ciErrCode(err))
	185	+	}
	186	+
	187	+	// A running job without heartbeats fails with GitHub's lost-runner message.
	188	+	func TestCILostRunnerFailsTheJob(t *testing.T) {
	189	+	c := newCIHarness(t, func(cfg *usecase.CIConfig) {
	190	+	cfg.HeartbeatDeadline, cfg.Tick, cfg.PollTimeout = 300time.Millisecond, 20time.Millisecond, 100*time.Millisecond
	191	+	})
	192	+	runner := c.registerRunner("alpha", "ubuntu-latest")
	193	+	c.push("main", map[string]string{".github/workflows/ci.yml": strings.Split(ciWorkflow, " test:")[0]}, "")
	194	+	job := runner.await()
	195	+	_, err := runner.jobAPI(job).runner.Heartbeat(c.ctx, connect.NewRequest(&pb.HeartbeatRequest{JobId: job.GetJobId()}))
	196	+	require.NoError(t, err)
	197	+
	198	+	c.waitFor("the job is lost", func() bool { return c.runs()[0].GetStatus() == pb.CiStatus_CI_STATUS_COMPLETED })
	199	+	j := c.getRun(c.runs()[0].GetId()).GetJobs()[0]
	200	+	assert.Equal(t, pb.CiConclusion_CI_CONCLUSION_FAILURE, j.GetConclusion())
	201	+	assert.Contains(t, j.GetError(), "The self-hosted runner: alpha lost communication with the server")
	202	+	}
	203	+
	204	+	// AC6: a re-run is a new attempt of the same run; the failed-only variant keeps
	205	+	// the jobs that passed.
	206	+	func TestCIRerun(t *testing.T) {
	207	+	c := newCIHarness(t, nil)
	208	+	runner := c.registerRunner("alpha", "ubuntu-latest")
	209	+	c.push("main", map[string]string{".github/workflows/ci.yml": ciWorkflow}, "")
	210	+	run := func() *pb.Run { return c.runs()[0] }
	211	+
	212	+	build := runner.await()
	213	+	runner.runJob(build, pb.CiConclusion_CI_CONCLUSION_SUCCESS, nil, "ok")
	214	+	test := runner.await()
	215	+	runner.runJob(test, pb.CiConclusion_CI_CONCLUSION_FAILURE, nil, "red")
	216	+	c.waitFor("the run fails", func() bool { return run().GetStatus() == pb.CiStatus_CI_STATUS_COMPLETED })
	217	+	assert.Equal(t, pb.CiConclusion_CI_CONCLUSION_FAILURE, run().GetConclusion())
	218	+	id, number := run().GetId(), run().GetNumber()
	219	+
	220	+	// A running run cannot be re-run; a completed one can, by who has ci:write.
	221	+	reader := c.api(c.token(scopes(domain.ScopeRepoRead)))
	222	+	_, err := reader.run.RerunRun(c.ctx, connect.NewRequest(&pb.RerunRunRequest{Repo: "proj", RunId: id}))
	223	+	assert.Equal(t, connect.CodePermissionDenied, ciErrCode(err))
	224	+
	225	+	res, err := c.root.run.RerunRun(c.ctx, connect.NewRequest(&pb.RerunRunRequest{Repo: "proj", RunId: id, FailedOnly: true}))
	226	+	require.NoError(t, err)
	227	+	assert.Equal(t, id, res.Msg.GetRun().GetId())
	228	+	assert.Equal(t, number, res.Msg.GetRun().GetNumber())
	229	+	assert.EqualValues(t, 2, res.Msg.GetRun().GetAttempt())
	230	+
	231	+	// build keeps its attempt-1 result; only test runs again.
	232	+	again := runner.await()
	233	+	assert.Equal(t, "test", again.GetJobKey())
	234	+	assert.EqualValues(t, 2, again.GetAttempt())
	235	+	runner.runJob(again, pb.CiConclusion_CI_CONCLUSION_SUCCESS, nil, "green")
	236	+	c.waitFor("attempt 2 passes", func() bool {
	237	+	r := run()
	238	+	return r.GetAttempt() == 2 && r.GetStatus() == pb.CiStatus_CI_STATUS_COMPLETED
	239	+	})
	240	+	assert.Equal(t, pb.CiConclusion_CI_CONCLUSION_SUCCESS, run().GetConclusion())
	241	+	detail := c.getRun(id)
	242	+	require.Len(t, detail.GetJobs(), 2)
	243	+	assert.EqualValues(t, 1, detail.GetJobs()[0].GetAttempt(), "build was kept from attempt 1")
	244	+	assert.EqualValues(t, 2, detail.GetJobs()[1].GetAttempt())
	245	+
	246	+	// Attempt 1 is still readable, with its failure.
	247	+	first, err := c.root.run.GetRun(c.ctx, connect.NewRequest(&pb.GetRunRequest{Repo: "proj", RunId: id, Attempt: 1}))
	248	+	require.NoError(t, err)
	249	+	assert.Equal(t, pb.CiConclusion_CI_CONCLUSION_FAILURE, first.Msg.GetRun().GetConclusion())
	250	+	assert.EqualValues(t, 2, first.Msg.GetLatestAttempt())
	251	+
	252	+	// Re-running a run with nothing failed is refused with failed-only.
	253	+	_, err = c.root.run.RerunRun(c.ctx, connect.NewRequest(&pb.RerunRunRequest{Repo: "proj", RunId: id, FailedOnly: true}))
	254	+	assert.Equal(t, connect.CodeFailedPrecondition, ciErrCode(err))
	255	+	}
	256	+
	257	+	// A matrix is expanded when its job is ready, with fail-fast.
	258	+	func TestCIMatrixAndFailFast(t *testing.T) {
	259	+	c := newCIHarness(t, nil)
	260	+	runner := c.registerRunner("alpha", "ubuntu-latest")
	261	+	c.push("main", map[string]string{".github/workflows/m.yml": `on: push
	262	+	jobs:
	263	+	build:
	264	+	runs-on: ubuntu-latest
	265	+	strategy:
	266	+	matrix:
	267	+	go: ['1.26', '1.27']
	268	+	steps:
	269	+	- run: echo ${{ matrix.go }}
	270	+	after:
	271	+	needs: build
	272	+	if: always()
	273	+	runs-on: ubuntu-latest
	274	+	steps:
	275	+	- run: echo done
	276	+	`}, "")
	277	+	a := runner.await()
	278	+	b := runner.await()
	279	+	assert.Equal(t, "build", a.GetJobKey())
	280	+	assert.JSONEq(t, `{"go":"1.26"}`, a.GetMatrix())
	281	+	assert.JSONEq(t, `{"go":"1.27"}`, b.GetMatrix())
	282	+	assert.Contains(t, a.GetContexts(), `"job-total":2`)
	283	+
	284	+	jobs := c.getRun(c.runs()[0].GetId()).GetJobs()
	285	+	var names []string
	286	+	for _, j := range jobs {
	287	+	names = append(names, j.GetName())
	288	+	}
	289	+	assert.Equal(t, []string{"build (1.26)", "build (1.27)", "after"}, names)
	290	+
	291	+	// The first fails: fail-fast asks the second to stop.
	292	+	apiB := runner.jobAPI(b)
	293	+	_, err := apiB.runner.Heartbeat(c.ctx, connect.NewRequest(&pb.HeartbeatRequest{JobId: b.GetJobId()}))
	294	+	require.NoError(t, err)
	295	+	watch, err := apiB.runner.WatchJob(c.ctx, connect.NewRequest(&pb.WatchJobRequest{JobId: b.GetJobId()}))
	296	+	require.NoError(t, err)
	297	+	runner.runJob(a, pb.CiConclusion_CI_CONCLUSION_FAILURE, nil, "x")
	298	+	var reason pb.CancelReason
	299	+	for watch.Receive() {
	300	+	if m := watch.Msg().GetCancel(); m != nil {
	301	+	reason = m.GetReason()
	302	+	break
	303	+	}
	304	+	}
	305	+	assert.Equal(t, pb.CancelReason_CANCEL_REASON_FAIL_FAST, reason)
	306	+	_, err = apiB.runner.CompleteJob(c.ctx, connect.NewRequest(&pb.CompleteJobRequest{JobId: b.GetJobId(), Conclusion: pb.CiConclusion_CI_CONCLUSION_CANCELLED}))
	307	+	require.NoError(t, err)
	308	+
	309	+	// "after" has if: always(), so it runs once both ended.
	310	+	after := runner.await()
	311	+	assert.Equal(t, "after", after.GetJobKey())
	312	+	assert.Contains(t, after.GetContexts(), `"result":"failure"`)
	313	+	runner.runJob(after, pb.CiConclusion_CI_CONCLUSION_SUCCESS, nil, "done")
	314	+	c.waitFor("the run ends", func() bool { return c.runs()[0].GetStatus() == pb.CiStatus_CI_STATUS_COMPLETED })
	315	+	assert.Equal(t, pb.CiConclusion_CI_CONCLUSION_FAILURE, c.runs()[0].GetConclusion())
	316	+	}
	317	+
	318	+	// A job with a feature of a later phase fails clearly, never silently.
	319	+	func TestCIUnsupportedFeatureFailsTheJob(t *testing.T) {
	320	+	c := newCIHarness(t, nil)
	321	+	c.registerRunner("alpha", "ubuntu-latest")
	322	+	c.push("main", map[string]string{".github/workflows/u.yml": `on: push
	323	+	jobs:
	324	+	j:
	325	+	runs-on: ubuntu-latest
	326	+	steps:
	327	+	- uses: actions/checkout@v4
	328	+	`}, "")
	329	+	c.waitFor("the run fails", func() bool { return len(c.runs()) == 1 && c.runs()[0].GetStatus() == pb.CiStatus_CI_STATUS_COMPLETED })
	330	+	j := c.getRun(c.runs()[0].GetId()).GetJobs()[0]
	331	+	assert.Equal(t, pb.CiConclusion_CI_CONCLUSION_FAILURE, j.GetConclusion())
	332	+	assert.Equal(t, "not supported yet: uses: actions/checkout@v4", j.GetError())
	333	+	}
	334	+
	335	+	// A job for a label no runner has is "waiting for a runner" and fails after the
	336	+	// queue timeout.
	337	+	func TestCIQueueTimeout(t *testing.T) {
	338	+	c := newCIHarness(t, func(cfg usecase.CIConfig) { cfg.QueueTimeout, cfg.Tick = 300time.Millisecond, 20*time.Millisecond })
	339	+	c.registerRunner("alpha", "ubuntu-latest")
	340	+	c.push("main", map[string]string{".github/workflows/q.yml": strings.Replace(workflowWith("branches: [main]"), "ubuntu-latest", "gpu", 1)}, "")
	341	+	c.waitFor("the job fails", func() bool { return len(c.runs()) == 1 && c.runs()[0].GetStatus() == pb.CiStatus_CI_STATUS_COMPLETED })
	342	+	j := c.getRun(c.runs()[0].GetId()).GetJobs()[0]
	343	+	assert.Equal(t, pb.CiConclusion_CI_CONCLUSION_FAILURE, j.GetConclusion())
	344	+	assert.Contains(t, j.GetError(), "was not acquired")
	345	+	}
	346	+
	347	+	// The runner API validates what it is given.
	348	+	func TestCIRunnerRegistrationRules(t *testing.T) {
	349	+	c := newCIHarness(t, nil)
	350	+	reg := func(api *clients, name string, replace bool) error {
	351	+	_, err := api.runner.RegisterRunner(c.ctx, connect.NewRequest(&pb.RegisterRunnerRequest{
	352	+	Name: name, Version: "v1", Isolation: pb.RunnerIsolation_RUNNER_ISOLATION_VM, Arch: "x64", Os: "linux", Replace: replace,
	353	+	}))
	354	+	return err
	355	+	}
	356	+	assert.Equal(t, connect.CodeUnauthenticated, ciErrCode(reg(c.api(""), "a", false)))
	357	+	assert.Equal(t, connect.CodePermissionDenied, ciErrCode(reg(c.api(c.token(scopes(domain.ScopeRepoRead))), "a", false)))
	358	+	assert.Equal(t, connect.CodePermissionDenied, ciErrCode(reg(c.api(c.token(scopes(domain.ScopeRunnerRegister), "proj")), "a", false)), "a repository-restricted token cannot register a runner")
	359	+	reg1 := c.api(c.token(scopes(domain.ScopeRunnerRegister)))
	360	+	require.NoError(t, reg(reg1, "a", false))
	361	+	assert.Equal(t, connect.CodeAlreadyExists, ciErrCode(reg(reg1, "a", false)))
	362	+	require.NoError(t, reg(reg1, "a", true), "replace")
	363	+	assert.Equal(t, connect.CodeInvalidArgument, ciErrCode(reg(reg1, "bad name!", false)))
	364	+
	365	+	list, err := reg1.run.ListRunners(c.ctx, connect.NewRequest(&pb.ListRunnersRequest{}))
	366	+	require.NoError(t, err)
	367	+	require.Len(t, list.Msg.GetRunners(), 1)
	368	+	assert.Equal(t, []string{"linux", "self-hosted", "x64"}, list.Msg.GetRunners()[0].GetLabels(), "implicit labels, sorted")
	369	+	}
	370	+
	371	+	// AC2: after a restart of the server, running jobs go on with their tokens,
	372	+	// queued jobs stay queued, and a job whose runner kept quiet during the outage
	373	+	// is not failed for it.
	374	+	func TestCIRestartKeepsRunningAndQueuedJobs(t *testing.T) {
	375	+	c := newCIHarness(t, func(cfg usecase.CIConfig) { cfg.HeartbeatDeadline = 2 time.Second })
	376	+	runner := c.registerRunner("alpha", "ubuntu-latest")
	377	+	c.push("main", map[string]string{".github/workflows/ci.yml": ciWorkflow}, "")
	378	+	job := runner.await() // "build"; "test" waits for it
	379	+	api := runner.jobAPI(job)
	380	+	_, err := api.runner.Heartbeat(c.ctx, connect.NewRequest(&pb.HeartbeatRequest{JobId: job.GetJobId()}))
	381	+	require.NoError(t, err)
	382	+	_, err = api.runner.AppendLog(c.ctx, connect.NewRequest(&pb.AppendLogRequest{
	383	+	JobId: job.GetJobId(), Step: 2, Offset: 0, Lines: []*pb.LogLine{{Time: now(), Text: "before the restart"}},
	384	+	}))
	385	+	require.NoError(t, err)
	386	+
	387	+	// Stop the server and start another on the same data directory.
	388	+	cfg, hookExe := c.forge.cfg, binary(t)
	389	+	c.srv.Close()
	390	+	c.forge.Close(c.ctx)
	391	+	f2, err := newForge(c.ctx, cfg, options{hookExe: hookExe, ci: func(cc usecase.CIConfig) { cc.HeartbeatDeadline = 2 time.Second }})
	392	+	require.NoError(t, err)
	393	+	t.Cleanup(func() { f2.Close(context.WithoutCancel(c.ctx)) })
	394	+	srv := httptest.NewServer(f2.server.Handler())
	395	+	t.Cleanup(srv.Close)
	396	+	api2 := newClients(srv.URL+"/-/api", job.GetJobToken())
	397	+
	398	+	// The job token still works, and the log goes on from where it was.
	399	+	_, err = api2.runner.Heartbeat(c.ctx, connect.NewRequest(&pb.HeartbeatRequest{JobId: job.GetJobId()}))
	400	+	require.NoError(t, err, "the job token survives a restart")
	401	+	next, err := api2.runner.AppendLog(c.ctx, connect.NewRequest(&pb.AppendLogRequest{
	402	+	JobId: job.GetJobId(), Step: 2, Offset: 1, Lines: []*pb.LogLine{{Time: now(), Text: "after the restart"}},
	403	+	}))
	404	+	require.NoError(t, err)
	405	+	assert.EqualValues(t, 2, next.Msg.GetNextOffset())
	406	+
	407	+	root2 := newClients(srv.URL+"/-/api", adminToken)
	408	+	runs, err := root2.run.ListRuns(c.ctx, connect.NewRequest(&pb.ListRunsRequest{Repo: "proj"}))
	409	+	require.NoError(t, err)
	410	+	require.Len(t, runs.Msg.GetRuns(), 1)
	411	+	assert.Equal(t, pb.CiStatus_CI_STATUS_IN_PROGRESS, runs.Msg.GetRuns()[0].GetStatus())
	412	+
	413	+	// Finish the job on the new server: its dependant is planned and ready.
	414	+	_, err = api2.runner.CompleteJob(c.ctx, connect.NewRequest(&pb.CompleteJobRequest{
	415	+	JobId: job.GetJobId(), Conclusion: pb.CiConclusion_CI_CONCLUSION_SUCCESS,
	416	+	}))
	417	+	require.NoError(t, err)
	418	+	got := make(chan *pb.Job, 1)
	419	+	go func() {
	420	+	for range 50 {
	421	+	res, err := newClients(srv.URL+"/-/api", c.runnerSecret(runner)).runner.FetchJob(c.ctx, connect.NewRequest(&pb.FetchJobRequest{}))
	422	+	if err == nil && res.Msg.GetJob() != nil {
	423	+	got <- res.Msg.GetJob()
	424	+	return
	425	+	}
	426	+	}
	427	+	got <- nil
	428	+	}()
	429	+	select {
	430	+	case j := <-got:
	431	+	require.NotNil(t, j, "the runner token and the queue survive a restart")
	432	+	assert.Equal(t, "test", j.GetJobKey())
	433	+	case <-time.After(30 * time.Second):
	434	+	t.Fatal("no job after the restart")
	435	+	}
	436	+	}
	437	+
	438	+	// AC5: a head without any check merges, and so does one whose checks passed; a
	439	+	// merge runs the push workflows on the target.
	440	+	func TestCIMergeWithoutChecksAndWithPassingChecks(t *testing.T) {
	441	+	c := newCIHarness(t, nil)
	442	+	runner := c.registerRunner("alpha", "ubuntu-latest")
	443	+	c.push("main", map[string]string{"README": "x\n"}, "")
	444	+	headOf := func(branch string) string { return strings.TrimSpace(c.bareGit("proj", "rev-parse", branch)) }
	445	+	merge := func(n int64, head string) error {
	446	+	_, err := c.root.mr.MergeMergeRequest(c.ctx, connect.NewRequest(&pb.MergeMergeRequestRequest{Repo: "proj", Number: n, HeadId: head}))
	447	+	return err
	448	+	}
	449	+
	450	+	// No workflow on the branch: no check, the merge goes through.
	451	+	c.push("plain", map[string]string{"a.txt": "a\n"}, "")
	452	+	mr1, err := c.root.mr.CreateMergeRequest(c.ctx, connect.NewRequest(&pb.CreateMergeRequestRequest{Repo: "proj", Title: "plain", SourceBranch: "plain"}))
	453	+	require.NoError(t, err)
	454	+	require.NoError(t, merge(mr1.Msg.GetMergeRequest().GetNumber(), headOf("plain")))
	455	+
	456	+	// A branch with a workflow: merge after its check passed. The merge itself
	457	+	// triggers the workflow on main (GitHub runs push workflows on a merge).
	458	+	c.mustGit(c.wt, "checkout", "-q", "main")
	459	+	c.mustGit(c.wt, "pull", "-q", c.urlAuth(adminToken, "/proj.git"), "main")
	460	+	c.mustGit(c.wt, "checkout", "-q", "-b", "ci")
	461	+	c.writeFile(c.wt, ".github/workflows/ci.yml", []byte(strings.Replace(workflowWith("branches: [main, ci]"), "echo hi", "echo ci", 1)))
	462	+	c.mustGit(c.wt, "push", "-q", c.urlAuth(adminToken, "/proj.git"), "ci")
	463	+	head := headOf("ci")
	464	+	c.waitFor("the run on ci", func() bool { return len(c.checks(head)) == 1 })
	465	+	mr2, err := c.root.mr.CreateMergeRequest(c.ctx, connect.NewRequest(&pb.CreateMergeRequestRequest{Repo: "proj", Title: "ci", SourceBranch: "ci"}))
	466	+	require.NoError(t, err)
	467	+	job := runner.await()
	468	+	runner.runJob(job, pb.CiConclusion_CI_CONCLUSION_SUCCESS, nil, "ok")
	469	+	c.waitFor("the check passes", func() bool {
	470	+	cs := c.checks(head)
	471	+	return len(cs) == 1 && cs[0].GetConclusion() == pb.CiConclusion_CI_CONCLUSION_SUCCESS
	472	+	})
	473	+	require.NoError(t, merge(mr2.Msg.GetMergeRequest().GetNumber(), head))
	474	+
	475	+	// The fast-forward merge moved main to the same commit: the workflow runs
	476	+	// again for main, with the owner as the actor.
	477	+	c.waitFor("the merge triggers a run on main", func() bool {
	478	+	for _, r := range c.runs() {
	479	+	if r.GetRef() == "refs/heads/main" && r.GetSha() == head {
	480	+	return true
	481	+	}
	482	+	}
	483	+	return false
	484	+	})
	485	+	for _, r := range c.runs() {
	486	+	if r.GetRef() == "refs/heads/main" {
	487	+	assert.Equal(t, "owner", r.GetActor())
	488	+	}
	489	+	}
	490	+	}
	491	+
	492	+	// Repository deletion removes the runs, the checks and the logs; a repository
	493	+	// created again starts at run number 1.
	494	+	func TestCIRepositoryDeletionRemovesRunsAndLogs(t *testing.T) {
	495	+	c := newCIHarness(t, nil)
	496	+	runner := c.registerRunner("alpha", "ubuntu-latest")
	497	+	sha := c.push("main", map[string]string{".github/workflows/ci.yml": strings.Split(ciWorkflow, " test:")[0]}, "")
	498	+	job := runner.await()
	499	+	runner.runJob(job, pb.CiConclusion_CI_CONCLUSION_SUCCESS, nil, "line")
	500	+	logDir := filepath.Join(c.data, "ci", "logs")
	501	+	entries, err := os.ReadDir(logDir)
	502	+	require.NoError(t, err)
	503	+	require.NotEmpty(t, entries, "the log is a file under ci/logs")
	504	+
	505	+	_, err = c.root.repo.DeleteRepo(c.ctx, connect.NewRequest(&pb.DeleteRepoRequest{Repo: "proj"}))
	506	+	require.NoError(t, err)
	507	+	_, err = c.root.run.ListRuns(c.ctx, connect.NewRequest(&pb.ListRunsRequest{Repo: "proj"}))
	508	+	assert.Equal(t, connect.CodeNotFound, ciErrCode(err))
	509	+	entries, err = os.ReadDir(logDir)
	510	+	require.NoError(t, err)
	511	+	assert.Empty(t, entries, "the logs are removed with the repository")
	512	+
	513	+	_, err = c.forge.Repos.Create(c.ctx, domain.Root(), ports.CreateRepoParams{Name: c.name("proj"), Visibility: domain.VisibilityPrivate})
	514	+	require.NoError(t, err)
	515	+	assert.Empty(t, c.checks(sha), "no check survives the repository")
	516	+	c.push("main", map[string]string{".github/workflows/ci.yml": strings.Split(ciWorkflow, " test:")[0] + "# again\n"}, "again")
	517	+	c.waitFor("a run", func() bool { return len(c.runs()) == 1 })
	518	+	assert.EqualValues(t, 1, c.runs()[0].GetNumber(), "run numbers start again with the repository")
	519	+	assert.Greater(t, c.runs()[0].GetId(), int64(1), "run ids are never reused")
	520	+	}
	521	+
	522	+	// The events of a run and its jobs are published on the repository topic.
	523	+	func TestCIEventsAreStreamedOnTheRepositoryTopic(t *testing.T) {
	524	+	c := newCIHarness(t, nil)
	525	+	runner := c.registerRunner("alpha", "ubuntu-latest")
	526	+	s := c.watch(c.root, "", "repo:proj")
	527	+	c.push("main", map[string]string{".github/workflows/ci.yml": strings.Split(ciWorkflow, " test:")[0]}, "")
	528	+	job := runner.await()
	529	+	runner.runJob(job, pb.CiConclusion_CI_CONCLUSION_SUCCESS, nil, "x")
	530	+
	531	+	var runUpdates, jobUpdates int
	532	+	var lastRun *pb.Run
	533	+	deadline := time.After(10 * time.Second)
	534	+	for lastRun.GetStatus() != pb.CiStatus_CI_STATUS_COMPLETED {
	535	+	select {
	536	+	case ev := <-s.events:
	537	+	if r := ev.GetRunUpdated(); r != nil {
	538	+	runUpdates++
	539	+	if r.GetRun().GetVersion() >= lastRun.GetVersion() {
	540	+	lastRun = r.GetRun()
	541	+	}
	542	+	}
	543	+	if ev.GetJobUpdated() != nil {
	544	+	jobUpdates++
	545	+	}
	546	+	case <-deadline:
	547	+	t.Fatalf("no completed run event: %d run and %d job events", runUpdates, jobUpdates)
	548	+	}
	549	+	}
	550	+	assert.GreaterOrEqual(t, runUpdates, 2)
	551	+	assert.GreaterOrEqual(t, jobUpdates, 2)
	552	+
	553	+	// They are never sent on "home".
	554	+	home := c.watch(c.root, "", "home")
	555	+	_, err := c.root.run.RerunRun(c.ctx, connect.NewRequest(&pb.RerunRunRequest{Repo: "proj", RunId: lastRun.GetId()}))
	556	+	require.NoError(t, err)
	557	+	c.sentinel(t, home, "sentinel")
	558	+	}

@@ -0,0 +1,558 @@
	1	+	// SPDX-License-Identifier: AGPL-3.0-or-later
	2	+
	3	+	package app
	4	+
	5	+	import (
	6	+	"context"
	7	+	"fmt"
	8	+	"net/http/httptest"
	9	+	"os"
	10	+	"path/filepath"
	11	+	"strings"
	12	+	"testing"
	13	+	"time"
	14	+
	15	+	"connectrpc.com/connect"
	16	+	"github.com/stretchr/testify/assert"
	17	+	"github.com/stretchr/testify/require"
	18	+
	19	+	pb "bnema.dev/ermite-api/gen/ermite"
	20	+	"bnema.dev/ermite/internal/domain"
	21	+	"bnema.dev/ermite/internal/ports"
	22	+	"bnema.dev/ermite/internal/usecase"
	23	+	)
	24	+
	25	+	const oneJob = `on:
	26	+	push:
	27	+	%s
	28	+	jobs:
	29	+	j:
	30	+	runs-on: ubuntu-latest
	31	+	steps:
	32	+	- run: echo hi
	33	+	`
	34	+
	35	+	func workflowWith(filter string) string { return fmt.Sprintf(oneJob, filter) }
	36	+
	37	+	// AC3: filters, tags, skip directives, invalid files and deletions.
	38	+	func TestCITriggerRules(t *testing.T) {
	39	+	c := newCIHarness(t, nil)
	40	+	c.registerRunner("alpha", "ubuntu-latest")
	41	+	count := func() int { return len(c.runs()) }
	42	+	settle := func() { time.Sleep(300 * time.Millisecond) } // a push that runs nothing leaves nothing to wait for
	43	+
	44	+	// A workflow with a branch filter does not run for another branch.
	45	+	c.push("main", map[string]string{".github/workflows/only-main.yml": workflowWith("branches: [main]"), "src/a.go": "package a\n"}, "")
	46	+	c.waitFor("main runs", func() bool { return count() == 1 })
	47	+	c.push("other", map[string]string{"x.txt": "x\n"}, "")
	48	+	settle()
	49	+	assert.Equal(t, 1, count(), "branches: [main] does not run for other")
	50	+
	51	+	// A paths filter looks at the files of the push.
	52	+	c.push("main", map[string]string{".github/workflows/paths.yml": workflowWith("paths: ['docs/**']")}, "")
	53	+	c.waitFor("only-main runs again for the new commit", func() bool { return count() == 2 })
	54	+	c.push("main", map[string]string{"docs/readme.md": "doc\n"}, "")
	55	+	c.waitFor("both run for docs", func() bool { return count() == 4 })
	56	+	c.push("main", map[string]string{"src/b.go": "package b\n"}, "")
	57	+	c.waitFor("only-main runs for src", func() bool { return count() == 5 })
	58	+	settle()
	59	+	assert.Equal(t, 5, count(), "the paths filter kept paths.yml out")
	60	+
	61	+	// [skip ci] on the head commit.
	62	+	c.push("main", map[string]string{"src/c.go": "package c\n"}, "wip [skip ci]")
	63	+	settle()
	64	+	assert.Equal(t, 5, count())
	65	+
	66	+	// An invalid workflow file is a failed run without a job.
	67	+	c.push("main", map[string]string{".github/workflows/broken.yml": "on: push\njobs:\n j:\n steps: [\n"}, "")
	68	+	c.waitFor("the invalid file gives a run", func() bool {
	69	+	for _, r := range c.runs() {
	70	+	if r.GetWorkflowPath() == ".github/workflows/broken.yml" {
	71	+	return true
	72	+	}
	73	+	}
	74	+	return false
	75	+	})
	76	+	var broken *pb.Run
	77	+	for _, r := range c.runs() {
	78	+	if r.GetWorkflowPath() == ".github/workflows/broken.yml" {
	79	+	broken = r
	80	+	}
	81	+	}
	82	+	assert.Equal(t, pb.CiStatus_CI_STATUS_COMPLETED, broken.GetStatus())
	83	+	assert.Equal(t, pb.CiConclusion_CI_CONCLUSION_FAILURE, broken.GetConclusion())
	84	+	assert.NotEmpty(t, broken.GetError())
	85	+	assert.Empty(t, c.getRun(broken.GetId()).GetJobs(), "an invalid file has no job")
	86	+	sha := strings.TrimSpace(c.mustGit(c.wt, "rev-parse", "HEAD"))
	87	+	var failed []string
	88	+	for _, ck := range c.checks(sha) {
	89	+	if ck.GetConclusion() == pb.CiConclusion_CI_CONCLUSION_FAILURE {
	90	+	failed = append(failed, ck.GetName())
	91	+	}
	92	+	}
	93	+	assert.Equal(t, []string{".github/workflows/broken.yml"}, failed, "and a failed check on the commit")
	94	+
	95	+	// A push never fails for CI: the push above succeeded (mustGit). A deletion
	96	+	// runs nothing.
	97	+	before := count()
	98	+	c.mustGit(c.wt, "push", "-q", c.urlAuth(adminToken, "/proj.git"), ":other")
	99	+	settle()
	100	+	assert.Equal(t, before, count())
	101	+	}
	102	+
	103	+	// AC3: tags run tag-filtered workflows; more than 3 tags in one push run none.
	104	+	func TestCITags(t *testing.T) {
	105	+	c := newCIHarness(t, nil)
	106	+	c.registerRunner("alpha", "ubuntu-latest")
	107	+	c.push("main", map[string]string{".github/workflows/rel.yml": workflowWith("tags: ['v*']")}, "")
	108	+	settle := func() { time.Sleep(300 * time.Millisecond) }
	109	+	settle()
	110	+	assert.Empty(t, c.runs(), "a tags-only workflow does not run for a branch")
	111	+
	112	+	c.mustGit(c.wt, "tag", "v1")
	113	+	c.mustGit(c.wt, "push", "-q", c.urlAuth(adminToken, "/proj.git"), "v1")
	114	+	c.waitFor("a tag runs", func() bool { return len(c.runs()) == 1 })
	115	+	assert.Equal(t, "refs/tags/v1", c.runs()[0].GetRef())
	116	+
	117	+	for _, tag := range []string{"v2", "v3", "v4", "v5"} {
	118	+	c.mustGit(c.wt, "tag", tag)
	119	+	}
	120	+	c.mustGit(c.wt, "push", "-q", c.urlAuth(adminToken, "/proj.git"), "v2", "v3", "v4", "v5")
	121	+	settle()
	122	+	assert.Len(t, c.runs(), 1, "more than 3 tags in one push run nothing")
	123	+	}
	124	+
	125	+	// AC6: cancelling a run asks the runner to stop through WatchJob; a runner that
	126	+	// does not answer is ended by the server.
	127	+	func TestCICancelReachesTheRunnerAndIsEnforced(t *testing.T) {
	128	+	c := newCIHarness(t, func(cfg *usecase.CIConfig) {
	129	+	cfg.CancelGrace, cfg.Tick, cfg.PollTimeout = 400time.Millisecond, 20time.Millisecond, 100*time.Millisecond
	130	+	})
	131	+	runner := c.registerRunner("alpha", "ubuntu-latest")
	132	+	c.push("main", map[string]string{".github/workflows/ci.yml": ciWorkflow}, "")
	133	+	job := runner.await()
	134	+	api := runner.jobAPI(job)
	135	+	_, err := api.runner.Heartbeat(c.ctx, connect.NewRequest(&pb.HeartbeatRequest{JobId: job.GetJobId()}))
	136	+	require.NoError(t, err)
	137	+
	138	+	watch, err := api.runner.WatchJob(c.ctx, connect.NewRequest(&pb.WatchJobRequest{JobId: job.GetJobId()}))
	139	+	require.NoError(t, err)
	140	+	run := c.runs()[0]
	141	+
	142	+	// A reader without ci:write cannot cancel.
	143	+	reader := c.api(c.token(scopes(domain.ScopeRepoRead)))
	144	+	_, err = reader.run.CancelRun(c.ctx, connect.NewRequest(&pb.CancelRunRequest{Repo: "proj", RunId: run.GetId()}))
	145	+	assert.Equal(t, connect.CodePermissionDenied, ciErrCode(err))
	146	+
	147	+	_, err = c.root.run.CancelRun(c.ctx, connect.NewRequest(&pb.CancelRunRequest{Repo: "proj", RunId: run.GetId()}))
	148	+	require.NoError(t, err)
	149	+	// Cancelling again changes nothing.
	150	+	_, err = c.root.run.CancelRun(c.ctx, connect.NewRequest(&pb.CancelRunRequest{Repo: "proj", RunId: run.GetId()}))
	151	+	require.NoError(t, err)
	152	+
	153	+	var cancel *pb.CancelRequest
	154	+	for watch.Receive() {
	155	+	if m := watch.Msg().GetCancel(); m != nil {
	156	+	cancel = m
	157	+	break
	158	+	}
	159	+	}
	160	+	require.NotNil(t, cancel, "the runner is told")
	161	+	assert.Equal(t, pb.CancelReason_CANCEL_REASON_USER, cancel.GetReason())
	162	+
	163	+	// The queued job that needs it never ran: cancelled at once. The running
	164	+	// one is ended by the server after the grace period.
	165	+	c.waitFor("the server ends the job", func() bool {
	166	+	return c.runs()[0].GetStatus() == pb.CiStatus_CI_STATUS_COMPLETED
	167	+	})
	168	+	detail := c.getRun(run.GetId())
	169	+	assert.Equal(t, pb.CiConclusion_CI_CONCLUSION_CANCELLED, detail.GetRun().GetConclusion())
	170	+	for _, j := range detail.GetJobs() {
	171	+	assert.Equal(t, pb.CiConclusion_CI_CONCLUSION_CANCELLED, j.GetConclusion(), j.GetKey())
	172	+	}
	173	+	assert.Contains(t, detail.GetJobs()[0].GetError(), "did not answer")
	174	+
	175	+	// The runner's late CompleteJob returns the server's result.
	176	+	res, err := api.runner.CompleteJob(c.ctx, connect.NewRequest(&pb.CompleteJobRequest{
	177	+	JobId: job.GetJobId(), Conclusion: pb.CiConclusion_CI_CONCLUSION_CANCELLED,
	178	+	}))
	179	+	require.NoError(t, err)
	180	+	assert.False(t, res.Msg.GetApplied())
	181	+
	182	+	// A completed run cannot be cancelled.
	183	+	_, err = c.root.run.CancelRun(c.ctx, connect.NewRequest(&pb.CancelRunRequest{Repo: "proj", RunId: run.GetId()}))
	184	+	assert.Equal(t, connect.CodeFailedPrecondition, ciErrCode(err))
	185	+	}
	186	+
	187	+	// A running job without heartbeats fails with GitHub's lost-runner message.
	188	+	func TestCILostRunnerFailsTheJob(t *testing.T) {
	189	+	c := newCIHarness(t, func(cfg *usecase.CIConfig) {
	190	+	cfg.HeartbeatDeadline, cfg.Tick, cfg.PollTimeout = 300time.Millisecond, 20time.Millisecond, 100*time.Millisecond
	191	+	})
	192	+	runner := c.registerRunner("alpha", "ubuntu-latest")
	193	+	c.push("main", map[string]string{".github/workflows/ci.yml": strings.Split(ciWorkflow, " test:")[0]}, "")
	194	+	job := runner.await()
	195	+	_, err := runner.jobAPI(job).runner.Heartbeat(c.ctx, connect.NewRequest(&pb.HeartbeatRequest{JobId: job.GetJobId()}))
	196	+	require.NoError(t, err)
	197	+
	198	+	c.waitFor("the job is lost", func() bool { return c.runs()[0].GetStatus() == pb.CiStatus_CI_STATUS_COMPLETED })
	199	+	j := c.getRun(c.runs()[0].GetId()).GetJobs()[0]
	200	+	assert.Equal(t, pb.CiConclusion_CI_CONCLUSION_FAILURE, j.GetConclusion())
	201	+	assert.Contains(t, j.GetError(), "The self-hosted runner: alpha lost communication with the server")
	202	+	}
	203	+
	204	+	// AC6: a re-run is a new attempt of the same run; the failed-only variant keeps
	205	+	// the jobs that passed.
	206	+	func TestCIRerun(t *testing.T) {
	207	+	c := newCIHarness(t, nil)
	208	+	runner := c.registerRunner("alpha", "ubuntu-latest")
	209	+	c.push("main", map[string]string{".github/workflows/ci.yml": ciWorkflow}, "")
	210	+	run := func() *pb.Run { return c.runs()[0] }
	211	+
	212	+	build := runner.await()
	213	+	runner.runJob(build, pb.CiConclusion_CI_CONCLUSION_SUCCESS, nil, "ok")
	214	+	test := runner.await()
	215	+	runner.runJob(test, pb.CiConclusion_CI_CONCLUSION_FAILURE, nil, "red")
	216	+	c.waitFor("the run fails", func() bool { return run().GetStatus() == pb.CiStatus_CI_STATUS_COMPLETED })
	217	+	assert.Equal(t, pb.CiConclusion_CI_CONCLUSION_FAILURE, run().GetConclusion())
	218	+	id, number := run().GetId(), run().GetNumber()
	219	+
	220	+	// A running run cannot be re-run; a completed one can, by who has ci:write.
	221	+	reader := c.api(c.token(scopes(domain.ScopeRepoRead)))
	222	+	_, err := reader.run.RerunRun(c.ctx, connect.NewRequest(&pb.RerunRunRequest{Repo: "proj", RunId: id}))
	223	+	assert.Equal(t, connect.CodePermissionDenied, ciErrCode(err))
	224	+
	225	+	res, err := c.root.run.RerunRun(c.ctx, connect.NewRequest(&pb.RerunRunRequest{Repo: "proj", RunId: id, FailedOnly: true}))
	226	+	require.NoError(t, err)
	227	+	assert.Equal(t, id, res.Msg.GetRun().GetId())
	228	+	assert.Equal(t, number, res.Msg.GetRun().GetNumber())
	229	+	assert.EqualValues(t, 2, res.Msg.GetRun().GetAttempt())
	230	+
	231	+	// build keeps its attempt-1 result; only test runs again.
	232	+	again := runner.await()
	233	+	assert.Equal(t, "test", again.GetJobKey())
	234	+	assert.EqualValues(t, 2, again.GetAttempt())
	235	+	runner.runJob(again, pb.CiConclusion_CI_CONCLUSION_SUCCESS, nil, "green")
	236	+	c.waitFor("attempt 2 passes", func() bool {
	237	+	r := run()
	238	+	return r.GetAttempt() == 2 && r.GetStatus() == pb.CiStatus_CI_STATUS_COMPLETED
	239	+	})
	240	+	assert.Equal(t, pb.CiConclusion_CI_CONCLUSION_SUCCESS, run().GetConclusion())
	241	+	detail := c.getRun(id)
	242	+	require.Len(t, detail.GetJobs(), 2)
	243	+	assert.EqualValues(t, 1, detail.GetJobs()[0].GetAttempt(), "build was kept from attempt 1")
	244	+	assert.EqualValues(t, 2, detail.GetJobs()[1].GetAttempt())
	245	+
	246	+	// Attempt 1 is still readable, with its failure.
	247	+	first, err := c.root.run.GetRun(c.ctx, connect.NewRequest(&pb.GetRunRequest{Repo: "proj", RunId: id, Attempt: 1}))
	248	+	require.NoError(t, err)
	249	+	assert.Equal(t, pb.CiConclusion_CI_CONCLUSION_FAILURE, first.Msg.GetRun().GetConclusion())
	250	+	assert.EqualValues(t, 2, first.Msg.GetLatestAttempt())
	251	+
	252	+	// Re-running a run with nothing failed is refused with failed-only.
	253	+	_, err = c.root.run.RerunRun(c.ctx, connect.NewRequest(&pb.RerunRunRequest{Repo: "proj", RunId: id, FailedOnly: true}))
	254	+	assert.Equal(t, connect.CodeFailedPrecondition, ciErrCode(err))
	255	+	}
	256	+
	257	+	// A matrix is expanded when its job is ready, with fail-fast.
	258	+	func TestCIMatrixAndFailFast(t *testing.T) {
	259	+	c := newCIHarness(t, nil)
	260	+	runner := c.registerRunner("alpha", "ubuntu-latest")
	261	+	c.push("main", map[string]string{".github/workflows/m.yml": `on: push
	262	+	jobs:
	263	+	build:
	264	+	runs-on: ubuntu-latest
	265	+	strategy:
	266	+	matrix:
	267	+	go: ['1.26', '1.27']
	268	+	steps:
	269	+	- run: echo ${{ matrix.go }}
	270	+	after:
	271	+	needs: build
	272	+	if: always()
	273	+	runs-on: ubuntu-latest
	274	+	steps:
	275	+	- run: echo done
	276	+	`}, "")
	277	+	a := runner.await()
	278	+	b := runner.await()
	279	+	assert.Equal(t, "build", a.GetJobKey())
	280	+	assert.JSONEq(t, `{"go":"1.26"}`, a.GetMatrix())
	281	+	assert.JSONEq(t, `{"go":"1.27"}`, b.GetMatrix())
	282	+	assert.Contains(t, a.GetContexts(), `"job-total":2`)
	283	+
	284	+	jobs := c.getRun(c.runs()[0].GetId()).GetJobs()
	285	+	var names []string
	286	+	for _, j := range jobs {
	287	+	names = append(names, j.GetName())
	288	+	}
	289	+	assert.Equal(t, []string{"build (1.26)", "build (1.27)", "after"}, names)
	290	+
	291	+	// The first fails: fail-fast asks the second to stop.
	292	+	apiB := runner.jobAPI(b)
	293	+	_, err := apiB.runner.Heartbeat(c.ctx, connect.NewRequest(&pb.HeartbeatRequest{JobId: b.GetJobId()}))
	294	+	require.NoError(t, err)
	295	+	watch, err := apiB.runner.WatchJob(c.ctx, connect.NewRequest(&pb.WatchJobRequest{JobId: b.GetJobId()}))
	296	+	require.NoError(t, err)
	297	+	runner.runJob(a, pb.CiConclusion_CI_CONCLUSION_FAILURE, nil, "x")
	298	+	var reason pb.CancelReason
	299	+	for watch.Receive() {
	300	+	if m := watch.Msg().GetCancel(); m != nil {
	301	+	reason = m.GetReason()
	302	+	break
	303	+	}
	304	+	}
	305	+	assert.Equal(t, pb.CancelReason_CANCEL_REASON_FAIL_FAST, reason)
	306	+	_, err = apiB.runner.CompleteJob(c.ctx, connect.NewRequest(&pb.CompleteJobRequest{JobId: b.GetJobId(), Conclusion: pb.CiConclusion_CI_CONCLUSION_CANCELLED}))
	307	+	require.NoError(t, err)
	308	+
	309	+	// "after" has if: always(), so it runs once both ended.
	310	+	after := runner.await()
	311	+	assert.Equal(t, "after", after.GetJobKey())
	312	+	assert.Contains(t, after.GetContexts(), `"result":"failure"`)
	313	+	runner.runJob(after, pb.CiConclusion_CI_CONCLUSION_SUCCESS, nil, "done")
	314	+	c.waitFor("the run ends", func() bool { return c.runs()[0].GetStatus() == pb.CiStatus_CI_STATUS_COMPLETED })
	315	+	assert.Equal(t, pb.CiConclusion_CI_CONCLUSION_FAILURE, c.runs()[0].GetConclusion())
	316	+	}
	317	+
	318	+	// A job with a feature of a later phase fails clearly, never silently.
	319	+	func TestCIUnsupportedFeatureFailsTheJob(t *testing.T) {
	320	+	c := newCIHarness(t, nil)
	321	+	c.registerRunner("alpha", "ubuntu-latest")
	322	+	c.push("main", map[string]string{".github/workflows/u.yml": `on: push
	323	+	jobs:
	324	+	j:
	325	+	runs-on: ubuntu-latest
	326	+	steps:
	327	+	- uses: actions/checkout@v4
	328	+	`}, "")
	329	+	c.waitFor("the run fails", func() bool { return len(c.runs()) == 1 && c.runs()[0].GetStatus() == pb.CiStatus_CI_STATUS_COMPLETED })
	330	+	j := c.getRun(c.runs()[0].GetId()).GetJobs()[0]
	331	+	assert.Equal(t, pb.CiConclusion_CI_CONCLUSION_FAILURE, j.GetConclusion())
	332	+	assert.Equal(t, "not supported yet: uses: actions/checkout@v4", j.GetError())
	333	+	}
	334	+
	335	+	// A job for a label no runner has is "waiting for a runner" and fails after the
	336	+	// queue timeout.
	337	+	func TestCIQueueTimeout(t *testing.T) {
	338	+	c := newCIHarness(t, func(cfg usecase.CIConfig) { cfg.QueueTimeout, cfg.Tick = 300time.Millisecond, 20*time.Millisecond })
	339	+	c.registerRunner("alpha", "ubuntu-latest")
	340	+	c.push("main", map[string]string{".github/workflows/q.yml": strings.Replace(workflowWith("branches: [main]"), "ubuntu-latest", "gpu", 1)}, "")
	341	+	c.waitFor("the job fails", func() bool { return len(c.runs()) == 1 && c.runs()[0].GetStatus() == pb.CiStatus_CI_STATUS_COMPLETED })
	342	+	j := c.getRun(c.runs()[0].GetId()).GetJobs()[0]
	343	+	assert.Equal(t, pb.CiConclusion_CI_CONCLUSION_FAILURE, j.GetConclusion())
	344	+	assert.Contains(t, j.GetError(), "was not acquired")
	345	+	}
	346	+
	347	+	// The runner API validates what it is given.
	348	+	func TestCIRunnerRegistrationRules(t *testing.T) {
	349	+	c := newCIHarness(t, nil)
	350	+	reg := func(api *clients, name string, replace bool) error {
	351	+	_, err := api.runner.RegisterRunner(c.ctx, connect.NewRequest(&pb.RegisterRunnerRequest{
	352	+	Name: name, Version: "v1", Isolation: pb.RunnerIsolation_RUNNER_ISOLATION_VM, Arch: "x64", Os: "linux", Replace: replace,
	353	+	}))
	354	+	return err
	355	+	}
	356	+	assert.Equal(t, connect.CodeUnauthenticated, ciErrCode(reg(c.api(""), "a", false)))
	357	+	assert.Equal(t, connect.CodePermissionDenied, ciErrCode(reg(c.api(c.token(scopes(domain.ScopeRepoRead))), "a", false)))
	358	+	assert.Equal(t, connect.CodePermissionDenied, ciErrCode(reg(c.api(c.token(scopes(domain.ScopeRunnerRegister), "proj")), "a", false)), "a repository-restricted token cannot register a runner")
	359	+	reg1 := c.api(c.token(scopes(domain.ScopeRunnerRegister)))
	360	+	require.NoError(t, reg(reg1, "a", false))
	361	+	assert.Equal(t, connect.CodeAlreadyExists, ciErrCode(reg(reg1, "a", false)))
	362	+	require.NoError(t, reg(reg1, "a", true), "replace")
	363	+	assert.Equal(t, connect.CodeInvalidArgument, ciErrCode(reg(reg1, "bad name!", false)))
	364	+
	365	+	list, err := reg1.run.ListRunners(c.ctx, connect.NewRequest(&pb.ListRunnersRequest{}))
	366	+	require.NoError(t, err)
	367	+	require.Len(t, list.Msg.GetRunners(), 1)
	368	+	assert.Equal(t, []string{"linux", "self-hosted", "x64"}, list.Msg.GetRunners()[0].GetLabels(), "implicit labels, sorted")
	369	+	}
	370	+
	371	+	// AC2: after a restart of the server, running jobs go on with their tokens,
	372	+	// queued jobs stay queued, and a job whose runner kept quiet during the outage
	373	+	// is not failed for it.
	374	+	func TestCIRestartKeepsRunningAndQueuedJobs(t *testing.T) {
	375	+	c := newCIHarness(t, func(cfg usecase.CIConfig) { cfg.HeartbeatDeadline = 2 time.Second })
	376	+	runner := c.registerRunner("alpha", "ubuntu-latest")
	377	+	c.push("main", map[string]string{".github/workflows/ci.yml": ciWorkflow}, "")
	378	+	job := runner.await() // "build"; "test" waits for it
	379	+	api := runner.jobAPI(job)
	380	+	_, err := api.runner.Heartbeat(c.ctx, connect.NewRequest(&pb.HeartbeatRequest{JobId: job.GetJobId()}))
	381	+	require.NoError(t, err)
	382	+	_, err = api.runner.AppendLog(c.ctx, connect.NewRequest(&pb.AppendLogRequest{
	383	+	JobId: job.GetJobId(), Step: 2, Offset: 0, Lines: []*pb.LogLine{{Time: now(), Text: "before the restart"}},
	384	+	}))
	385	+	require.NoError(t, err)
	386	+
	387	+	// Stop the server and start another on the same data directory.
	388	+	cfg, hookExe := c.forge.cfg, binary(t)
	389	+	c.srv.Close()
	390	+	c.forge.Close(c.ctx)
	391	+	f2, err := newForge(c.ctx, cfg, options{hookExe: hookExe, ci: func(cc usecase.CIConfig) { cc.HeartbeatDeadline = 2 time.Second }})
	392	+	require.NoError(t, err)
	393	+	t.Cleanup(func() { f2.Close(context.WithoutCancel(c.ctx)) })
	394	+	srv := httptest.NewServer(f2.server.Handler())
	395	+	t.Cleanup(srv.Close)
	396	+	api2 := newClients(srv.URL+"/-/api", job.GetJobToken())
	397	+
	398	+	// The job token still works, and the log goes on from where it was.
	399	+	_, err = api2.runner.Heartbeat(c.ctx, connect.NewRequest(&pb.HeartbeatRequest{JobId: job.GetJobId()}))
	400	+	require.NoError(t, err, "the job token survives a restart")
	401	+	next, err := api2.runner.AppendLog(c.ctx, connect.NewRequest(&pb.AppendLogRequest{
	402	+	JobId: job.GetJobId(), Step: 2, Offset: 1, Lines: []*pb.LogLine{{Time: now(), Text: "after the restart"}},
	403	+	}))
	404	+	require.NoError(t, err)
	405	+	assert.EqualValues(t, 2, next.Msg.GetNextOffset())
	406	+
	407	+	root2 := newClients(srv.URL+"/-/api", adminToken)
	408	+	runs, err := root2.run.ListRuns(c.ctx, connect.NewRequest(&pb.ListRunsRequest{Repo: "proj"}))
	409	+	require.NoError(t, err)
	410	+	require.Len(t, runs.Msg.GetRuns(), 1)
	411	+	assert.Equal(t, pb.CiStatus_CI_STATUS_IN_PROGRESS, runs.Msg.GetRuns()[0].GetStatus())
	412	+
	413	+	// Finish the job on the new server: its dependant is planned and ready.
	414	+	_, err = api2.runner.CompleteJob(c.ctx, connect.NewRequest(&pb.CompleteJobRequest{
	415	+	JobId: job.GetJobId(), Conclusion: pb.CiConclusion_CI_CONCLUSION_SUCCESS,
	416	+	}))
	417	+	require.NoError(t, err)
	418	+	got := make(chan *pb.Job, 1)
	419	+	go func() {
	420	+	for range 50 {
	421	+	res, err := newClients(srv.URL+"/-/api", c.runnerSecret(runner)).runner.FetchJob(c.ctx, connect.NewRequest(&pb.FetchJobRequest{}))
	422	+	if err == nil && res.Msg.GetJob() != nil {
	423	+	got <- res.Msg.GetJob()
	424	+	return
	425	+	}
	426	+	}
	427	+	got <- nil
	428	+	}()
	429	+	select {
	430	+	case j := <-got:
	431	+	require.NotNil(t, j, "the runner token and the queue survive a restart")
	432	+	assert.Equal(t, "test", j.GetJobKey())
	433	+	case <-time.After(30 * time.Second):
	434	+	t.Fatal("no job after the restart")
	435	+	}
	436	+	}
	437	+
	438	+	// AC5: a head without any check merges, and so does one whose checks passed; a
	439	+	// merge runs the push workflows on the target.
	440	+	func TestCIMergeWithoutChecksAndWithPassingChecks(t *testing.T) {
	441	+	c := newCIHarness(t, nil)
	442	+	runner := c.registerRunner("alpha", "ubuntu-latest")
	443	+	c.push("main", map[string]string{"README": "x\n"}, "")
	444	+	headOf := func(branch string) string { return strings.TrimSpace(c.bareGit("proj", "rev-parse", branch)) }
	445	+	merge := func(n int64, head string) error {
	446	+	_, err := c.root.mr.MergeMergeRequest(c.ctx, connect.NewRequest(&pb.MergeMergeRequestRequest{Repo: "proj", Number: n, HeadId: head}))
	447	+	return err
	448	+	}
	449	+
	450	+	// No workflow on the branch: no check, the merge goes through.
	451	+	c.push("plain", map[string]string{"a.txt": "a\n"}, "")
	452	+	mr1, err := c.root.mr.CreateMergeRequest(c.ctx, connect.NewRequest(&pb.CreateMergeRequestRequest{Repo: "proj", Title: "plain", SourceBranch: "plain"}))
	453	+	require.NoError(t, err)
	454	+	require.NoError(t, merge(mr1.Msg.GetMergeRequest().GetNumber(), headOf("plain")))
	455	+
	456	+	// A branch with a workflow: merge after its check passed. The merge itself
	457	+	// triggers the workflow on main (GitHub runs push workflows on a merge).
	458	+	c.mustGit(c.wt, "checkout", "-q", "main")
	459	+	c.mustGit(c.wt, "pull", "-q", c.urlAuth(adminToken, "/proj.git"), "main")
	460	+	c.mustGit(c.wt, "checkout", "-q", "-b", "ci")
	461	+	c.writeFile(c.wt, ".github/workflows/ci.yml", []byte(strings.Replace(workflowWith("branches: [main, ci]"), "echo hi", "echo ci", 1)))
	462	+	c.mustGit(c.wt, "push", "-q", c.urlAuth(adminToken, "/proj.git"), "ci")
	463	+	head := headOf("ci")
	464	+	c.waitFor("the run on ci", func() bool { return len(c.checks(head)) == 1 })
	465	+	mr2, err := c.root.mr.CreateMergeRequest(c.ctx, connect.NewRequest(&pb.CreateMergeRequestRequest{Repo: "proj", Title: "ci", SourceBranch: "ci"}))
	466	+	require.NoError(t, err)
	467	+	job := runner.await()
	468	+	runner.runJob(job, pb.CiConclusion_CI_CONCLUSION_SUCCESS, nil, "ok")
	469	+	c.waitFor("the check passes", func() bool {
	470	+	cs := c.checks(head)
	471	+	return len(cs) == 1 && cs[0].GetConclusion() == pb.CiConclusion_CI_CONCLUSION_SUCCESS
	472	+	})
	473	+	require.NoError(t, merge(mr2.Msg.GetMergeRequest().GetNumber(), head))
	474	+
	475	+	// The fast-forward merge moved main to the same commit: the workflow runs
	476	+	// again for main, with the owner as the actor.
	477	+	c.waitFor("the merge triggers a run on main", func() bool {
	478	+	for _, r := range c.runs() {
	479	+	if r.GetRef() == "refs/heads/main" && r.GetSha() == head {
	480	+	return true
	481	+	}
	482	+	}
	483	+	return false
	484	+	})
	485	+	for _, r := range c.runs() {
	486	+	if r.GetRef() == "refs/heads/main" {
	487	+	assert.Equal(t, "owner", r.GetActor())
	488	+	}
	489	+	}
	490	+	}
	491	+
	492	+	// Repository deletion removes the runs, the checks and the logs; a repository
	493	+	// created again starts at run number 1.
	494	+	func TestCIRepositoryDeletionRemovesRunsAndLogs(t *testing.T) {
	495	+	c := newCIHarness(t, nil)
	496	+	runner := c.registerRunner("alpha", "ubuntu-latest")
	497	+	sha := c.push("main", map[string]string{".github/workflows/ci.yml": strings.Split(ciWorkflow, " test:")[0]}, "")
	498	+	job := runner.await()
	499	+	runner.runJob(job, pb.CiConclusion_CI_CONCLUSION_SUCCESS, nil, "line")
	500	+	logDir := filepath.Join(c.data, "ci", "logs")
	501	+	entries, err := os.ReadDir(logDir)
	502	+	require.NoError(t, err)
	503	+	require.NotEmpty(t, entries, "the log is a file under ci/logs")
	504	+
	505	+	_, err = c.root.repo.DeleteRepo(c.ctx, connect.NewRequest(&pb.DeleteRepoRequest{Repo: "proj"}))
	506	+	require.NoError(t, err)
	507	+	_, err = c.root.run.ListRuns(c.ctx, connect.NewRequest(&pb.ListRunsRequest{Repo: "proj"}))
	508	+	assert.Equal(t, connect.CodeNotFound, ciErrCode(err))
	509	+	entries, err = os.ReadDir(logDir)
	510	+	require.NoError(t, err)
	511	+	assert.Empty(t, entries, "the logs are removed with the repository")
	512	+
	513	+	_, err = c.forge.Repos.Create(c.ctx, domain.Root(), ports.CreateRepoParams{Name: c.name("proj"), Visibility: domain.VisibilityPrivate})
	514	+	require.NoError(t, err)
	515	+	assert.Empty(t, c.checks(sha), "no check survives the repository")
	516	+	c.push("main", map[string]string{".github/workflows/ci.yml": strings.Split(ciWorkflow, " test:")[0] + "# again\n"}, "again")
	517	+	c.waitFor("a run", func() bool { return len(c.runs()) == 1 })
	518	+	assert.EqualValues(t, 1, c.runs()[0].GetNumber(), "run numbers start again with the repository")
	519	+	assert.Greater(t, c.runs()[0].GetId(), int64(1), "run ids are never reused")
	520	+	}
	521	+
	522	+	// The events of a run and its jobs are published on the repository topic.
	523	+	func TestCIEventsAreStreamedOnTheRepositoryTopic(t *testing.T) {
	524	+	c := newCIHarness(t, nil)
	525	+	runner := c.registerRunner("alpha", "ubuntu-latest")
	526	+	s := c.watch(c.root, "", "repo:proj")
	527	+	c.push("main", map[string]string{".github/workflows/ci.yml": strings.Split(ciWorkflow, " test:")[0]}, "")
	528	+	job := runner.await()
	529	+	runner.runJob(job, pb.CiConclusion_CI_CONCLUSION_SUCCESS, nil, "x")
	530	+
	531	+	var runUpdates, jobUpdates int
	532	+	var lastRun *pb.Run
	533	+	deadline := time.After(10 * time.Second)
	534	+	for lastRun.GetStatus() != pb.CiStatus_CI_STATUS_COMPLETED {
	535	+	select {
	536	+	case ev := <-s.events:
	537	+	if r := ev.GetRunUpdated(); r != nil {
	538	+	runUpdates++
	539	+	if r.GetRun().GetVersion() >= lastRun.GetVersion() {
	540	+	lastRun = r.GetRun()
	541	+	}
	542	+	}
	543	+	if ev.GetJobUpdated() != nil {
	544	+	jobUpdates++
	545	+	}
	546	+	case <-deadline:
	547	+	t.Fatalf("no completed run event: %d run and %d job events", runUpdates, jobUpdates)
	548	+	}
	549	+	}
	550	+	assert.GreaterOrEqual(t, runUpdates, 2)
	551	+	assert.GreaterOrEqual(t, jobUpdates, 2)
	552	+
	553	+	// They are never sent on "home".
	554	+	home := c.watch(c.root, "", "home")
	555	+	_, err := c.root.run.RerunRun(c.ctx, connect.NewRequest(&pb.RerunRunRequest{Repo: "proj", RunId: lastRun.GetId()}))
	556	+	require.NoError(t, err)
	557	+	c.sentinel(t, home, "sentinel")
	558	+	}

M internal/ports/ci.go +9 -1


@@ -280,7 +280,10 @@ type RunnerService interface {
280	280		UpdateJob(ctx context.Context, secret, jobID string, steps []StepUpdate) error
281	281		AppendLog(ctx context.Context, secret, jobID string, step int32, offset int64, lines []domain.CILogLine) (next int64, err error)
282	282		Heartbeat(ctx context.Context, secret, jobID string) error
283		-	WatchJob(ctx context.Context, secret, jobID string, send func(JobSignal) error) error
	283	+	// WatchJob calls ready once the job token is valid and the stream is
	284	+	// registered, before any message: the adapter flushes the response headers
	285	+	// then, because a client's call returns only when they arrive.
	286	+	WatchJob(ctx context.Context, secret, jobID string, ready func(), send func(JobSignal) error) error
284	287		CompleteJob(ctx context.Context, secret string, params CompleteJobParams) (CompleteJobResult, error)
285	288		}
286	289

@@ -340,6 +343,11 @@ type GetJobLogParams struct {
340	343		Attempt int32
341	344		Offset int64
342	345		Follow bool
	346	+	// Ready, when not nil, is called once the access is checked and the
	347	+	// stream is counted, before any batch: the adapter flushes the response
	348	+	// headers then (a follow on a job with no line yet would keep the client's
	349	+	// call waiting for the first one).
	350	+	Ready func()
343	351		// Client identifies where the call comes from, for the limit of streams
344	352		// per client; empty counts against the total only.
345	353		Client string

 	UpdateJob(ctx context.Context, secret, jobID string, steps []StepUpdate) error
 	AppendLog(ctx context.Context, secret, jobID string, step int32, offset int64, lines []domain.CILogLine) (next int64, err error)
 	Heartbeat(ctx context.Context, secret, jobID string) error
-	WatchJob(ctx context.Context, secret, jobID string, send func(JobSignal) error) error
+	// WatchJob calls ready once the job token is valid and the stream is
+	// registered, before any message: the adapter flushes the response headers
+	// then, because a client's call returns only when they arrive.
+	WatchJob(ctx context.Context, secret, jobID string, ready func(), send func(JobSignal) error) error
 	CompleteJob(ctx context.Context, secret string, params CompleteJobParams) (CompleteJobResult, error)
 }
 	Attempt int32
 	Offset  int64
 	Follow  bool
+	// Ready, when not nil, is called once the access is checked and the
+	// stream is counted, before any batch: the adapter flushes the response
+	// headers then (a follow on a job with no line yet would keep the client's
+	// call waiting for the first one).
+	Ready func()
 	// Client identifies where the call comes from, for the limit of streams
 	// per client; empty counts against the total only.
 	Client string

M internal/ports/mocks/RunnerService.go +16 -10


@@ -487,16 +487,16 @@ func (_c *RunnerService_UpdateJob_Call) RunAndReturn(run func(ctx context.Contex
487	487		}
488	488
489	489		// WatchJob provides a mock function for the type RunnerService
490		-	func (_mock *RunnerService) WatchJob(ctx context.Context, secret string, jobID string, send func(ports.JobSignal) error) error {
491		-	ret := _mock.Called(ctx, secret, jobID, send)
	490	+	func (_mock *RunnerService) WatchJob(ctx context.Context, secret string, jobID string, ready func(), send func(ports.JobSignal) error) error {
	491	+	ret := _mock.Called(ctx, secret, jobID, ready, send)
492	492
493	493		if len(ret) == 0 {
494	494		panic("no return value specified for WatchJob")
495	495		}
496	496
497	497		var r0 error
498		-	if returnFunc, ok := ret.Get(0).(func(context.Context, string, string, func(ports.JobSignal) error) error); ok {
499		-	r0 = returnFunc(ctx, secret, jobID, send)
	498	+	if returnFunc, ok := ret.Get(0).(func(context.Context, string, string, func(), func(ports.JobSignal) error) error); ok {
	499	+	r0 = returnFunc(ctx, secret, jobID, ready, send)
500	500		} else {
501	501		r0 = ret.Error(0)
502	502		}

@@ -512,12 +512,13 @@ type RunnerService_WatchJob_Call struct {
512	512		// - ctx context.Context
513	513		// - secret string
514	514		// - jobID string
	515	+	// - ready func()
515	516		// - send func(ports.JobSignal) error
516		-	func (_e RunnerService_Expecter) WatchJob(ctx any, secret any, jobID any, send any) RunnerService_WatchJob_Call {
517		-	return &RunnerService_WatchJob_Call{Call: _e.mock.On("WatchJob", ctx, secret, jobID, send)}
	517	+	func (_e RunnerService_Expecter) WatchJob(ctx any, secret any, jobID any, ready any, send any) RunnerService_WatchJob_Call {
	518	+	return &RunnerService_WatchJob_Call{Call: _e.mock.On("WatchJob", ctx, secret, jobID, ready, send)}
518	519		}
519	520
520		-	func (_c RunnerService_WatchJob_Call) Run(run func(ctx context.Context, secret string, jobID string, send func(ports.JobSignal) error)) RunnerService_WatchJob_Call {
	521	+	func (_c RunnerService_WatchJob_Call) Run(run func(ctx context.Context, secret string, jobID string, ready func(), send func(ports.JobSignal) error)) RunnerService_WatchJob_Call {
521	522		_c.Call.Run(func(args mock.Arguments) {
522	523		var arg0 context.Context
523	524		if args[0] != nil {

@@ -531,15 +532,20 @@ func (_c *RunnerService_WatchJob_Call) Run(run func(ctx context.Context, secret
531	532		if args[2] != nil {
532	533		arg2 = args[2].(string)
533	534		}
534		-	var arg3 func(ports.JobSignal) error
	535	+	var arg3 func()
535	536		if args[3] != nil {
536		-	arg3 = args[3].(func(ports.JobSignal) error)
	537	+	arg3 = args[3].(func())
	538	+	}
	539	+	var arg4 func(ports.JobSignal) error
	540	+	if args[4] != nil {
	541	+	arg4 = args[4].(func(ports.JobSignal) error)
537	542		}
538	543		run(
539	544		arg0,
540	545		arg1,
541	546		arg2,
542	547		arg3,
	548	+	arg4,
543	549		)
544	550		})
545	551		return _c

@@ -550,7 +556,7 @@ func (_c RunnerService_WatchJob_Call) Return(err error) RunnerService_WatchJob
550	556		return _c
551	557		}
552	558
553		-	func (_c RunnerService_WatchJob_Call) RunAndReturn(run func(ctx context.Context, secret string, jobID string, send func(ports.JobSignal) error) error) RunnerService_WatchJob_Call {
	559	+	func (_c RunnerService_WatchJob_Call) RunAndReturn(run func(ctx context.Context, secret string, jobID string, ready func(), send func(ports.JobSignal) error) error) RunnerService_WatchJob_Call {
554	560		_c.Call.Return(run)
555	561		return _c
556	562		}

 }
 // WatchJob provides a mock function for the type RunnerService
-func (_mock *RunnerService) WatchJob(ctx context.Context, secret string, jobID string, send func(ports.JobSignal) error) error {
+func (_mock *RunnerService) WatchJob(ctx context.Context, secret string, jobID string, ready func(), send func(ports.JobSignal) error) error {
-	ret := _mock.Called(ctx, secret, jobID, send)
+	ret := _mock.Called(ctx, secret, jobID, ready, send)
 	if len(ret) == 0 {
 		panic("no return value specified for WatchJob")
 	}
 	var r0 error
-	if returnFunc, ok := ret.Get(0).(func(context.Context, string, string, func(ports.JobSignal) error) error); ok {
+	if returnFunc, ok := ret.Get(0).(func(context.Context, string, string, func(), func(ports.JobSignal) error) error); ok {
-		r0 = returnFunc(ctx, secret, jobID, send)
+		r0 = returnFunc(ctx, secret, jobID, ready, send)
 	} else {
 		r0 = ret.Error(0)
 	}
 //   - ctx context.Context
 //   - secret string
 //   - jobID string
+//   - ready func()
 //   - send func(ports.JobSignal) error
-func (_e *RunnerService_Expecter) WatchJob(ctx any, secret any, jobID any, send any) *RunnerService_WatchJob_Call {
+func (_e *RunnerService_Expecter) WatchJob(ctx any, secret any, jobID any, ready any, send any) *RunnerService_WatchJob_Call {
-	return &RunnerService_WatchJob_Call{Call: _e.mock.On("WatchJob", ctx, secret, jobID, send)}
+	return &RunnerService_WatchJob_Call{Call: _e.mock.On("WatchJob", ctx, secret, jobID, ready, send)}
 }
-func (_c *RunnerService_WatchJob_Call) Run(run func(ctx context.Context, secret string, jobID string, send func(ports.JobSignal) error)) *RunnerService_WatchJob_Call {
+func (_c *RunnerService_WatchJob_Call) Run(run func(ctx context.Context, secret string, jobID string, ready func(), send func(ports.JobSignal) error)) *RunnerService_WatchJob_Call {
 	_c.Call.Run(func(args mock.Arguments) {
 		var arg0 context.Context
 		if args[0] != nil {
 		if args[2] != nil {
 			arg2 = args[2].(string)
 		}
-		var arg3 func(ports.JobSignal) error
+		var arg3 func()
 		if args[3] != nil {
-			arg3 = args[3].(func(ports.JobSignal) error)
+			arg3 = args[3].(func())
+		}
+		var arg4 func(ports.JobSignal) error
+		if args[4] != nil {
+			arg4 = args[4].(func(ports.JobSignal) error)
 		}
 		run(
 			arg0,
 			arg1,
 			arg2,
 			arg3,
+			arg4,
 		)
 	})
 	return _c
 	return _c
 }
-func (_c *RunnerService_WatchJob_Call) RunAndReturn(run func(ctx context.Context, secret string, jobID string, send func(ports.JobSignal) error) error) *RunnerService_WatchJob_Call {
+func (_c *RunnerService_WatchJob_Call) RunAndReturn(run func(ctx context.Context, secret string, jobID string, ready func(), send func(ports.JobSignal) error) error) *RunnerService_WatchJob_Call {
 	_c.Call.Return(run)
 	return _c
 }

M internal/usecase/ci_runner.go +4 -1


@@ -623,7 +623,7 @@ func (c *CI) Heartbeat(ctx context.Context, secret, jobID string) error {
623	623
624	624		// WatchJob streams what the runner must react to: the cancel request (the
625	625		// first message when one is already pending), keepalives, and the end.
626		-	func (c *CI) WatchJob(ctx context.Context, secret, jobID string, send func(ports.JobSignal) error) error {
	626	+	func (c *CI) WatchJob(ctx context.Context, secret, jobID string, ready func(), send func(ports.JobSignal) error) error {
627	627		j, err := c.jobFor(ctx, secret, jobID)
628	628		if err != nil {
629	629		return err

@@ -635,6 +635,9 @@ func (c *CI) WatchJob(ctx context.Context, secret, jobID string, send func(ports
635	635		keepalive := time.NewTicker(c.cfg.Keepalive)
636	636		defer keepalive.Stop()
637	637		cancelSent := false
	638	+	if ready != nil {
	639	+	ready()
	640	+	}
638	641		for {
639	642		wake := c.jobs.wait(ref)
640	643		cur, err := c.cfg.Store.GetJob(ctx, j.RunID, j.RunAttempt, j.ID)

 // WatchJob streams what the runner must react to: the cancel request (the
 // first message when one is already pending), keepalives, and the end.
-func (c *CI) WatchJob(ctx context.Context, secret, jobID string, send func(ports.JobSignal) error) error {
+func (c *CI) WatchJob(ctx context.Context, secret, jobID string, ready func(), send func(ports.JobSignal) error) error {
 	j, err := c.jobFor(ctx, secret, jobID)
 	if err != nil {
 		return err
 	keepalive := time.NewTicker(c.cfg.Keepalive)
 	defer keepalive.Stop()
 	cancelSent := false
+	if ready != nil {
+		ready()
+	}
 	for {
 		wake := c.jobs.wait(ref)
 		cur, err := c.cfg.Store.GetJob(ctx, j.RunID, j.RunAttempt, j.ID)

M internal/usecase/ci_runs.go +3 -0


@@ -227,6 +227,9 @@ func (c *CI) GetJobLog(ctx context.Context, p domain.Principal, name domain.Repo
227	227		return err
228	228		}
229	229		defer release()
	230	+	if params.Ready != nil {
	231	+	params.Ready()
	232	+	}
230	233		ref := logRef(j)
231	234		jref := jobRef(j)
232	235		offset := params.Offset


@@ -227,6 +227,9 @@ func (c *CI) GetJobLog(ctx context.Context, p domain.Principal, name domain.Repo
227	return err	227		return err
228	}	228		}
229	defer release()	229		defer release()
		230	+	if params.Ready != nil {
		231	+	params.Ready()
		232	+	}
230	ref := logRef(j)	233		ref := logRef(j)
231	jref := jobRef(j)	234		jref := jobRef(j)
232	offset := params.Offset	235		offset := params.Offset

A internal/usecase/ci_unit_test.go +167 -0

@@ -0,0 +1,167 @@
	1	+	// SPDX-License-Identifier: AGPL-3.0-or-later
	2	+
	3	+	package usecase_test
	4	+
	5	+	import (
	6	+	"context"
	7	+	"testing"
	8	+
	9	+	"github.com/stretchr/testify/assert"
	10	+	"github.com/stretchr/testify/mock"
	11	+	"github.com/stretchr/testify/require"
	12	+
	13	+	"bnema.dev/ermite/internal/domain"
	14	+	"bnema.dev/ermite/internal/ports"
	15	+	"bnema.dev/ermite/internal/ports/mocks"
	16	+	"bnema.dev/ermite/internal/usecase"
	17	+	)
	18	+
	19	+	// mergeUnit is a merge request that is ready to merge, with a checks reader.
	20	+	type mergeUnit struct {
	21	+	*mrUnit
	22	+	checks *mocks.CIChecksReader
	23	+	trig *mocks.WorkflowTrigger
	24	+	head domain.ObjectID
	25	+	}
	26	+
	27	+	func newMergeUnit(t testing.T, block bool) mergeUnit {
	28	+	t.Helper()
	29	+	u := &mrUnit{
	30	+	meta: mocks.NewRepoMetaStore(t),
	31	+	store: mocks.NewMergeRequestStore(t),
	32	+	repos: mocks.NewGitRepos(t),
	33	+	git: mocks.NewMergeGit(t),
	34	+	repo: domain.Repo{Name: name(t, "proj"), Visibility: domain.VisibilityPrivate},
	35	+	}
	36	+	u.repo.Settings = domain.DefaultRepoSettings()
	37	+	u.repo.Settings.FailingChecksBlock = block
	38	+	clock := mocks.NewClock(t)
	39	+	clock.EXPECT().Now().Return(now).Maybe()
	40	+	sched := mocks.NewMaintenanceScheduler(t)
	41	+	sched.EXPECT().Schedule(mock.Anything, mock.Anything).Return().Maybe()
	42	+	m := &mergeUnit{mrUnit: u, checks: mocks.NewCIChecksReader(t), trig: mocks.NewWorkflowTrigger(t), head: someID(t)}
	43	+	u.mrs = usecase.NewMergeRequests(usecase.MergeRequestsConfig{
	44	+	Meta: u.meta, Store: u.store, Repos: u.repos, Git: u.git,
	45	+	Signer: mocks.NewCommitSigner(t), Bus: mocks.NewEventBus(t), Clock: clock, Maint: sched,
	46	+	CI: m.trig, Checks: m.checks,
	47	+	})
	48	+	u.meta.EXPECT().Get(mock.Anything, u.repo.Name).Return(u.repo, nil)
	49	+	u.store.EXPECT().Get(mock.Anything, u.repo.Name, int64(1)).Return(domain.MergeRequest{
	50	+	Repo: u.repo.Name, Number: 1, State: domain.MRStateOpen, Source: "feat", Target: "main",
	51	+	}, nil)
	52	+	tip, err := domain.ParseObjectID("1111111111111111111111111111111111111111")
	53	+	require.NoError(t, err)
	54	+	u.git.EXPECT().BranchHead(mock.Anything, u.repo.Name, "feat").Return(m.head, nil)
	55	+	u.git.EXPECT().BranchHead(mock.Anything, u.repo.Name, "main").Return(tip, nil)
	56	+	u.git.EXPECT().IsAncestor(mock.Anything, u.repo.Name, tip, m.head).Return(true, nil)
	57	+	return m
	58	+	}
	59	+
	60	+	func (m *mergeUnit) merge() error {
	61	+	_, err := m.mrs.Merge(context.Background(), domain.Root(), m.repo.Name, 1, ports.MergeParams{HeadID: m.head})
	62	+	return err
	63	+	}
	64	+
	65	+	// AC5: the merge rule under FailingChecksBlock.
	66	+	func TestMergeRuleOfFailingChecksBlock(t *testing.T) {
	67	+	t.Parallel()
	68	+	check := func(status domain.CIStatus, c domain.CIConclusion, name string) domain.CICheck {
	69	+	return domain.CICheck{Name: name, Status: status, Conclusion: c}
	70	+	}
	71	+	tests := []struct {
	72	+	name string
	73	+	checks []domain.CICheck
	74	+	blocked string // part of the message
	75	+	}{
	76	+	{"a failure blocks", []domain.CICheck{check(domain.CICompleted, domain.CISuccess, "a"), check(domain.CICompleted, domain.CIFailure, "build")}, "failing checks block the merge: build"},
	77	+	{"a timeout blocks", []domain.CICheck{check(domain.CICompleted, domain.CITimedOut, "slow")}, "slow"},
	78	+	{"a cancelled check blocks", []domain.CICheck{check(domain.CICompleted, domain.CICancelled, "c")}, "failing checks"},
	79	+	{"action required blocks", []domain.CICheck{check(domain.CICompleted, domain.CIActionRequired, "env")}, "failing checks"},
	80	+	{"a queued check blocks", []domain.CICheck{check(domain.CIQueued, "", "q")}, "have not finished: q"},
	81	+	{"a running check blocks", []domain.CICheck{check(domain.CIInProgress, "", "r")}, "have not finished: r"},
	82	+	}
	83	+	for _, tt := range tests {
	84	+	t.Run(tt.name, func(t *testing.T) {
	85	+	t.Parallel()
	86	+	m := newMergeUnit(t, true)
	87	+	m.checks.EXPECT().CommitChecks(mock.Anything, m.repo.Name, m.head.String()).Return(tt.checks, nil)
	88	+	// The merge stops at the rule: the branch mock expects no move.
	89	+	err := m.merge()
	90	+	require.ErrorIs(t, err, domain.ErrPreconditionFailed)
	91	+	assert.Contains(t, err.Error(), tt.blocked)
	92	+	})
	93	+	}
	94	+	// The cases where the merge goes on (no check, all passing, rule off) need
	95	+	// the whole merge: see TestCIMergeIsBlockedByChecks, which merges for real.
	96	+	}
	97	+
	98	+	func TestCombineChecks(t *testing.T) {
	99	+	t.Parallel()
	100	+	ck := func(s domain.CIStatus, c domain.CIConclusion) domain.CICheck {
	101	+	return domain.CICheck{Status: s, Conclusion: c}
	102	+	}
	103	+	assert.Equal(t, domain.CheckStateNone, domain.CombineChecks(nil))
	104	+	assert.Equal(t, domain.CheckStateSuccess, domain.CombineChecks([]domain.CICheck{ck(domain.CICompleted, domain.CISuccess), ck(domain.CICompleted, domain.CISkipped)}))
	105	+	assert.Equal(t, domain.CheckStatePending, domain.CombineChecks([]domain.CICheck{ck(domain.CICompleted, domain.CISuccess), ck(domain.CIQueued, "")}))
	106	+	assert.Equal(t, domain.CheckStateFailure, domain.CombineChecks([]domain.CICheck{ck(domain.CIInProgress, ""), ck(domain.CICompleted, domain.CIFailure)}), "a failure wins over pending")
	107	+	}
	108	+
	109	+	func TestLabelsMatchAndRunnerLabels(t *testing.T) {
	110	+	t.Parallel()
	111	+	have := domain.RunnerLabels([]string{"ubuntu-latest", "gpu"}, "linux", "x64")
	112	+	assert.Equal(t, []string{"gpu", "linux", "self-hosted", "ubuntu-latest", "x64"}, have)
	113	+	assert.True(t, domain.LabelsMatch([]string{"Ubuntu-Latest"}, have), "case-insensitive")
	114	+	assert.True(t, domain.LabelsMatch([]string{"self-hosted", "linux", "x64"}, have))
	115	+	assert.False(t, domain.LabelsMatch([]string{"self-hosted", "arm64"}, have))
	116	+	assert.False(t, domain.LabelsMatch([]string{"windows-latest"}, have))
	117	+	assert.True(t, domain.LabelsMatch(nil, have), "no requirement matches")
	118	+	assert.False(t, domain.LabelsMatch([]string{"runner-name"}, have), "the runner name is not a label")
	119	+	}
	120	+
	121	+	func TestStepRanges(t *testing.T) {
	122	+	t.Parallel()
	123	+	steps := []domain.CIStep{
	124	+	{Number: 1, HasLines: true, LogStart: 0},
	125	+	{Number: 2},
	126	+	{Number: 3, HasLines: true, LogStart: 4},
	127	+	{Number: 4, HasLines: true, LogStart: 9},
	128	+	{Number: 5},
	129	+	}
	130	+	got := domain.StepRanges(steps, 12)
	131	+	assert.Equal(t, []domain.StepRange{{0, 4}, {4, 4}, {4, 9}, {9, 12}, {12, 12}}, got)
	132	+	assert.EqualValues(t, 3, domain.StepOfLine(steps, 5))
	133	+	assert.EqualValues(t, 1, domain.StepOfLine(steps, 0))
	134	+	assert.EqualValues(t, 4, domain.StepOfLine(steps, 11))
	135	+	}
	136	+
	137	+	func TestJobIDsRoundTripAndRejectOthers(t *testing.T) {
	138	+	t.Parallel()
	139	+	id := domain.NewJobID(42, 3, 1)
	140	+	run, index, combo, ok := domain.ParseJobID(id)
	141	+	require.True(t, ok)
	142	+	assert.Equal(t, []int{42, 3, 1}, []int{int(run), index, combo})
	143	+	for _, bad := range []string{"", "1-2", "1-2-3-4", "a-b-c", "0-1-1", "1--2", "01-2-3", "../../x", "1-2-3/4"} {
	144	+	_, _, _, ok := domain.ParseJobID(bad)
	145	+	assert.False(t, ok, bad)
	146	+	}
	147	+	}
	148	+
	149	+	func TestOutputLimitsAndNames(t *testing.T) {
	150	+	t.Parallel()
	151	+	assert.NoError(t, domain.CheckOutputs(map[string]string{"a": "1", "_b-c": "2"}))
	152	+	assert.Error(t, domain.CheckOutputs(map[string]string{"1a": "x"}), "starts with a digit")
	153	+	assert.Error(t, domain.CheckOutputs(map[string]string{"": "x"}))
	154	+	assert.Error(t, domain.CheckOutputs(map[string]string{"A": "x", "a": "y"}), "names differ only by case")
	155	+	assert.Error(t, domain.CheckOutputs(map[string]string{"big": string(make([]byte, domain.CIMaxOutputBytes+1))}))
	156	+	assert.Error(t, domain.CheckOutputs(map[string]string{"a": string(make([]byte, domain.CIMaxOutputBytes/2+1)), "b": string(make([]byte, domain.CIMaxOutputBytes/2+1))}), "1 MiB together")
	157	+	}
	158	+
	159	+	func TestTokenShapes(t *testing.T) {
	160	+	t.Parallel()
	161	+	good := "ermitejob_" + "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMN-_0"
	162	+	assert.True(t, domain.IsJobToken(good))
	163	+	assert.False(t, domain.IsRunnerToken(good))
	164	+	assert.False(t, domain.IsJobToken(good[:len(good)-1]))
	165	+	assert.False(t, domain.IsJobToken("ermite_"+good[10:]))
	166	+	assert.False(t, domain.IsJobToken(good+"="))
	167	+	}

@@ -0,0 +1,167 @@
	1	+	// SPDX-License-Identifier: AGPL-3.0-or-later
	2	+
	3	+	package usecase_test
	4	+
	5	+	import (
	6	+	"context"
	7	+	"testing"
	8	+
	9	+	"github.com/stretchr/testify/assert"
	10	+	"github.com/stretchr/testify/mock"
	11	+	"github.com/stretchr/testify/require"
	12	+
	13	+	"bnema.dev/ermite/internal/domain"
	14	+	"bnema.dev/ermite/internal/ports"
	15	+	"bnema.dev/ermite/internal/ports/mocks"
	16	+	"bnema.dev/ermite/internal/usecase"
	17	+	)
	18	+
	19	+	// mergeUnit is a merge request that is ready to merge, with a checks reader.
	20	+	type mergeUnit struct {
	21	+	*mrUnit
	22	+	checks *mocks.CIChecksReader
	23	+	trig *mocks.WorkflowTrigger
	24	+	head domain.ObjectID
	25	+	}
	26	+
	27	+	func newMergeUnit(t testing.T, block bool) mergeUnit {
	28	+	t.Helper()
	29	+	u := &mrUnit{
	30	+	meta: mocks.NewRepoMetaStore(t),
	31	+	store: mocks.NewMergeRequestStore(t),
	32	+	repos: mocks.NewGitRepos(t),
	33	+	git: mocks.NewMergeGit(t),
	34	+	repo: domain.Repo{Name: name(t, "proj"), Visibility: domain.VisibilityPrivate},
	35	+	}
	36	+	u.repo.Settings = domain.DefaultRepoSettings()
	37	+	u.repo.Settings.FailingChecksBlock = block
	38	+	clock := mocks.NewClock(t)
	39	+	clock.EXPECT().Now().Return(now).Maybe()
	40	+	sched := mocks.NewMaintenanceScheduler(t)
	41	+	sched.EXPECT().Schedule(mock.Anything, mock.Anything).Return().Maybe()
	42	+	m := &mergeUnit{mrUnit: u, checks: mocks.NewCIChecksReader(t), trig: mocks.NewWorkflowTrigger(t), head: someID(t)}
	43	+	u.mrs = usecase.NewMergeRequests(usecase.MergeRequestsConfig{
	44	+	Meta: u.meta, Store: u.store, Repos: u.repos, Git: u.git,
	45	+	Signer: mocks.NewCommitSigner(t), Bus: mocks.NewEventBus(t), Clock: clock, Maint: sched,
	46	+	CI: m.trig, Checks: m.checks,
	47	+	})
	48	+	u.meta.EXPECT().Get(mock.Anything, u.repo.Name).Return(u.repo, nil)
	49	+	u.store.EXPECT().Get(mock.Anything, u.repo.Name, int64(1)).Return(domain.MergeRequest{
	50	+	Repo: u.repo.Name, Number: 1, State: domain.MRStateOpen, Source: "feat", Target: "main",
	51	+	}, nil)
	52	+	tip, err := domain.ParseObjectID("1111111111111111111111111111111111111111")
	53	+	require.NoError(t, err)
	54	+	u.git.EXPECT().BranchHead(mock.Anything, u.repo.Name, "feat").Return(m.head, nil)
	55	+	u.git.EXPECT().BranchHead(mock.Anything, u.repo.Name, "main").Return(tip, nil)
	56	+	u.git.EXPECT().IsAncestor(mock.Anything, u.repo.Name, tip, m.head).Return(true, nil)
	57	+	return m
	58	+	}
	59	+
	60	+	func (m *mergeUnit) merge() error {
	61	+	_, err := m.mrs.Merge(context.Background(), domain.Root(), m.repo.Name, 1, ports.MergeParams{HeadID: m.head})
	62	+	return err
	63	+	}
	64	+
	65	+	// AC5: the merge rule under FailingChecksBlock.
	66	+	func TestMergeRuleOfFailingChecksBlock(t *testing.T) {
	67	+	t.Parallel()
	68	+	check := func(status domain.CIStatus, c domain.CIConclusion, name string) domain.CICheck {
	69	+	return domain.CICheck{Name: name, Status: status, Conclusion: c}
	70	+	}
	71	+	tests := []struct {
	72	+	name string
	73	+	checks []domain.CICheck
	74	+	blocked string // part of the message
	75	+	}{
	76	+	{"a failure blocks", []domain.CICheck{check(domain.CICompleted, domain.CISuccess, "a"), check(domain.CICompleted, domain.CIFailure, "build")}, "failing checks block the merge: build"},
	77	+	{"a timeout blocks", []domain.CICheck{check(domain.CICompleted, domain.CITimedOut, "slow")}, "slow"},
	78	+	{"a cancelled check blocks", []domain.CICheck{check(domain.CICompleted, domain.CICancelled, "c")}, "failing checks"},
	79	+	{"action required blocks", []domain.CICheck{check(domain.CICompleted, domain.CIActionRequired, "env")}, "failing checks"},
	80	+	{"a queued check blocks", []domain.CICheck{check(domain.CIQueued, "", "q")}, "have not finished: q"},
	81	+	{"a running check blocks", []domain.CICheck{check(domain.CIInProgress, "", "r")}, "have not finished: r"},
	82	+	}
	83	+	for _, tt := range tests {
	84	+	t.Run(tt.name, func(t *testing.T) {
	85	+	t.Parallel()
	86	+	m := newMergeUnit(t, true)
	87	+	m.checks.EXPECT().CommitChecks(mock.Anything, m.repo.Name, m.head.String()).Return(tt.checks, nil)
	88	+	// The merge stops at the rule: the branch mock expects no move.
	89	+	err := m.merge()
	90	+	require.ErrorIs(t, err, domain.ErrPreconditionFailed)
	91	+	assert.Contains(t, err.Error(), tt.blocked)
	92	+	})
	93	+	}
	94	+	// The cases where the merge goes on (no check, all passing, rule off) need
	95	+	// the whole merge: see TestCIMergeIsBlockedByChecks, which merges for real.
	96	+	}
	97	+
	98	+	func TestCombineChecks(t *testing.T) {
	99	+	t.Parallel()
	100	+	ck := func(s domain.CIStatus, c domain.CIConclusion) domain.CICheck {
	101	+	return domain.CICheck{Status: s, Conclusion: c}
	102	+	}
	103	+	assert.Equal(t, domain.CheckStateNone, domain.CombineChecks(nil))
	104	+	assert.Equal(t, domain.CheckStateSuccess, domain.CombineChecks([]domain.CICheck{ck(domain.CICompleted, domain.CISuccess), ck(domain.CICompleted, domain.CISkipped)}))
	105	+	assert.Equal(t, domain.CheckStatePending, domain.CombineChecks([]domain.CICheck{ck(domain.CICompleted, domain.CISuccess), ck(domain.CIQueued, "")}))
	106	+	assert.Equal(t, domain.CheckStateFailure, domain.CombineChecks([]domain.CICheck{ck(domain.CIInProgress, ""), ck(domain.CICompleted, domain.CIFailure)}), "a failure wins over pending")
	107	+	}
	108	+
	109	+	func TestLabelsMatchAndRunnerLabels(t *testing.T) {
	110	+	t.Parallel()
	111	+	have := domain.RunnerLabels([]string{"ubuntu-latest", "gpu"}, "linux", "x64")
	112	+	assert.Equal(t, []string{"gpu", "linux", "self-hosted", "ubuntu-latest", "x64"}, have)
	113	+	assert.True(t, domain.LabelsMatch([]string{"Ubuntu-Latest"}, have), "case-insensitive")
	114	+	assert.True(t, domain.LabelsMatch([]string{"self-hosted", "linux", "x64"}, have))
	115	+	assert.False(t, domain.LabelsMatch([]string{"self-hosted", "arm64"}, have))
	116	+	assert.False(t, domain.LabelsMatch([]string{"windows-latest"}, have))
	117	+	assert.True(t, domain.LabelsMatch(nil, have), "no requirement matches")
	118	+	assert.False(t, domain.LabelsMatch([]string{"runner-name"}, have), "the runner name is not a label")
	119	+	}
	120	+
	121	+	func TestStepRanges(t *testing.T) {
	122	+	t.Parallel()
	123	+	steps := []domain.CIStep{
	124	+	{Number: 1, HasLines: true, LogStart: 0},
	125	+	{Number: 2},
	126	+	{Number: 3, HasLines: true, LogStart: 4},
	127	+	{Number: 4, HasLines: true, LogStart: 9},
	128	+	{Number: 5},
	129	+	}
	130	+	got := domain.StepRanges(steps, 12)
	131	+	assert.Equal(t, []domain.StepRange{{0, 4}, {4, 4}, {4, 9}, {9, 12}, {12, 12}}, got)
	132	+	assert.EqualValues(t, 3, domain.StepOfLine(steps, 5))
	133	+	assert.EqualValues(t, 1, domain.StepOfLine(steps, 0))
	134	+	assert.EqualValues(t, 4, domain.StepOfLine(steps, 11))
	135	+	}
	136	+
	137	+	func TestJobIDsRoundTripAndRejectOthers(t *testing.T) {
	138	+	t.Parallel()
	139	+	id := domain.NewJobID(42, 3, 1)
	140	+	run, index, combo, ok := domain.ParseJobID(id)
	141	+	require.True(t, ok)
	142	+	assert.Equal(t, []int{42, 3, 1}, []int{int(run), index, combo})
	143	+	for _, bad := range []string{"", "1-2", "1-2-3-4", "a-b-c", "0-1-1", "1--2", "01-2-3", "../../x", "1-2-3/4"} {
	144	+	_, _, _, ok := domain.ParseJobID(bad)
	145	+	assert.False(t, ok, bad)
	146	+	}
	147	+	}
	148	+
	149	+	func TestOutputLimitsAndNames(t *testing.T) {
	150	+	t.Parallel()
	151	+	assert.NoError(t, domain.CheckOutputs(map[string]string{"a": "1", "_b-c": "2"}))
	152	+	assert.Error(t, domain.CheckOutputs(map[string]string{"1a": "x"}), "starts with a digit")
	153	+	assert.Error(t, domain.CheckOutputs(map[string]string{"": "x"}))
	154	+	assert.Error(t, domain.CheckOutputs(map[string]string{"A": "x", "a": "y"}), "names differ only by case")
	155	+	assert.Error(t, domain.CheckOutputs(map[string]string{"big": string(make([]byte, domain.CIMaxOutputBytes+1))}))
	156	+	assert.Error(t, domain.CheckOutputs(map[string]string{"a": string(make([]byte, domain.CIMaxOutputBytes/2+1)), "b": string(make([]byte, domain.CIMaxOutputBytes/2+1))}), "1 MiB together")
	157	+	}
	158	+
	159	+	func TestTokenShapes(t *testing.T) {
	160	+	t.Parallel()
	161	+	good := "ermitejob_" + "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMN-_0"
	162	+	assert.True(t, domain.IsJobToken(good))
	163	+	assert.False(t, domain.IsRunnerToken(good))
	164	+	assert.False(t, domain.IsJobToken(good[:len(good)-1]))
	165	+	assert.False(t, domain.IsJobToken("ermite_"+good[10:]))
	166	+	assert.False(t, domain.IsJobToken(good+"="))
	167	+	}