feat(host): forward annotations, summaries and step outcomes to ermite

The executor sends StepAnnotations and StepSummary frames; the host checks the
limits of the contract again, sends the annotations with the UpdateJob that
completes the step (replace semantics, one call at a time), the rest with
CompleteJob, and summaries with UploadSummary. Both are gated on the server
features ci.annotations and ci.summaries (GetVersion, once per job).

commit: 9116f1eeda7b3880c19bc5170e907d75d6a2eeed
author: brice · 9d ago · 2026-10-13 15:10 UTC
parent: 77f2262
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

12 files · +822 -25

M host_cmd.go +1 -0


@@ -159,6 +159,7 @@ func runCmd(ctx context.Context) int {
159	159		Job: func(tok string) ermiteconnect.RunnerServiceClient {
160	160		return hcfg.Client(url, tok)
161	161		},
	162	+	Features: hcfg.ServerFeaturesFunc(url),
162	163		}
163	164		r.Cfg.URL = url
164	165		if err := r.Run(ctx); err != nil {


@@ -159,6 +159,7 @@ func runCmd(ctx context.Context) int {
159	Job: func(tok string) ermiteconnect.RunnerServiceClient {	159		Job: func(tok string) ermiteconnect.RunnerServiceClient {
160	return hcfg.Client(url, tok)	160		return hcfg.Client(url, tok)
161	},	161		},
		162	+	Features: hcfg.ServerFeaturesFunc(url),
162	}	163		}
163	r.Cfg.URL = url	164		r.Cfg.URL = url
164	if err := r.Run(ctx); err != nil {	165		if err := r.Run(ctx); err != nil {

M internal/execstream/frame.go +12 -0


@@ -115,6 +115,10 @@ type JobHandler struct {
115	115		StepCompleted func(*execpb.StepCompleted)
116	116		Log func(*execpb.Log)
117	117		JobResult func(*execpb.JobResult)
	118	+	// StepAnnotations and StepSummary come before the StepCompleted of their
	119	+	// step.
	120	+	StepAnnotations func(*execpb.StepAnnotations)
	121	+	StepSummary func(*execpb.StepSummary)
118	122		}
119	123
120	124		// DriveJob is Drive for a session that carries a job: besides Output frames it

@@ -179,6 +183,14 @@ func DriveJob(rw io.ReadWriter, run execpb.Run, cancel <-chan execpb.Cancel, h
179	183		if h.JobResult != nil {
180	184		h.JobResult(b.JobResult)
181	185		}
	186	+	case *execpb.Frame_StepAnnotations:
	187	+	if h.StepAnnotations != nil {
	188	+	h.StepAnnotations(b.StepAnnotations)
	189	+	}
	190	+	case *execpb.Frame_StepSummary:
	191	+	if h.StepSummary != nil {
	192	+	h.StepSummary(b.StepSummary)
	193	+	}
182	194		case *execpb.Frame_Exit:
183	195		return hello, b.Exit, nil
184	196		}


@@ -115,6 +115,10 @@ type JobHandler struct {
115	StepCompleted func(*execpb.StepCompleted)	115		StepCompleted func(*execpb.StepCompleted)
116	Log func(*execpb.Log)	116		Log func(*execpb.Log)
117	JobResult func(*execpb.JobResult)	117		JobResult func(*execpb.JobResult)
		118	+	// StepAnnotations and StepSummary come before the StepCompleted of their
		119	+	// step.
		120	+	StepAnnotations func(*execpb.StepAnnotations)
		121	+	StepSummary func(*execpb.StepSummary)
118	}	122		}
119		123
120	// DriveJob is Drive for a session that carries a job: besides Output frames it	124		// DriveJob is Drive for a session that carries a job: besides Output frames it

@@ -179,6 +183,14 @@ func DriveJob(rw io.ReadWriter, run execpb.Run, cancel <-chan execpb.Cancel, h
179	if h.JobResult != nil {	183		if h.JobResult != nil {
180	h.JobResult(b.JobResult)	184		h.JobResult(b.JobResult)
181	}	185		}
		186	+	case *execpb.Frame_StepAnnotations:
		187	+	if h.StepAnnotations != nil {
		188	+	h.StepAnnotations(b.StepAnnotations)
		189	+	}
		190	+	case *execpb.Frame_StepSummary:
		191	+	if h.StepSummary != nil {
		192	+	h.StepSummary(b.StepSummary)
		193	+	}
182	case *execpb.Frame_Exit:	194		case *execpb.Frame_Exit:
183	return hello, b.Exit, nil	195		return hello, b.Exit, nil
184	}	196		}

M internal/guest/jobexec.go +48 -1


@@ -9,6 +9,7 @@ import (
9	9		"os/user"
10	10		"strconv"
11	11		"strings"
	12	+	"sync"
12	13		"syscall"
13	14		"time"
14	15

@@ -22,6 +23,9 @@ import (
22	23		// frameReporter is the executor.Reporter that writes frames on the stream.
23	24		type frameReporter struct {
24	25		conn *execstream.Conn
	26	+	// outcomes keeps steps.<id>.outcome from StepResults until StepCompleted.
	27	+	omu sync.Mutex
	28	+	outcomes map[int]executor.Conclusion
25	29		// err keeps the first send error; the job goes on without a host.
26	30		failed chan struct{}
27	31		}

@@ -65,7 +69,49 @@ func (r *frameReporter) Log(n int, at time.Time, line string) {
65	69		}
66	70
67	71		func (r *frameReporter) StepCompleted(n int, name string, c executor.Conclusion, at time.Time) {
68		-	r.send(&execpb.Frame{Body: &execpb.Frame_StepCompleted{StepCompleted: &execpb.StepCompleted{Number: int32(n), Name: name, Conclusion: conclusionProto(c), UnixNano: ts(at)}}})
	72	+	r.omu.Lock()
	73	+	outcome := r.outcomes[n]
	74	+	delete(r.outcomes, n)
	75	+	r.omu.Unlock()
	76	+	r.send(&execpb.Frame{Body: &execpb.Frame_StepCompleted{StepCompleted: &execpb.StepCompleted{
	77	+	Number: int32(n), Name: name, Conclusion: conclusionProto(c), Outcome: conclusionProto(outcome), UnixNano: ts(at),
	78	+	}}})
	79	+	}
	80	+
	81	+	func levelProto(l executor.AnnotationLevel) execpb.AnnotationLevel {
	82	+	switch l {
	83	+	case executor.LevelNotice:
	84	+	return execpb.AnnotationLevel_ANNOTATION_LEVEL_NOTICE
	85	+	case executor.LevelWarning:
	86	+	return execpb.AnnotationLevel_ANNOTATION_LEVEL_WARNING
	87	+	case executor.LevelFailure:
	88	+	return execpb.AnnotationLevel_ANNOTATION_LEVEL_FAILURE
	89	+	}
	90	+	return execpb.AnnotationLevel_ANNOTATION_LEVEL_UNSPECIFIED
	91	+	}
	92	+
	93	+	// StepResults sends the annotations and the summary of a step, and keeps its
	94	+	// outcome for the StepCompleted frame that follows.
	95	+	func (r *frameReporter) StepResults(n int, res executor.StepResults) {
	96	+	r.omu.Lock()
	97	+	if r.outcomes == nil {
	98	+	r.outcomes = map[int]executor.Conclusion{}
	99	+	}
	100	+	r.outcomes[n] = res.Outcome
	101	+	r.omu.Unlock()
	102	+	if len(res.Annotations) > 0 {
	103	+	sa := &execpb.StepAnnotations{Step: int32(n)}
	104	+	for _, a := range res.Annotations {
	105	+	sa.Annotations = append(sa.Annotations, &execpb.Annotation{
	106	+	Level: levelProto(a.Level), Title: a.Title, Message: a.Message, File: a.File,
	107	+	StartLine: a.StartLine, EndLine: a.EndLine, StartColumn: a.StartColumn, EndColumn: a.EndColumn,
	108	+	})
	109	+	}
	110	+	r.send(&execpb.Frame{Body: &execpb.Frame_StepAnnotations{StepAnnotations: sa}})
	111	+	}
	112	+	if res.HasSummary {
	113	+	r.send(&execpb.Frame{Body: &execpb.Frame_StepSummary{StepSummary: &execpb.StepSummary{Step: int32(n), Markdown: res.Summary}}})
	114	+	}
69	115		}
70	116
71	117		func (r *frameReporter) JobCompleted(res executor.Result) {

@@ -207,6 +253,7 @@ func runJob(ctx context.Context, conn execstream.Conn, cfg ExecConfig, spec ex
207	253		RunnerVersion: spec.GetRunnerVersion(),
208	254		Home: su.Home, ToolCache: os.Getenv("RUNNER_TOOL_CACHE"), BaseEnv: stepEnv(os.Environ(), su), Credential: su.Credential,
209	255		Debug: spec.GetDebug(), Reporter: rep, Log: zerowrap.FromCtx(ctx),
	256	+	SendAnnotations: spec.GetSendAnnotations(), SendSummaries: spec.GetSendSummaries(),
210	257		TermAfter: cfg.TermAfter, KillAfter: cfg.KillAfter, WaitDelay: cfg.WaitDelay,
211	258		})
212	259		if res.Conclusion == executor.ConclusionSuccess {


@@ -9,6 +9,7 @@ import (
9		"os/user"	9		"os/user"
10		"strconv"	10		"strconv"
11		"strings"	11		"strings"
			12	+	"sync"
12		"syscall"	13		"syscall"
13		"time"	14		"time"
14			15

@@ -22,6 +23,9 @@ import (
22		// frameReporter is the executor.Reporter that writes frames on the stream.	23		// frameReporter is the executor.Reporter that writes frames on the stream.
23		type frameReporter struct {	24		type frameReporter struct {
24		conn *execstream.Conn	25		conn *execstream.Conn
			26	+	// outcomes keeps steps.<id>.outcome from StepResults until StepCompleted.
			27	+	omu sync.Mutex
			28	+	outcomes map[int]executor.Conclusion
25		// err keeps the first send error; the job goes on without a host.	29		// err keeps the first send error; the job goes on without a host.
26		failed chan struct{}	30		failed chan struct{}
27		}	31		}

@@ -65,7 +69,49 @@ func (r *frameReporter) Log(n int, at time.Time, line string) {
65		}	69		}
66			70
67		func (r *frameReporter) StepCompleted(n int, name string, c executor.Conclusion, at time.Time) {	71		func (r *frameReporter) StepCompleted(n int, name string, c executor.Conclusion, at time.Time) {
68	-	r.send(&execpb.Frame{Body: &execpb.Frame_StepCompleted{StepCompleted: &execpb.StepCompleted{Number: int32(n), Name: name, Conclusion: conclusionProto(c), UnixNano: ts(at)}}})	72	+	r.omu.Lock()
			73	+	outcome := r.outcomes[n]
			74	+	delete(r.outcomes, n)
			75	+	r.omu.Unlock()
			76	+	r.send(&execpb.Frame{Body: &execpb.Frame_StepCompleted{StepCompleted: &execpb.StepCompleted{
			77	+	Number: int32(n), Name: name, Conclusion: conclusionProto(c), Outcome: conclusionProto(outcome), UnixNano: ts(at),
			78	+	}}})
			79	+	}
			80	+
			81	+	func levelProto(l executor.AnnotationLevel) execpb.AnnotationLevel {
			82	+	switch l {
			83	+	case executor.LevelNotice:
			84	+	return execpb.AnnotationLevel_ANNOTATION_LEVEL_NOTICE
			85	+	case executor.LevelWarning:
			86	+	return execpb.AnnotationLevel_ANNOTATION_LEVEL_WARNING
			87	+	case executor.LevelFailure:
			88	+	return execpb.AnnotationLevel_ANNOTATION_LEVEL_FAILURE
			89	+	}
			90	+	return execpb.AnnotationLevel_ANNOTATION_LEVEL_UNSPECIFIED
			91	+	}
			92	+
			93	+	// StepResults sends the annotations and the summary of a step, and keeps its
			94	+	// outcome for the StepCompleted frame that follows.
			95	+	func (r *frameReporter) StepResults(n int, res executor.StepResults) {
			96	+	r.omu.Lock()
			97	+	if r.outcomes == nil {
			98	+	r.outcomes = map[int]executor.Conclusion{}
			99	+	}
			100	+	r.outcomes[n] = res.Outcome
			101	+	r.omu.Unlock()
			102	+	if len(res.Annotations) > 0 {
			103	+	sa := &execpb.StepAnnotations{Step: int32(n)}
			104	+	for _, a := range res.Annotations {
			105	+	sa.Annotations = append(sa.Annotations, &execpb.Annotation{
			106	+	Level: levelProto(a.Level), Title: a.Title, Message: a.Message, File: a.File,
			107	+	StartLine: a.StartLine, EndLine: a.EndLine, StartColumn: a.StartColumn, EndColumn: a.EndColumn,
			108	+	})
			109	+	}
			110	+	r.send(&execpb.Frame{Body: &execpb.Frame_StepAnnotations{StepAnnotations: sa}})
			111	+	}
			112	+	if res.HasSummary {
			113	+	r.send(&execpb.Frame{Body: &execpb.Frame_StepSummary{StepSummary: &execpb.StepSummary{Step: int32(n), Markdown: res.Summary}}})
			114	+	}
69		}	115		}
70			116
71		func (r *frameReporter) JobCompleted(res executor.Result) {	117		func (r *frameReporter) JobCompleted(res executor.Result) {

@@ -207,6 +253,7 @@ func runJob(ctx context.Context, conn execstream.Conn, cfg ExecConfig, spec ex
207		RunnerVersion: spec.GetRunnerVersion(),	253		RunnerVersion: spec.GetRunnerVersion(),
208		Home: su.Home, ToolCache: os.Getenv("RUNNER_TOOL_CACHE"), BaseEnv: stepEnv(os.Environ(), su), Credential: su.Credential,	254		Home: su.Home, ToolCache: os.Getenv("RUNNER_TOOL_CACHE"), BaseEnv: stepEnv(os.Environ(), su), Credential: su.Credential,
209		Debug: spec.GetDebug(), Reporter: rep, Log: zerowrap.FromCtx(ctx),	255		Debug: spec.GetDebug(), Reporter: rep, Log: zerowrap.FromCtx(ctx),
			256	+	SendAnnotations: spec.GetSendAnnotations(), SendSummaries: spec.GetSendSummaries(),
210		TermAfter: cfg.TermAfter, KillAfter: cfg.KillAfter, WaitDelay: cfg.WaitDelay,	257		TermAfter: cfg.TermAfter, KillAfter: cfg.KillAfter, WaitDelay: cfg.WaitDelay,
211		})	258		})
212		if res.Conclusion == executor.ConclusionSuccess {	259		if res.Conclusion == executor.ConclusionSuccess {

M internal/host/fake_test.go +46 -5


@@ -49,6 +49,10 @@ type fakeServer struct {
49	49		lines []string // log text, one per line (mirror of store)
50	50		cont []bool
51	51		steps map[int32]*ermite.StepUpdate
	52	+	annots map[int32][]*ermite.Annotation // by step, replaced by each call that names the step
	53	+	summaries map[int32]string
	54	+	updates []*ermite.UpdateJobRequest
	55	+	updateHook func(n int, req *ermite.UpdateJobRequest) error
52	56		appendReq []*ermite.AppendLogRequest
53	57		appendHook func(n int, req *ermite.AppendLogRequest) (applyThenFail bool, err error)
54	58		appendN int

@@ -170,12 +174,47 @@ func (s fakeServer) UpdateJob(_ context.Context, r connect.Request[ermite.Upda
170	174		}
171	175		return nil, endedErr(je)
172	176		}
	177	+	s.updates = append(s.updates, r.Msg)
	178	+	if s.updateHook != nil {
	179	+	if err := s.updateHook(len(s.updates), r.Msg); err != nil {
	180	+	return nil, err
	181	+	}
	182	+	}
173	183		for _, u := range r.Msg.Steps {
174	184		s.steps[u.Number] = u
175	185		}
	186	+	s.replaceAnnotations(r.Msg.Annotations)
176	187		return connect.NewResponse(&ermite.UpdateJobResponse{}), nil
177	188		}
178	189
	190	+	// replaceAnnotations applies the replace-per-step rule of runner.proto.
	191	+	func (s fakeServer) replaceAnnotations(anns []ermite.Annotation) {
	192	+	if s.annots == nil {
	193	+	s.annots = map[int32][]*ermite.Annotation{}
	194	+	}
	195	+	named := map[int32]bool{}
	196	+	for _, a := range anns {
	197	+	if !named[a.StepNumber] {
	198	+	named[a.StepNumber] = true
	199	+	s.annots[a.StepNumber] = nil
	200	+	}
	201	+	s.annots[a.StepNumber] = append(s.annots[a.StepNumber], a)
	202	+	}
	203	+	}
	204	+
	205	+	func (s fakeServer) UploadSummary(_ context.Context, r connect.Request[ermite.UploadSummaryRequest]) (*connect.Response[ermite.UploadSummaryResponse], error) {
	206	+	s.mu.Lock()
	207	+	defer s.mu.Unlock()
	208	+	if _, err := s.checkJob(r.Header(), r.Msg.JobId); err != nil {
	209	+	return nil, err
	210	+	}
	211	+	if s.summaries == nil {
	212	+	s.summaries = map[int32]string{}
	213	+	}
	214	+	s.summaries[r.Msg.StepNumber] = r.Msg.Markdown
	215	+	return connect.NewResponse(&ermite.UploadSummaryResponse{}), nil
	216	+	}
	217	+
179	218		func (s fakeServer) AppendLog(_ context.Context, r connect.Request[ermite.AppendLogRequest]) (*connect.Response[ermite.AppendLogResponse], error) {
180	219		s.mu.Lock()
181	220		defer s.mu.Unlock()

@@ -223,6 +262,7 @@ func (s fakeServer) CompleteJob(_ context.Context, r connect.Request[ermite.Co
223	262		}
224	263		s.completeN++
225	264		s.completes = append(s.completes, r.Msg)
	265	+	s.replaceAnnotations(r.Msg.Annotations)
226	266		if s.completeHook != nil {
227	267		thenFail, err := s.completeHook(s.completeN)
228	268		if err != nil {

@@ -507,11 +547,12 @@ func newHarness(t testing.T, sc script) harness {
507	547		}
508	548		})
509	549		h.runner = &Runner{
510		-	Cfg: Config{URL: url, Name: "r1", Capacity: 1, Version: "test"},
511		-	Driver: drv,
512		-	Poll: NewClient(url, "ermiterun_tok", http.DefaultTransport),
513		-	Job: func(tok string) ermiteconnect.RunnerServiceClient { return NewClient(url, tok, http.DefaultTransport) },
514		-	Timing: fastTiming(),
	550	+	Cfg: Config{URL: url, Name: "r1", Capacity: 1, Version: "test"},
	551	+	Driver: drv,
	552	+	Poll: NewClient(url, "ermiterun_tok", http.DefaultTransport),
	553	+	Features: func(context.Context) ([]string, error) { return []string{FeatureAnnotations, FeatureSummaries}, nil },
	554	+	Job: func(tok string) ermiteconnect.RunnerServiceClient { return NewClient(url, tok, http.DefaultTransport) },
	555	+	Timing: fastTiming(),
515	556		}
516	557		return h
517	558		}


@@ -49,6 +49,10 @@ type fakeServer struct {
49		lines []string // log text, one per line (mirror of store)	49		lines []string // log text, one per line (mirror of store)
50		cont []bool	50		cont []bool
51		steps map[int32]*ermite.StepUpdate	51		steps map[int32]*ermite.StepUpdate
			52	+	annots map[int32][]*ermite.Annotation // by step, replaced by each call that names the step
			53	+	summaries map[int32]string
			54	+	updates []*ermite.UpdateJobRequest
			55	+	updateHook func(n int, req *ermite.UpdateJobRequest) error
52		appendReq []*ermite.AppendLogRequest	56		appendReq []*ermite.AppendLogRequest
53		appendHook func(n int, req *ermite.AppendLogRequest) (applyThenFail bool, err error)	57		appendHook func(n int, req *ermite.AppendLogRequest) (applyThenFail bool, err error)
54		appendN int	58		appendN int

@@ -170,12 +174,47 @@ func (s fakeServer) UpdateJob(_ context.Context, r connect.Request[ermite.Upda
170		}	174		}
171		return nil, endedErr(je)	175		return nil, endedErr(je)
172		}	176		}
			177	+	s.updates = append(s.updates, r.Msg)
			178	+	if s.updateHook != nil {
			179	+	if err := s.updateHook(len(s.updates), r.Msg); err != nil {
			180	+	return nil, err
			181	+	}
			182	+	}
173		for _, u := range r.Msg.Steps {	183		for _, u := range r.Msg.Steps {
174		s.steps[u.Number] = u	184		s.steps[u.Number] = u
175		}	185		}
			186	+	s.replaceAnnotations(r.Msg.Annotations)
176		return connect.NewResponse(&ermite.UpdateJobResponse{}), nil	187		return connect.NewResponse(&ermite.UpdateJobResponse{}), nil
177		}	188		}
178			189
			190	+	// replaceAnnotations applies the replace-per-step rule of runner.proto.
			191	+	func (s fakeServer) replaceAnnotations(anns []ermite.Annotation) {
			192	+	if s.annots == nil {
			193	+	s.annots = map[int32][]*ermite.Annotation{}
			194	+	}
			195	+	named := map[int32]bool{}
			196	+	for _, a := range anns {
			197	+	if !named[a.StepNumber] {
			198	+	named[a.StepNumber] = true
			199	+	s.annots[a.StepNumber] = nil
			200	+	}
			201	+	s.annots[a.StepNumber] = append(s.annots[a.StepNumber], a)
			202	+	}
			203	+	}
			204	+
			205	+	func (s fakeServer) UploadSummary(_ context.Context, r connect.Request[ermite.UploadSummaryRequest]) (*connect.Response[ermite.UploadSummaryResponse], error) {
			206	+	s.mu.Lock()
			207	+	defer s.mu.Unlock()
			208	+	if _, err := s.checkJob(r.Header(), r.Msg.JobId); err != nil {
			209	+	return nil, err
			210	+	}
			211	+	if s.summaries == nil {
			212	+	s.summaries = map[int32]string{}
			213	+	}
			214	+	s.summaries[r.Msg.StepNumber] = r.Msg.Markdown
			215	+	return connect.NewResponse(&ermite.UploadSummaryResponse{}), nil
			216	+	}
			217	+
179		func (s fakeServer) AppendLog(_ context.Context, r connect.Request[ermite.AppendLogRequest]) (*connect.Response[ermite.AppendLogResponse], error) {	218		func (s fakeServer) AppendLog(_ context.Context, r connect.Request[ermite.AppendLogRequest]) (*connect.Response[ermite.AppendLogResponse], error) {
180		s.mu.Lock()	219		s.mu.Lock()
181		defer s.mu.Unlock()	220		defer s.mu.Unlock()

@@ -223,6 +262,7 @@ func (s fakeServer) CompleteJob(_ context.Context, r connect.Request[ermite.Co
223		}	262		}
224		s.completeN++	263		s.completeN++
225		s.completes = append(s.completes, r.Msg)	264		s.completes = append(s.completes, r.Msg)
			265	+	s.replaceAnnotations(r.Msg.Annotations)
226		if s.completeHook != nil {	266		if s.completeHook != nil {
227		thenFail, err := s.completeHook(s.completeN)	267		thenFail, err := s.completeHook(s.completeN)
228		if err != nil {	268		if err != nil {

@@ -507,11 +547,12 @@ func newHarness(t testing.T, sc script) harness {
507		}	547		}
508		})	548		})
509		h.runner = &Runner{	549		h.runner = &Runner{
510	-	Cfg: Config{URL: url, Name: "r1", Capacity: 1, Version: "test"},	550	+	Cfg: Config{URL: url, Name: "r1", Capacity: 1, Version: "test"},
511	-	Driver: drv,	551	+	Driver: drv,
512	-	Poll: NewClient(url, "ermiterun_tok", http.DefaultTransport),	552	+	Poll: NewClient(url, "ermiterun_tok", http.DefaultTransport),
513	-	Job: func(tok string) ermiteconnect.RunnerServiceClient { return NewClient(url, tok, http.DefaultTransport) },	553	+	Features: func(context.Context) ([]string, error) { return []string{FeatureAnnotations, FeatureSummaries}, nil },
514	-	Timing: fastTiming(),	554	+	Job: func(tok string) ermiteconnect.RunnerServiceClient { return NewClient(url, tok, http.DefaultTransport) },
			555	+	Timing: fastTiming(),
515		}	556		}
516		return h	557		return h
517		}	558		}

M internal/host/hosttest/hosttest.go +49 -5


@@ -32,9 +32,13 @@ type Line struct {
32	32
33	33		// Result is what a job reported.
34	34		type Result struct {
35		-	Complete *ermite.CompleteJobRequest
36		-	Lines []Line
37		-	Steps map[int32]*ermite.StepUpdate
	35	+	// Annotations are those stored, by step (replace-per-step rule), and
	36	+	// Summaries the Markdown stored by step.
	37	+	Annotations map[int32][]*ermite.Annotation
	38	+	Summaries map[int32]string
	39	+	Complete *ermite.CompleteJobRequest
	40	+	Lines []Line
	41	+	Steps map[int32]*ermite.StepUpdate
38	42		}
39	43
40	44		// Server is the fake. Create it with New; URL is where to point the runner.

@@ -55,6 +59,8 @@ type Server struct {
55	59		}
56	60
57	61		type jobState struct {
	62	+	annots map[int32][]*ermite.Annotation
	63	+	sums map[int32]string
58	64		log *LogStore
59	65		steps map[int32]*ermite.StepUpdate
60	66		complete *ermite.CompleteJobRequest

@@ -108,7 +114,15 @@ func (s *Server) Result(jobID string) Result {
108	114		for k, v := range st.steps {
109	115		steps[k] = v
110	116		}
111		-	return Result{Complete: st.complete, Lines: st.log.Lines(), Steps: steps}
	117	+	anns := map[int32][]*ermite.Annotation{}
	118	+	for k, v := range st.annots {
	119	+	anns[k] = v
	120	+	}
	121	+	sums := map[int32]string{}
	122	+	for k, v := range st.sums {
	123	+	sums[k] = v
	124	+	}
	125	+	return Result{Complete: st.complete, Lines: st.log.Lines(), Steps: steps, Annotations: anns, Summaries: sums}
112	126		}
113	127
114	128		func (s *Server) violation(msg string) { s.Errors = append(s.Errors, msg) } // called with s.mu held

@@ -147,7 +161,7 @@ func (s Server) FetchJob(_ context.Context, r connect.Request[ermite.FetchJobR
147	161		j := s.queue[0]
148	162		s.queue = s.queue[1:]
149	163		s.assigned[j.JobToken] = j
150		-	s.jobs[j.JobId] = &jobState{log: &LogStore{OnViolation: s.violation}, steps: map[int32]*ermite.StepUpdate{}}
	164	+	s.jobs[j.JobId] = &jobState{log: &LogStore{OnViolation: s.violation}, steps: map[int32]ermite.StepUpdate{}, annots: map[int32][]ermite.Annotation{}, sums: map[int32]string{}}
151	165		return connect.NewResponse(&ermite.FetchJobResponse{Job: j}), nil
152	166		}
153	167

@@ -182,6 +196,7 @@ func (s Server) UpdateJob(_ context.Context, r connect.Request[ermite.UpdateJo
182	196		}
183	197		st.steps[u.Number] = u
184	198		}
	199	+	replace(st.annots, r.Msg.Annotations)
185	200		return connect.NewResponse(&ermite.UpdateJobResponse{}), nil
186	201		}
187	202

@@ -219,6 +234,7 @@ func (s Server) CompleteJob(_ context.Context, r connect.Request[ermite.Comple
219	234		for _, u := range r.Msg.Steps {
220	235		st.steps[u.Number] = u
221	236		}
	237	+	replace(st.annots, r.Msg.Annotations)
222	238		s.done <- r.Msg.JobId
223	239		return connect.NewResponse(&ermite.CompleteJobResponse{Conclusion: r.Msg.Conclusion, Applied: true}), nil
224	240		}

@@ -234,3 +250,31 @@ func (s Server) WatchJob(ctx context.Context, r connect.Request[ermite.WatchJo
234	250		<-ctx.Done()
235	251		return ctx.Err()
236	252		}
	253	+
	254	+	// replace applies the rule of runner.proto: the annotations of each step the
	255	+	// call names replace those stored.
	256	+	func replace(stored map[int32][]ermite.Annotation, anns []ermite.Annotation) {
	257	+	named := map[int32]bool{}
	258	+	for _, a := range anns {
	259	+	if !named[a.StepNumber] {
	260	+	named[a.StepNumber] = true
	261	+	stored[a.StepNumber] = nil
	262	+	}
	263	+	stored[a.StepNumber] = append(stored[a.StepNumber], a)
	264	+	}
	265	+	}
	266	+
	267	+	// UploadSummary stores the summary of a step.
	268	+	func (s Server) UploadSummary(_ context.Context, r connect.Request[ermite.UploadSummaryRequest]) (*connect.Response[ermite.UploadSummaryResponse], error) {
	269	+	s.mu.Lock()
	270	+	defer s.mu.Unlock()
	271	+	st, err := s.job(r.Header(), r.Msg.JobId)
	272	+	if err != nil {
	273	+	return nil, err
	274	+	}
	275	+	if st.complete != nil {
	276	+	return nil, ended(st.complete.Conclusion)
	277	+	}
	278	+	st.sums[r.Msg.StepNumber] = r.Msg.Markdown
	279	+	return connect.NewResponse(&ermite.UploadSummaryResponse{}), nil
	280	+	}

 // Result is what a job reported.
 type Result struct {
-	Complete *ermite.CompleteJobRequest
+	// Annotations are those stored, by step (replace-per-step rule), and
-	Lines    []Line
+	// Summaries the Markdown stored by step.
-	Steps    map[int32]*ermite.StepUpdate
+	Annotations map[int32][]*ermite.Annotation
+	Summaries   map[int32]string
+	Complete    *ermite.CompleteJobRequest
+	Lines       []Line
+	Steps       map[int32]*ermite.StepUpdate
 }
 // Server is the fake. Create it with New; URL is where to point the runner.
 }
 type jobState struct {
+	annots   map[int32][]*ermite.Annotation
+	sums     map[int32]string
 	log      *LogStore
 	steps    map[int32]*ermite.StepUpdate
 	complete *ermite.CompleteJobRequest
 	for k, v := range st.steps {
 		steps[k] = v
 	}
-	return Result{Complete: st.complete, Lines: st.log.Lines(), Steps: steps}
+	anns := map[int32][]*ermite.Annotation{}
+	for k, v := range st.annots {
+		anns[k] = v
+	}
+	sums := map[int32]string{}
+	for k, v := range st.sums {
+		sums[k] = v
+	}
+	return Result{Complete: st.complete, Lines: st.log.Lines(), Steps: steps, Annotations: anns, Summaries: sums}
 }
 func (s *Server) violation(msg string) { s.Errors = append(s.Errors, msg) } // called with s.mu held
 	j := s.queue[0]
 	s.queue = s.queue[1:]
 	s.assigned[j.JobToken] = j
-	s.jobs[j.JobId] = &jobState{log: &LogStore{OnViolation: s.violation}, steps: map[int32]*ermite.StepUpdate{}}
+	s.jobs[j.JobId] = &jobState{log: &LogStore{OnViolation: s.violation}, steps: map[int32]*ermite.StepUpdate{}, annots: map[int32][]*ermite.Annotation{}, sums: map[int32]string{}}
 	return connect.NewResponse(&ermite.FetchJobResponse{Job: j}), nil
 }
 		}
 		st.steps[u.Number] = u
 	}
+	replace(st.annots, r.Msg.Annotations)
 	return connect.NewResponse(&ermite.UpdateJobResponse{}), nil
 }
 	for _, u := range r.Msg.Steps {
 		st.steps[u.Number] = u
 	}
+	replace(st.annots, r.Msg.Annotations)
 	s.done <- r.Msg.JobId
 	return connect.NewResponse(&ermite.CompleteJobResponse{Conclusion: r.Msg.Conclusion, Applied: true}), nil
 }
 	<-ctx.Done()
 	return ctx.Err()
 }
+// replace applies the rule of runner.proto: the annotations of each step the
+// call names replace those stored.
+func replace(stored map[int32][]*ermite.Annotation, anns []*ermite.Annotation) {
+	named := map[int32]bool{}
+	for _, a := range anns {
+		if !named[a.StepNumber] {
+			named[a.StepNumber] = true
+			stored[a.StepNumber] = nil
+		}
+		stored[a.StepNumber] = append(stored[a.StepNumber], a)
+	}
+}
+// UploadSummary stores the summary of a step.
+func (s *Server) UploadSummary(_ context.Context, r *connect.Request[ermite.UploadSummaryRequest]) (*connect.Response[ermite.UploadSummaryResponse], error) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	st, err := s.job(r.Header(), r.Msg.JobId)
+	if err != nil {
+		return nil, err
+	}
+	if st.complete != nil {
+		return nil, ended(st.complete.Conclusion)
+	}
+	st.sums[r.Msg.StepNumber] = r.Msg.Markdown
+	return connect.NewResponse(&ermite.UploadSummaryResponse{}), nil
+}

M internal/host/job.go +23 -4


@@ -70,6 +70,10 @@ type jobRun struct {
70	70		// contexts is Job.Contexts as handed to the executor: with the server URLs
71	71		// pointed at the runner's URL when guest forge access is on.
72	72		contexts string
	73	+	// features is what the server accepts; annLists are the annotations the
	74	+	// executor sent for steps that have not completed yet (see results.go).
	75	+	features jobFeatures
	76	+	annLists map[int32][]*ermite.Annotation
73	77		// hard stops the sandbox at once; set before the session starts.
74	78		hard func()
75	79		result *execpb.JobResult

@@ -282,7 +286,7 @@ func (j jobRun) stepStarted(s execpb.StepStarted) {
282	286		u := &ermite.StepUpdate{Number: s.Number, Name: cutName(s.Name), State: ermite.StepState_STEP_STATE_IN_PROGRESS, StartedAt: tsOf(s.UnixNano)}
283	287		j.steps[s.Number] = u
284	288		j.mu.Unlock()
285		-	j.outbox.Step(u)
	289	+	j.outbox.Step(u, nil)
286	290		}
287	291
288	292		func (j jobRun) stepCompleted(s execpb.StepCompleted) {

@@ -312,9 +316,18 @@ func (j jobRun) stepCompleted(s execpb.StepCompleted) {
312	316		u.Conclusion = ermite.CiConclusion_CI_CONCLUSION_FAILURE
313	317		}
314	318		u.CompletedAt = tsOf(s.UnixNano)
	319	+	// The outcome is how the step ended before continue-on-error; a runner that
	320	+	// does not say leaves it unspecified, which means the conclusion.
	321	+	if o := conclusionOf(s.Outcome); o != ermite.CiConclusion_CI_CONCLUSION_UNSPECIFIED {
	322	+	if o == ermite.CiConclusion_CI_CONCLUSION_TIMED_OUT {
	323	+	o = ermite.CiConclusion_CI_CONCLUSION_FAILURE
	324	+	}
	325	+	u.Outcome = o
	326	+	}
315	327		j.steps[s.Number] = u
	328	+	anns := j.takeAnnotations(s.Number)
316	329		j.mu.Unlock()
317		-	j.outbox.Step(u)
	330	+	j.outbox.Step(u, anns)
318	331		}
319	332
320	333		// logLine checks and queues a line of the log. The step must be one that was

@@ -577,6 +590,7 @@ func (j jobRun) execute(ctx context.Context, rt Runner, stopHard <-chan struct
577	590		}
578	591		// Stop runs to completion even when the job context is done.
579	592		defer j.stopSandbox(ctx, rt, sb)
	593	+	j.features = j.loadFeatures(ctx, rt)
580	594
581	595		conn, err := j.startSandbox(ctx, rt, sb, stopHard)
582	596		if err != nil {

@@ -643,11 +657,13 @@ func (j jobRun) drive(conn sandbox.Conn, rt Runner) (*execpb.Exit, error) {
643	657		Matrix: j.job.Matrix, Contexts: j.contexts, Repo: j.job.Repo, TimeoutMinutes: int32(j.timeoutMinutes()),
644	658		DeadlineMs: max(1, time.Until(j.deadline()).Milliseconds()),
645	659		RunnerName: rt.Cfg.Name, RunnerVersion: rt.Cfg.Version, RunnerEnvironment: rt.Cfg.environment(),
	660	+	SendAnnotations: j.features.annotations, SendSummaries: j.features.summaries,
646	661		}}
647	662		_, exit, err := execstream.DriveJob(conn, run, j.cancelCh, execstream.JobHandler{
648	663		StepStarted: j.stepStarted, StepCompleted: j.stepCompleted,
649		-	Log: j.logLine,
650		-	JobResult: func(r *execpb.JobResult) { j.mu.Lock(); j.result = r; j.mu.Unlock() },
	664	+	Log: j.logLine,
	665	+	JobResult: func(r *execpb.JobResult) { j.mu.Lock(); j.result = r; j.mu.Unlock() },
	666	+	StepAnnotations: j.stepAnnotations, StepSummary: j.stepSummary,
651	667		})
652	668		return exit, err
653	669		}

@@ -742,6 +758,9 @@ func (j *jobRun) complete(ctx context.Context, conclusion ermite.CiConclusion, o
742	758		JobId: j.job.JobId, Conclusion: conclusion, Outputs: outputs,
743	759		RunnerShutdown: shutdown && conclusion == ermite.CiConclusion_CI_CONCLUSION_CANCELLED,
744	760		Steps: j.finalSteps(conclusion),
	761	+	// The annotations no UpdateJob has carried: all of them with the
	762	+	// step they belong to, whole lists (replace semantics).
	763	+	Annotations: j.allPendingAnnotations(),
745	764		}
746	765		err := j.sendComplete(ctx, req, conclusion)
747	766		if err == nil \|\| ctx.Err() != nil \|\| classify(err) != kindPermanent {

 	// contexts is Job.Contexts as handed to the executor: with the server URLs
 	// pointed at the runner's URL when guest forge access is on.
 	contexts string
+	// features is what the server accepts; annLists are the annotations the
+	// executor sent for steps that have not completed yet (see results.go).
+	features jobFeatures
+	annLists map[int32][]*ermite.Annotation
 	// hard stops the sandbox at once; set before the session starts.
 	hard    func()
 	result  *execpb.JobResult
 	u := &ermite.StepUpdate{Number: s.Number, Name: cutName(s.Name), State: ermite.StepState_STEP_STATE_IN_PROGRESS, StartedAt: tsOf(s.UnixNano)}
 	j.steps[s.Number] = u
 	j.mu.Unlock()
-	j.outbox.Step(u)
+	j.outbox.Step(u, nil)
 }
 func (j *jobRun) stepCompleted(s *execpb.StepCompleted) {
 		u.Conclusion = ermite.CiConclusion_CI_CONCLUSION_FAILURE
 	}
 	u.CompletedAt = tsOf(s.UnixNano)
+	// The outcome is how the step ended before continue-on-error; a runner that
+	// does not say leaves it unspecified, which means the conclusion.
+	if o := conclusionOf(s.Outcome); o != ermite.CiConclusion_CI_CONCLUSION_UNSPECIFIED {
+		if o == ermite.CiConclusion_CI_CONCLUSION_TIMED_OUT {
+			o = ermite.CiConclusion_CI_CONCLUSION_FAILURE
+		}
+		u.Outcome = o
+	}
 	j.steps[s.Number] = u
+	anns := j.takeAnnotations(s.Number)
 	j.mu.Unlock()
-	j.outbox.Step(u)
+	j.outbox.Step(u, anns)
 }
 // logLine checks and queues a line of the log. The step must be one that was
 	}
 	// Stop runs to completion even when the job context is done.
 	defer j.stopSandbox(ctx, rt, sb)
+	j.features = j.loadFeatures(ctx, rt)
 	conn, err := j.startSandbox(ctx, rt, sb, stopHard)
 	if err != nil {
 		Matrix: j.job.Matrix, Contexts: j.contexts, Repo: j.job.Repo, TimeoutMinutes: int32(j.timeoutMinutes()),
 		DeadlineMs: max(1, time.Until(j.deadline()).Milliseconds()),
 		RunnerName: rt.Cfg.Name, RunnerVersion: rt.Cfg.Version, RunnerEnvironment: rt.Cfg.environment(),
+		SendAnnotations: j.features.annotations, SendSummaries: j.features.summaries,
 	}}
 	_, exit, err := execstream.DriveJob(conn, run, j.cancelCh, execstream.JobHandler{
 		StepStarted: j.stepStarted, StepCompleted: j.stepCompleted,
-		Log:       j.logLine,
+		Log:             j.logLine,
-		JobResult: func(r *execpb.JobResult) { j.mu.Lock(); j.result = r; j.mu.Unlock() },
+		JobResult:       func(r *execpb.JobResult) { j.mu.Lock(); j.result = r; j.mu.Unlock() },
+		StepAnnotations: j.stepAnnotations, StepSummary: j.stepSummary,
 	})
 	return exit, err
 }
 		JobId: j.job.JobId, Conclusion: conclusion, Outputs: outputs,
 		RunnerShutdown: shutdown && conclusion == ermite.CiConclusion_CI_CONCLUSION_CANCELLED,
 		Steps:          j.finalSteps(conclusion),
+		// The annotations no UpdateJob has carried: all of them with the
+		// step they belong to, whole lists (replace semantics).
+		Annotations: j.allPendingAnnotations(),
 	}
 	err := j.sendComplete(ctx, req, conclusion)
 	if err == nil || ctx.Err() != nil || classify(err) != kindPermanent {

M internal/host/outbox.go +107 -10


@@ -4,6 +4,7 @@ package host
4	4
5	5		import (
6	6		"context"
	7	+	"sort"
7	8		"sync"
8	9		"time"
9	10

@@ -37,6 +38,17 @@ type logLine struct {
37	38		type event struct {
38	39		line *logLine
39	40		step *ermite.StepUpdate
	41	+	// anns are the annotations that go with a step snapshot, by step number:
	42	+	// the whole list of each step they name (replace semantics).
	43	+	anns map[int32][]*ermite.Annotation
	44	+	// summary is a step summary for UploadSummary.
	45	+	summary *summary
	46	+	}
	47	+
	48	+	// summary is the Markdown of one step.
	49	+	type summary struct {
	50	+	step int32
	51	+	markdown string
40	52		}
41	53
42	54		// outbox sends logs and step updates of one job to ermite, in order, one call

@@ -61,6 +73,11 @@ type outbox struct {
61	73		dead bool
62	74		done chan struct{}
63	75
	76	+	// pendingAnns are the annotations the server may not hold yet, by step: the
	77	+	// last list queued for the step, until an UpdateJob that carried it
	78	+	// succeeded. CompleteJob sends what is left.
	79	+	pendingAnns map[int32][]*ermite.Annotation
	80	+
64	81		offset int64 // lines the server holds, as far as we know
65	82		// offsetMaybeStale is set when a send was cut short by its context while
66	83		// the call may have been applied by the server with the response lost: the

@@ -71,6 +88,7 @@ type outbox struct {
71	88		// callClient is the part of the RunnerService client the job uses.
72	89		type callClient interface {
73	90		UpdateJob(context.Context, connect.Request[ermite.UpdateJobRequest]) (connect.Response[ermite.UpdateJobResponse], error)
	91	+	UploadSummary(context.Context, connect.Request[ermite.UploadSummaryRequest]) (connect.Response[ermite.UploadSummaryResponse], error)
74	92		AppendLog(context.Context, connect.Request[ermite.AppendLogRequest]) (connect.Response[ermite.AppendLogResponse], error)
75	93		}
76	94

@@ -121,18 +139,57 @@ func (o *outbox) Log(l logLine) {
121	139		o.cond.Broadcast()
122	140		}
123	141
124		-	// Step queues a step snapshot.
125		-	func (o outbox) Step(s ermite.StepUpdate) {
	142	+	// Step queues a step snapshot, with the annotations that changed (may be nil).
	143	+	func (o outbox) Step(s ermite.StepUpdate, anns map[int32][]*ermite.Annotation) {
	144	+	o.mu.Lock()
	145	+	if o.dead {
	146	+	o.mu.Unlock()
	147	+	return
	148	+	}
	149	+	for n, l := range anns {
	150	+	if o.pendingAnns == nil {
	151	+	o.pendingAnns = map[int32][]*ermite.Annotation{}
	152	+	}
	153	+	o.pendingAnns[n] = l
	154	+	}
	155	+	o.events = append(o.events, event{step: s, anns: anns})
	156	+	o.mu.Unlock()
	157	+	o.cond.Broadcast()
	158	+	}
	159	+
	160	+	// Summary queues the summary of a step.
	161	+	func (o *outbox) Summary(step int32, markdown string) {
126	162		o.mu.Lock()
127	163		if o.dead {
128	164		o.mu.Unlock()
129	165		return
130	166		}
131		-	o.events = append(o.events, event{step: s})
	167	+	o.events = append(o.events, event{summary: &summary{step, markdown}})
132	168		o.mu.Unlock()
133	169		o.cond.Broadcast()
134	170		}
135	171
	172	+	// PendingAnnotations returns the annotations that no successful UpdateJob has
	173	+	// carried, for CompleteJob, in step order.
	174	+	func (o outbox) PendingAnnotations() []ermite.Annotation {
	175	+	o.mu.Lock()
	176	+	defer o.mu.Unlock()
	177	+	return flattenAnnotations(o.pendingAnns)
	178	+	}
	179	+
	180	+	func flattenAnnotations(m map[int32][]ermite.Annotation) []ermite.Annotation {
	181	+	steps := make([]int32, 0, len(m))
	182	+	for n := range m {
	183	+	steps = append(steps, n)
	184	+	}
	185	+	sort.Slice(steps, func(a, b int) bool { return steps[a] < steps[b] })
	186	+	var out []*ermite.Annotation
	187	+	for _, n := range steps {
	188	+	out = append(out, m[n]...)
	189	+	}
	190	+	return out
	191	+	}
	192	+
136	193		// Close sends what is queued, then stops the sender. It waits at most until
137	194		// ctx ends.
138	195		func (o *outbox) Close(ctx context.Context) {

@@ -179,7 +236,7 @@ func (o *outbox) next(ctx context.Context) []event {
179	236		return nil
180	237		}
181	238		first := o.events[0]
182		-	if first.step != nil {
	239	+	if first.step != nil \|\| first.summary != nil {
183	240		o.events = o.events[1:]
184	241		return []event{first}
185	242		}

@@ -208,9 +265,12 @@ func (o *outbox) loop(ctx context.Context) {
208	265		if batch == nil {
209	266		return
210	267		}
211		-	if batch[0].step != nil {
212		-	o.sendStep(ctx, batch[0].step)
213		-	} else {
	268	+	switch {
	269	+	case batch[0].step != nil:
	270	+	o.sendStep(ctx, batch[0].step, batch[0].anns)
	271	+	case batch[0].summary != nil:
	272	+	o.sendSummary(ctx, batch[0].summary)
	273	+	default:
214	274		o.sendLines(ctx, batch)
215	275		}
216	276		o.mu.Lock()

@@ -251,13 +311,50 @@ func (o *outbox) retry(ctx context.Context, what string, call func(context.Conte
251	311		}
252	312		}
253	313
254		-	func (o outbox) sendStep(ctx context.Context, s ermite.StepUpdate) {
	314	+	func (o outbox) sendStep(ctx context.Context, s ermite.StepUpdate, anns map[int32][]*ermite.Annotation) {
	315	+	req := &ermite.UpdateJobRequest{JobId: o.jobID, Steps: []*ermite.StepUpdate{s}, Annotations: flattenAnnotations(anns)}
255	316		err := o.retry(ctx, "UpdateJob", func(c context.Context) error {
256		-	_, err := o.client.UpdateJob(c, connect.NewRequest(&ermite.UpdateJobRequest{JobId: o.jobID, Steps: []*ermite.StepUpdate{s}}))
	317	+	_, err := o.client.UpdateJob(c, connect.NewRequest(req))
257	318		return err
258	319		})
259		-	if err != nil && ctx.Err() == nil && classify(err) == kindPermanent {
	320	+	if err == nil {
	321	+	// The server holds these lists now, unless a later one replaced them.
	322	+	o.mu.Lock()
	323	+	for n, l := range anns {
	324	+	if cur := o.pendingAnns[n]; len(cur) == len(l) && (len(l) == 0 \|\| &cur[0] == &l[0]) {
	325	+	delete(o.pendingAnns, n)
	326	+	}
	327	+	}
	328	+	o.mu.Unlock()
	329	+	return
	330	+	}
	331	+	if ctx.Err() == nil && classify(err) == kindPermanent {
260	332		o.log.Error().Err(err).Int32("step", s.Number).Msg("UpdateJob refused")
	333	+	if len(anns) > 0 {
	334	+	// The annotations may be what the server refused: the step update
	335	+	// alone must not be lost with them.
	336	+	o.mu.Lock()
	337	+	for n := range anns {
	338	+	delete(o.pendingAnns, n)
	339	+	}
	340	+	o.mu.Unlock()
	341	+	if err := o.retry(ctx, "UpdateJob", func(c context.Context) error {
	342	+	_, err := o.client.UpdateJob(c, connect.NewRequest(&ermite.UpdateJobRequest{JobId: o.jobID, Steps: []*ermite.StepUpdate{s}}))
	343	+	return err
	344	+	}); err != nil && ctx.Err() == nil && classify(err) == kindPermanent {
	345	+	o.log.Error().Err(err).Int32("step", s.Number).Msg("UpdateJob refused without annotations")
	346	+	}
	347	+	}
	348	+	}
	349	+	}
	350	+
	351	+	func (o outbox) sendSummary(ctx context.Context, s summary) {
	352	+	err := o.retry(ctx, "UploadSummary", func(c context.Context) error {
	353	+	_, err := o.client.UploadSummary(c, connect.NewRequest(&ermite.UploadSummaryRequest{JobId: o.jobID, StepNumber: s.step, Markdown: s.markdown}))
	354	+	return err
	355	+	})
	356	+	if err != nil && ctx.Err() == nil && classify(err) == kindPermanent {
	357	+	o.log.Error().Err(err).Int32("step", s.step).Msg("UploadSummary refused, the summary is dropped")
261	358		}
262	359		}
263	360


@@ -4,6 +4,7 @@ package host
4			4
5		import (	5		import (
6		"context"	6		"context"
			7	+	"sort"
7		"sync"	8		"sync"
8		"time"	9		"time"
9			10

@@ -37,6 +38,17 @@ type logLine struct {
37		type event struct {	38		type event struct {
38		line *logLine	39		line *logLine
39		step *ermite.StepUpdate	40		step *ermite.StepUpdate
			41	+	// anns are the annotations that go with a step snapshot, by step number:
			42	+	// the whole list of each step they name (replace semantics).
			43	+	anns map[int32][]*ermite.Annotation
			44	+	// summary is a step summary for UploadSummary.
			45	+	summary *summary
			46	+	}
			47	+
			48	+	// summary is the Markdown of one step.
			49	+	type summary struct {
			50	+	step int32
			51	+	markdown string
40		}	52		}
41			53
42		// outbox sends logs and step updates of one job to ermite, in order, one call	54		// outbox sends logs and step updates of one job to ermite, in order, one call

@@ -61,6 +73,11 @@ type outbox struct {
61		dead bool	73		dead bool
62		done chan struct{}	74		done chan struct{}
63			75
			76	+	// pendingAnns are the annotations the server may not hold yet, by step: the
			77	+	// last list queued for the step, until an UpdateJob that carried it
			78	+	// succeeded. CompleteJob sends what is left.
			79	+	pendingAnns map[int32][]*ermite.Annotation
			80	+
64		offset int64 // lines the server holds, as far as we know	81		offset int64 // lines the server holds, as far as we know
65		// offsetMaybeStale is set when a send was cut short by its context while	82		// offsetMaybeStale is set when a send was cut short by its context while
66		// the call may have been applied by the server with the response lost: the	83		// the call may have been applied by the server with the response lost: the

@@ -71,6 +88,7 @@ type outbox struct {
71		// callClient is the part of the RunnerService client the job uses.	88		// callClient is the part of the RunnerService client the job uses.
72		type callClient interface {	89		type callClient interface {
73		UpdateJob(context.Context, connect.Request[ermite.UpdateJobRequest]) (connect.Response[ermite.UpdateJobResponse], error)	90		UpdateJob(context.Context, connect.Request[ermite.UpdateJobRequest]) (connect.Response[ermite.UpdateJobResponse], error)
			91	+	UploadSummary(context.Context, connect.Request[ermite.UploadSummaryRequest]) (connect.Response[ermite.UploadSummaryResponse], error)
74		AppendLog(context.Context, connect.Request[ermite.AppendLogRequest]) (connect.Response[ermite.AppendLogResponse], error)	92		AppendLog(context.Context, connect.Request[ermite.AppendLogRequest]) (connect.Response[ermite.AppendLogResponse], error)
75		}	93		}
76			94

@@ -121,18 +139,57 @@ func (o *outbox) Log(l logLine) {
121		o.cond.Broadcast()	139		o.cond.Broadcast()
122		}	140		}
123			141
124	-	// Step queues a step snapshot.	142	+	// Step queues a step snapshot, with the annotations that changed (may be nil).
125	-	func (o outbox) Step(s ermite.StepUpdate) {	143	+	func (o outbox) Step(s ermite.StepUpdate, anns map[int32][]*ermite.Annotation) {
			144	+	o.mu.Lock()
			145	+	if o.dead {
			146	+	o.mu.Unlock()
			147	+	return
			148	+	}
			149	+	for n, l := range anns {
			150	+	if o.pendingAnns == nil {
			151	+	o.pendingAnns = map[int32][]*ermite.Annotation{}
			152	+	}
			153	+	o.pendingAnns[n] = l
			154	+	}
			155	+	o.events = append(o.events, event{step: s, anns: anns})
			156	+	o.mu.Unlock()
			157	+	o.cond.Broadcast()
			158	+	}
			159	+
			160	+	// Summary queues the summary of a step.
			161	+	func (o *outbox) Summary(step int32, markdown string) {
126		o.mu.Lock()	162		o.mu.Lock()
127		if o.dead {	163		if o.dead {
128		o.mu.Unlock()	164		o.mu.Unlock()
129		return	165		return
130		}	166		}
131	-	o.events = append(o.events, event{step: s})	167	+	o.events = append(o.events, event{summary: &summary{step, markdown}})
132		o.mu.Unlock()	168		o.mu.Unlock()
133		o.cond.Broadcast()	169		o.cond.Broadcast()
134		}	170		}
135			171
			172	+	// PendingAnnotations returns the annotations that no successful UpdateJob has
			173	+	// carried, for CompleteJob, in step order.
			174	+	func (o outbox) PendingAnnotations() []ermite.Annotation {
			175	+	o.mu.Lock()
			176	+	defer o.mu.Unlock()
			177	+	return flattenAnnotations(o.pendingAnns)
			178	+	}
			179	+
			180	+	func flattenAnnotations(m map[int32][]ermite.Annotation) []ermite.Annotation {
			181	+	steps := make([]int32, 0, len(m))
			182	+	for n := range m {
			183	+	steps = append(steps, n)
			184	+	}
			185	+	sort.Slice(steps, func(a, b int) bool { return steps[a] < steps[b] })
			186	+	var out []*ermite.Annotation
			187	+	for _, n := range steps {
			188	+	out = append(out, m[n]...)
			189	+	}
			190	+	return out
			191	+	}
			192	+
136		// Close sends what is queued, then stops the sender. It waits at most until	193		// Close sends what is queued, then stops the sender. It waits at most until
137		// ctx ends.	194		// ctx ends.
138		func (o *outbox) Close(ctx context.Context) {	195		func (o *outbox) Close(ctx context.Context) {

@@ -179,7 +236,7 @@ func (o *outbox) next(ctx context.Context) []event {
179		return nil	236		return nil
180		}	237		}
181		first := o.events[0]	238		first := o.events[0]
182	-	if first.step != nil {	239	+	if first.step != nil \|\| first.summary != nil {
183		o.events = o.events[1:]	240		o.events = o.events[1:]
184		return []event{first}	241		return []event{first}
185		}	242		}

@@ -208,9 +265,12 @@ func (o *outbox) loop(ctx context.Context) {
208		if batch == nil {	265		if batch == nil {
209		return	266		return
210		}	267		}
211	-	if batch[0].step != nil {	268	+	switch {
212	-	o.sendStep(ctx, batch[0].step)	269	+	case batch[0].step != nil:
213	-	} else {	270	+	o.sendStep(ctx, batch[0].step, batch[0].anns)
			271	+	case batch[0].summary != nil:
			272	+	o.sendSummary(ctx, batch[0].summary)
			273	+	default:
214		o.sendLines(ctx, batch)	274		o.sendLines(ctx, batch)
215		}	275		}
216		o.mu.Lock()	276		o.mu.Lock()

@@ -251,13 +311,50 @@ func (o *outbox) retry(ctx context.Context, what string, call func(context.Conte
251		}	311		}
252		}	312		}
253			313
254	-	func (o outbox) sendStep(ctx context.Context, s ermite.StepUpdate) {	314	+	func (o outbox) sendStep(ctx context.Context, s ermite.StepUpdate, anns map[int32][]*ermite.Annotation) {
			315	+	req := &ermite.UpdateJobRequest{JobId: o.jobID, Steps: []*ermite.StepUpdate{s}, Annotations: flattenAnnotations(anns)}
255		err := o.retry(ctx, "UpdateJob", func(c context.Context) error {	316		err := o.retry(ctx, "UpdateJob", func(c context.Context) error {
256	-	_, err := o.client.UpdateJob(c, connect.NewRequest(&ermite.UpdateJobRequest{JobId: o.jobID, Steps: []*ermite.StepUpdate{s}}))	317	+	_, err := o.client.UpdateJob(c, connect.NewRequest(req))
257		return err	318		return err
258		})	319		})
259	-	if err != nil && ctx.Err() == nil && classify(err) == kindPermanent {	320	+	if err == nil {
			321	+	// The server holds these lists now, unless a later one replaced them.
			322	+	o.mu.Lock()
			323	+	for n, l := range anns {
			324	+	if cur := o.pendingAnns[n]; len(cur) == len(l) && (len(l) == 0 \|\| &cur[0] == &l[0]) {
			325	+	delete(o.pendingAnns, n)
			326	+	}
			327	+	}
			328	+	o.mu.Unlock()
			329	+	return
			330	+	}
			331	+	if ctx.Err() == nil && classify(err) == kindPermanent {
260		o.log.Error().Err(err).Int32("step", s.Number).Msg("UpdateJob refused")	332		o.log.Error().Err(err).Int32("step", s.Number).Msg("UpdateJob refused")
			333	+	if len(anns) > 0 {
			334	+	// The annotations may be what the server refused: the step update
			335	+	// alone must not be lost with them.
			336	+	o.mu.Lock()
			337	+	for n := range anns {
			338	+	delete(o.pendingAnns, n)
			339	+	}
			340	+	o.mu.Unlock()
			341	+	if err := o.retry(ctx, "UpdateJob", func(c context.Context) error {
			342	+	_, err := o.client.UpdateJob(c, connect.NewRequest(&ermite.UpdateJobRequest{JobId: o.jobID, Steps: []*ermite.StepUpdate{s}}))
			343	+	return err
			344	+	}); err != nil && ctx.Err() == nil && classify(err) == kindPermanent {
			345	+	o.log.Error().Err(err).Int32("step", s.Number).Msg("UpdateJob refused without annotations")
			346	+	}
			347	+	}
			348	+	}
			349	+	}
			350	+
			351	+	func (o outbox) sendSummary(ctx context.Context, s summary) {
			352	+	err := o.retry(ctx, "UploadSummary", func(c context.Context) error {
			353	+	_, err := o.client.UploadSummary(c, connect.NewRequest(&ermite.UploadSummaryRequest{JobId: o.jobID, StepNumber: s.step, Markdown: s.markdown}))
			354	+	return err
			355	+	})
			356	+	if err != nil && ctx.Err() == nil && classify(err) == kindPermanent {
			357	+	o.log.Error().Err(err).Int32("step", s.step).Msg("UploadSummary refused, the summary is dropped")
261		}	358		}
262		}	359		}
263			360

A internal/host/results.go +202 -0

@@ -0,0 +1,202 @@
	1	+	// SPDX-License-Identifier: MIT
	2	+
	3	+	package host
	4	+
	5	+	import (
	6	+	"context"
	7	+	"strings"
	8	+	"unicode/utf8"
	9	+
	10	+	"connectrpc.com/connect"
	11	+
	12	+	ermite "bnema.dev/ermite-api/gen/ermite"
	13	+
	14	+	"bnema.dev/ermite-runner/internal/execstream/execpb"
	15	+	)
	16	+
	17	+	// Features of the server that the host uses (GetVersionResponse.features).
	18	+	const (
	19	+	FeatureAnnotations = "ci.annotations"
	20	+	FeatureSummaries = "ci.summaries"
	21	+	)
	22	+
	23	+	// Limits of Annotation and UploadSummaryRequest (runner.proto). The executor
	24	+	// keeps within them; the guest is untrusted code, so the host enforces them
	25	+	// again, by dropping or cutting what is over rather than failing the job: the
	26	+	// server rejects a whole call that breaks one.
	27	+	const (
	28	+	maxAnnotationsPerLevel = 10
	29	+	maxAnnotationMessage = 4096
	30	+	maxAnnotationTitle = 255
	31	+	maxAnnotationFile = 1024
	32	+	maxSummaryBytes = 1 << 20
	33	+	)
	34	+
	35	+	// ServerFeatures asks the server for its features. It returns nil when it
	36	+	// cannot be reached: the optional results are then not sent.
	37	+	type ServerFeatures func(ctx context.Context) ([]string, error)
	38	+
	39	+	// jobFeatures is what the server accepts for this job.
	40	+	type jobFeatures struct{ annotations, summaries bool }
	41	+
	42	+	func featuresOf(list []string) jobFeatures {
	43	+	var f jobFeatures
	44	+	for _, x := range list {
	45	+	switch x {
	46	+	case FeatureAnnotations:
	47	+	f.annotations = true
	48	+	case FeatureSummaries:
	49	+	f.summaries = true
	50	+	}
	51	+	}
	52	+	return f
	53	+	}
	54	+
	55	+	// loadFeatures reads the features of the server for this job (GetVersion).
	56	+	func (j jobRun) loadFeatures(ctx context.Context, rt Runner) jobFeatures {
	57	+	if rt.Features == nil {
	58	+	return jobFeatures{}
	59	+	}
	60	+	cctx, cancel := context.WithTimeout(ctx, j.t.CallTimeout)
	61	+	defer cancel()
	62	+	list, err := rt.Features(cctx)
	63	+	if err != nil {
	64	+	j.log.Warn().Err(err).Msg("GetVersion failed: annotations and summaries are not sent for this job")
	65	+	return jobFeatures{}
	66	+	}
	67	+	return featuresOf(list)
	68	+	}
	69	+
	70	+	func levelOf(l execpb.AnnotationLevel) ermite.AnnotationLevel {
	71	+	switch l {
	72	+	case execpb.AnnotationLevel_ANNOTATION_LEVEL_NOTICE:
	73	+	return ermite.AnnotationLevel_ANNOTATION_LEVEL_NOTICE
	74	+	case execpb.AnnotationLevel_ANNOTATION_LEVEL_WARNING:
	75	+	return ermite.AnnotationLevel_ANNOTATION_LEVEL_WARNING
	76	+	case execpb.AnnotationLevel_ANNOTATION_LEVEL_FAILURE:
	77	+	return ermite.AnnotationLevel_ANNOTATION_LEVEL_FAILURE
	78	+	}
	79	+	return ermite.AnnotationLevel_ANNOTATION_LEVEL_UNSPECIFIED
	80	+	}
	81	+
	82	+	// cutRunes bounds s to n characters (code points), after making it valid UTF-8.
	83	+	func cutRunes(s string, n int) string {
	84	+	s = strings.ToValidUTF8(s, "\ufffd")
	85	+	if utf8.RuneCountInString(s) <= n {
	86	+	return s
	87	+	}
	88	+	i := 0
	89	+	for k := range s {
	90	+	if i == n {
	91	+	return s[:k]
	92	+	}
	93	+	i++
	94	+	}
	95	+	return s
	96	+	}
	97	+
	98	+	// annotationsFrom turns the annotations of a step into the ones of the
	99	+	// contract: the ones that break a rule of the contract are dropped (no level,
	100	+	// no message, a message over 4096 characters, more than ten of a level), the
	101	+	// numbers are made consistent (negative values and an end before the start
	102	+	// become "not given").
	103	+	func annotationsFrom(step int32, in []execpb.Annotation) []ermite.Annotation {
	104	+	var out []*ermite.Annotation
	105	+	count := map[ermite.AnnotationLevel]int{}
	106	+	for _, a := range in {
	107	+	lvl := levelOf(a.GetLevel())
	108	+	msg := strings.ToValidUTF8(a.GetMessage(), "\ufffd")
	109	+	if lvl == ermite.AnnotationLevel_ANNOTATION_LEVEL_UNSPECIFIED \|\| msg == "" \|\| utf8.RuneCountInString(msg) > maxAnnotationMessage {
	110	+	continue
	111	+	}
	112	+	if count[lvl] >= maxAnnotationsPerLevel {
	113	+	continue
	114	+	}
	115	+	count[lvl]++
	116	+	n := func(v int32) int32 { return max(v, 0) }
	117	+	o := &ermite.Annotation{
	118	+	Level: lvl, Title: cutRunes(a.GetTitle(), maxAnnotationTitle), Message: msg, StepNumber: step,
	119	+	StartLine: n(a.GetStartLine()), EndLine: n(a.GetEndLine()), StartColumn: n(a.GetStartColumn()), EndColumn: n(a.GetEndColumn()),
	120	+	}
	121	+	if f := strings.ToValidUTF8(a.GetFile(), "\ufffd"); utf8.RuneCountInString(f) <= maxAnnotationFile {
	122	+	o.File = f
	123	+	}
	124	+	if o.EndLine != 0 && o.EndLine < o.StartLine {
	125	+	o.EndLine = 0
	126	+	}
	127	+	out = append(out, o)
	128	+	}
	129	+	return out
	130	+	}
	131	+
	132	+	// stepAnnotations records the annotations of a step; they are sent with its
	133	+	// completion (UpdateJob replaces the annotations of the step by this list).
	134	+	func (j jobRun) stepAnnotations(a execpb.StepAnnotations) {
	135	+	j.mu.Lock()
	136	+	defer j.mu.Unlock()
	137	+	if j.violation != "" \|\| !j.features.annotations {
	138	+	return
	139	+	}
	140	+	if _, ok := j.steps[a.Step]; !ok {
	141	+	j.mu.Unlock()
	142	+	j.violate("annotations for step %d, which has not started", a.Step)
	143	+	j.mu.Lock()
	144	+	return
	145	+	}
	146	+	if j.annLists == nil {
	147	+	j.annLists = map[int32][]*ermite.Annotation{}
	148	+	}
	149	+	j.annLists[a.Step] = annotationsFrom(a.Step, a.Annotations)
	150	+	}
	151	+
	152	+	// takeAnnotations returns the list recorded for step n, once.
	153	+	func (j jobRun) takeAnnotations(n int32) map[int32][]ermite.Annotation {
	154	+	l, ok := j.annLists[n]
	155	+	if !ok {
	156	+	return nil
	157	+	}
	158	+	delete(j.annLists, n)
	159	+	return map[int32][]*ermite.Annotation{n: l}
	160	+	}
	161	+
	162	+	// stepSummary queues UploadSummary for a step that wrote GITHUB_STEP_SUMMARY.
	163	+	func (j jobRun) stepSummary(s execpb.StepSummary) {
	164	+	j.mu.Lock()
	165	+	if j.violation != "" \|\| !j.features.summaries {
	166	+	j.mu.Unlock()
	167	+	return
	168	+	}
	169	+	_, started := j.steps[s.Step]
	170	+	j.mu.Unlock()
	171	+	if !started {
	172	+	j.violate("a summary for step %d, which has not started", s.Step)
	173	+	return
	174	+	}
	175	+	md := strings.ToValidUTF8(s.Markdown, "\ufffd")
	176	+	if len(md) > maxSummaryBytes {
	177	+	j.log.Warn().Int32("step", s.Step).Int("bytes", len(md)).Msg("a step summary over 1 MiB is dropped")
	178	+	return
	179	+	}
	180	+	j.outbox.Summary(s.Step, md)
	181	+	}
	182	+
	183	+	// uploadSummaryClient is what the outbox needs for summaries.
	184	+	var _ interface {
	185	+	UploadSummary(context.Context, connect.Request[ermite.UploadSummaryRequest]) (connect.Response[ermite.UploadSummaryResponse], error)
	186	+	} = jobClient(nil)
	187	+
	188	+	// allPendingAnnotations are the annotations for CompleteJob: the lists no
	189	+	// UpdateJob has carried, and those of steps that never completed (the
	190	+	// executor was stopped), in step order.
	191	+	func (j jobRun) allPendingAnnotations() []ermite.Annotation {
	192	+	m := map[int32][]*ermite.Annotation{}
	193	+	for _, a := range j.outbox.PendingAnnotations() {
	194	+	m[a.StepNumber] = append(m[a.StepNumber], a)
	195	+	}
	196	+	j.mu.Lock()
	197	+	for n, l := range j.annLists {
	198	+	m[n] = l
	199	+	}
	200	+	j.mu.Unlock()
	201	+	return flattenAnnotations(m)
	202	+	}

@@ -0,0 +1,202 @@
	1	+	// SPDX-License-Identifier: MIT
	2	+
	3	+	package host
	4	+
	5	+	import (
	6	+	"context"
	7	+	"strings"
	8	+	"unicode/utf8"
	9	+
	10	+	"connectrpc.com/connect"
	11	+
	12	+	ermite "bnema.dev/ermite-api/gen/ermite"
	13	+
	14	+	"bnema.dev/ermite-runner/internal/execstream/execpb"
	15	+	)
	16	+
	17	+	// Features of the server that the host uses (GetVersionResponse.features).
	18	+	const (
	19	+	FeatureAnnotations = "ci.annotations"
	20	+	FeatureSummaries = "ci.summaries"
	21	+	)
	22	+
	23	+	// Limits of Annotation and UploadSummaryRequest (runner.proto). The executor
	24	+	// keeps within them; the guest is untrusted code, so the host enforces them
	25	+	// again, by dropping or cutting what is over rather than failing the job: the
	26	+	// server rejects a whole call that breaks one.
	27	+	const (
	28	+	maxAnnotationsPerLevel = 10
	29	+	maxAnnotationMessage = 4096
	30	+	maxAnnotationTitle = 255
	31	+	maxAnnotationFile = 1024
	32	+	maxSummaryBytes = 1 << 20
	33	+	)
	34	+
	35	+	// ServerFeatures asks the server for its features. It returns nil when it
	36	+	// cannot be reached: the optional results are then not sent.
	37	+	type ServerFeatures func(ctx context.Context) ([]string, error)
	38	+
	39	+	// jobFeatures is what the server accepts for this job.
	40	+	type jobFeatures struct{ annotations, summaries bool }
	41	+
	42	+	func featuresOf(list []string) jobFeatures {
	43	+	var f jobFeatures
	44	+	for _, x := range list {
	45	+	switch x {
	46	+	case FeatureAnnotations:
	47	+	f.annotations = true
	48	+	case FeatureSummaries:
	49	+	f.summaries = true
	50	+	}
	51	+	}
	52	+	return f
	53	+	}
	54	+
	55	+	// loadFeatures reads the features of the server for this job (GetVersion).
	56	+	func (j jobRun) loadFeatures(ctx context.Context, rt Runner) jobFeatures {
	57	+	if rt.Features == nil {
	58	+	return jobFeatures{}
	59	+	}
	60	+	cctx, cancel := context.WithTimeout(ctx, j.t.CallTimeout)
	61	+	defer cancel()
	62	+	list, err := rt.Features(cctx)
	63	+	if err != nil {
	64	+	j.log.Warn().Err(err).Msg("GetVersion failed: annotations and summaries are not sent for this job")
	65	+	return jobFeatures{}
	66	+	}
	67	+	return featuresOf(list)
	68	+	}
	69	+
	70	+	func levelOf(l execpb.AnnotationLevel) ermite.AnnotationLevel {
	71	+	switch l {
	72	+	case execpb.AnnotationLevel_ANNOTATION_LEVEL_NOTICE:
	73	+	return ermite.AnnotationLevel_ANNOTATION_LEVEL_NOTICE
	74	+	case execpb.AnnotationLevel_ANNOTATION_LEVEL_WARNING:
	75	+	return ermite.AnnotationLevel_ANNOTATION_LEVEL_WARNING
	76	+	case execpb.AnnotationLevel_ANNOTATION_LEVEL_FAILURE:
	77	+	return ermite.AnnotationLevel_ANNOTATION_LEVEL_FAILURE
	78	+	}
	79	+	return ermite.AnnotationLevel_ANNOTATION_LEVEL_UNSPECIFIED
	80	+	}
	81	+
	82	+	// cutRunes bounds s to n characters (code points), after making it valid UTF-8.
	83	+	func cutRunes(s string, n int) string {
	84	+	s = strings.ToValidUTF8(s, "\ufffd")
	85	+	if utf8.RuneCountInString(s) <= n {
	86	+	return s
	87	+	}
	88	+	i := 0
	89	+	for k := range s {
	90	+	if i == n {
	91	+	return s[:k]
	92	+	}
	93	+	i++
	94	+	}
	95	+	return s
	96	+	}
	97	+
	98	+	// annotationsFrom turns the annotations of a step into the ones of the
	99	+	// contract: the ones that break a rule of the contract are dropped (no level,
	100	+	// no message, a message over 4096 characters, more than ten of a level), the
	101	+	// numbers are made consistent (negative values and an end before the start
	102	+	// become "not given").
	103	+	func annotationsFrom(step int32, in []execpb.Annotation) []ermite.Annotation {
	104	+	var out []*ermite.Annotation
	105	+	count := map[ermite.AnnotationLevel]int{}
	106	+	for _, a := range in {
	107	+	lvl := levelOf(a.GetLevel())
	108	+	msg := strings.ToValidUTF8(a.GetMessage(), "\ufffd")
	109	+	if lvl == ermite.AnnotationLevel_ANNOTATION_LEVEL_UNSPECIFIED \|\| msg == "" \|\| utf8.RuneCountInString(msg) > maxAnnotationMessage {
	110	+	continue
	111	+	}
	112	+	if count[lvl] >= maxAnnotationsPerLevel {
	113	+	continue
	114	+	}
	115	+	count[lvl]++
	116	+	n := func(v int32) int32 { return max(v, 0) }
	117	+	o := &ermite.Annotation{
	118	+	Level: lvl, Title: cutRunes(a.GetTitle(), maxAnnotationTitle), Message: msg, StepNumber: step,
	119	+	StartLine: n(a.GetStartLine()), EndLine: n(a.GetEndLine()), StartColumn: n(a.GetStartColumn()), EndColumn: n(a.GetEndColumn()),
	120	+	}
	121	+	if f := strings.ToValidUTF8(a.GetFile(), "\ufffd"); utf8.RuneCountInString(f) <= maxAnnotationFile {
	122	+	o.File = f
	123	+	}
	124	+	if o.EndLine != 0 && o.EndLine < o.StartLine {
	125	+	o.EndLine = 0
	126	+	}
	127	+	out = append(out, o)
	128	+	}
	129	+	return out
	130	+	}
	131	+
	132	+	// stepAnnotations records the annotations of a step; they are sent with its
	133	+	// completion (UpdateJob replaces the annotations of the step by this list).
	134	+	func (j jobRun) stepAnnotations(a execpb.StepAnnotations) {
	135	+	j.mu.Lock()
	136	+	defer j.mu.Unlock()
	137	+	if j.violation != "" \|\| !j.features.annotations {
	138	+	return
	139	+	}
	140	+	if _, ok := j.steps[a.Step]; !ok {
	141	+	j.mu.Unlock()
	142	+	j.violate("annotations for step %d, which has not started", a.Step)
	143	+	j.mu.Lock()
	144	+	return
	145	+	}
	146	+	if j.annLists == nil {
	147	+	j.annLists = map[int32][]*ermite.Annotation{}
	148	+	}
	149	+	j.annLists[a.Step] = annotationsFrom(a.Step, a.Annotations)
	150	+	}
	151	+
	152	+	// takeAnnotations returns the list recorded for step n, once.
	153	+	func (j jobRun) takeAnnotations(n int32) map[int32][]ermite.Annotation {
	154	+	l, ok := j.annLists[n]
	155	+	if !ok {
	156	+	return nil
	157	+	}
	158	+	delete(j.annLists, n)
	159	+	return map[int32][]*ermite.Annotation{n: l}
	160	+	}
	161	+
	162	+	// stepSummary queues UploadSummary for a step that wrote GITHUB_STEP_SUMMARY.
	163	+	func (j jobRun) stepSummary(s execpb.StepSummary) {
	164	+	j.mu.Lock()
	165	+	if j.violation != "" \|\| !j.features.summaries {
	166	+	j.mu.Unlock()
	167	+	return
	168	+	}
	169	+	_, started := j.steps[s.Step]
	170	+	j.mu.Unlock()
	171	+	if !started {
	172	+	j.violate("a summary for step %d, which has not started", s.Step)
	173	+	return
	174	+	}
	175	+	md := strings.ToValidUTF8(s.Markdown, "\ufffd")
	176	+	if len(md) > maxSummaryBytes {
	177	+	j.log.Warn().Int32("step", s.Step).Int("bytes", len(md)).Msg("a step summary over 1 MiB is dropped")
	178	+	return
	179	+	}
	180	+	j.outbox.Summary(s.Step, md)
	181	+	}
	182	+
	183	+	// uploadSummaryClient is what the outbox needs for summaries.
	184	+	var _ interface {
	185	+	UploadSummary(context.Context, connect.Request[ermite.UploadSummaryRequest]) (connect.Response[ermite.UploadSummaryResponse], error)
	186	+	} = jobClient(nil)
	187	+
	188	+	// allPendingAnnotations are the annotations for CompleteJob: the lists no
	189	+	// UpdateJob has carried, and those of steps that never completed (the
	190	+	// executor was stopped), in step order.
	191	+	func (j jobRun) allPendingAnnotations() []ermite.Annotation {
	192	+	m := map[int32][]*ermite.Annotation{}
	193	+	for _, a := range j.outbox.PendingAnnotations() {
	194	+	m[a.StepNumber] = append(m[a.StepNumber], a)
	195	+	}
	196	+	j.mu.Lock()
	197	+	for n, l := range j.annLists {
	198	+	m[n] = l
	199	+	}
	200	+	j.mu.Unlock()
	201	+	return flattenAnnotations(m)
	202	+	}

A internal/host/results_test.go +308 -0

@@ -0,0 +1,308 @@
	1	+	// SPDX-License-Identifier: MIT
	2	+
	3	+	package host
	4	+
	5	+	import (
	6	+	"context"
	7	+	"errors"
	8	+	"strings"
	9	+	"sync/atomic"
	10	+	"testing"
	11	+	"time"
	12	+
	13	+	"connectrpc.com/connect"
	14	+
	15	+	ermite "bnema.dev/ermite-api/gen/ermite"
	16	+
	17	+	"bnema.dev/ermite-runner/internal/execstream"
	18	+	"bnema.dev/ermite-runner/internal/execstream/execpb"
	19	+	)
	20	+
	21	+	func annFrame(c execstream.Conn, step int, as ...execpb.Annotation) {
	22	+	send(c, &execpb.Frame{Body: &execpb.Frame_StepAnnotations{StepAnnotations: &execpb.StepAnnotations{Step: int32(step), Annotations: as}}})
	23	+	}
	24	+
	25	+	func sumFrame(c *execstream.Conn, step int, md string) {
	26	+	send(c, &execpb.Frame{Body: &execpb.Frame_StepSummary{StepSummary: &execpb.StepSummary{Step: int32(step), Markdown: md}}})
	27	+	}
	28	+
	29	+	func stepDoneOutcome(c *execstream.Conn, n int, name string, concl, outcome execpb.StepConclusion) {
	30	+	send(c, &execpb.Frame{Body: &execpb.Frame_StepCompleted{StepCompleted: &execpb.StepCompleted{Number: int32(n), Name: name, Conclusion: concl, Outcome: outcome, UnixNano: time.Now().UnixNano()}}})
	31	+	}
	32	+
	33	+	const (
	34	+	lWarn = execpb.AnnotationLevel_ANNOTATION_LEVEL_WARNING
	35	+	lFail = execpb.AnnotationLevel_ANNOTATION_LEVEL_FAILURE
	36	+	)
	37	+
	38	+	// resultsScript runs a job whose step 2 has annotations, a summary and an
	39	+	// outcome that differs from its conclusion (continue-on-error).
	40	+	func resultsScript(extra func(c *execstream.Conn)) script {
	41	+	return func(_ testing.T, c execstream.Conn, run execpb.Run, _ <-chan execpb.Cancel) {
	42	+	seenRun.Store(run)
	43	+	stepStart(c, 1, "Set up job")
	44	+	stepDone(c, 1, "Set up job", cOK)
	45	+	stepStart(c, 2, "Run it")
	46	+	logLineFrame(c, 2, "hi", false)
	47	+	annFrame(c, 2,
	48	+	&execpb.Annotation{Level: lWarn, Title: "T", Message: "careful", File: "a.go", StartLine: 3, EndLine: 4},
	49	+	&execpb.Annotation{Level: lFail, Message: "broken"})
	50	+	sumFrame(c, 2, "# Summary\n")
	51	+	stepDoneOutcome(c, 2, "Run it", cOK, cFail)
	52	+	if extra != nil {
	53	+	extra(c)
	54	+	}
	55	+	stepStart(c, 3, "Complete job")
	56	+	stepDone(c, 3, "Complete job", cOK)
	57	+	jobDone(c, cOK, nil, false)
	58	+	}
	59	+	}
	60	+
	61	+	// seenRun is the last Run frame the scripted executor received.
	62	+	var seenRun atomic.Pointer[execpb.Run]
	63	+
	64	+	func runResultsJob(t testing.T, sc script, tweak func(harness)) *harness {
	65	+	t.Helper()
	66	+	h := newHarness(t, sc)
	67	+	if tweak != nil {
	68	+	tweak(h)
	69	+	}
	70	+	h.srv.jobs = []*ermite.Job{testJob("j1")}
	71	+	stop := h.run(t)
	72	+	eventually(t, "CompleteJob", func() bool { return h.srv.completeCount() == 1 })
	73	+	if err := stop(); err != nil {
	74	+	t.Fatal(err)
	75	+	}
	76	+	return h
	77	+	}
	78	+
	79	+	func TestAnnotationsSummariesAndOutcomeAreForwarded(t *testing.T) {
	80	+	h := runResultsJob(t, resultsScript(nil), nil)
	81	+	if r := seenRun.Load(); r == nil \|\| !r.Job.SendAnnotations \|\| !r.Job.SendSummaries {
	82	+	t.Errorf("the executor was not told to send: %+v", r)
	83	+	}
	84	+	s := h.srv
	85	+	s.mu.Lock()
	86	+	defer s.mu.Unlock()
	87	+	// the annotations travel with the UpdateJob that completes the step
	88	+	var withAnn *ermite.UpdateJobRequest
	89	+	for _, u := range s.updates {
	90	+	if len(u.Annotations) > 0 {
	91	+	withAnn = u
	92	+	}
	93	+	}
	94	+	if withAnn == nil \|\| len(withAnn.Steps) != 1 \|\| withAnn.Steps[0].Number != 2 \|\| withAnn.Steps[0].State != ermite.StepState_STEP_STATE_COMPLETED {
	95	+	t.Fatalf("annotations did not ride the completion of step 2: %+v", s.updates)
	96	+	}
	97	+	got := s.annots[2]
	98	+	if len(got) != 2 \|\| got[0].Level != ermite.AnnotationLevel_ANNOTATION_LEVEL_WARNING \|\| got[0].Title != "T" \|\| got[0].File != "a.go" \|\|
	99	+	got[0].StartLine != 3 \|\| got[0].EndLine != 4 \|\| got[0].StepNumber != 2 \|\| got[1].Level != ermite.AnnotationLevel_ANNOTATION_LEVEL_FAILURE {
	100	+	t.Errorf("annotations %+v", got)
	101	+	}
	102	+	if s.summaries[2] != "# Summary\n" {
	103	+	t.Errorf("summaries %v", s.summaries)
	104	+	}
	105	+	u := s.steps[2]
	106	+	if u.Conclusion != ermite.CiConclusion_CI_CONCLUSION_SUCCESS \|\| u.Outcome != ermite.CiConclusion_CI_CONCLUSION_FAILURE {
	107	+	t.Errorf("conclusion %v outcome %v", u.Conclusion, u.Outcome)
	108	+	}
	109	+	if o := s.steps[3].Outcome; o != ermite.CiConclusion_CI_CONCLUSION_UNSPECIFIED {
	110	+	t.Errorf("a step that sends no outcome carries none: %v", o)
	111	+	}
	112	+	// nothing is left for CompleteJob when the UpdateJob went through
	113	+	if n := len(s.completes[0].Annotations); n != 0 {
	114	+	t.Errorf("CompleteJob repeats %d annotations", n)
	115	+	}
	116	+	}
	117	+
	118	+	// Without the server features, nothing is sent.
	119	+	func TestResultsAreNotSentWithoutTheFeatures(t *testing.T) {
	120	+	h := runResultsJob(t, resultsScript(nil), func(h *harness) {
	121	+	h.runner.Features = func(context.Context) ([]string, error) { return []string{"ci.runners"}, nil }
	122	+	})
	123	+	if r := seenRun.Load(); r == nil \|\| r.Job.SendAnnotations \|\| r.Job.SendSummaries {
	124	+	t.Errorf("the executor was told to send: %+v", r)
	125	+	}
	126	+	s := h.srv
	127	+	s.mu.Lock()
	128	+	defer s.mu.Unlock()
	129	+	if len(s.annots) != 0 \|\| len(s.summaries) != 0 {
	130	+	t.Errorf("annotations %v summaries %v", s.annots, s.summaries)
	131	+	}
	132	+	if len(s.completes[0].Annotations) != 0 {
	133	+	t.Error("CompleteJob carries annotations")
	134	+	}
	135	+	// the outcome is part of the StepUpdate, not of a feature
	136	+	if s.steps[2].Outcome != ermite.CiConclusion_CI_CONCLUSION_FAILURE {
	137	+	t.Error("outcome")
	138	+	}
	139	+	}
	140	+
	141	+	func TestFeaturesUnreachableMeansNothingIsSent(t *testing.T) {
	142	+	h := runResultsJob(t, resultsScript(nil), func(h *harness) {
	143	+	h.runner.Features = func(context.Context) ([]string, error) { return nil, errors.New("down") }
	144	+	})
	145	+	h.srv.mu.Lock()
	146	+	defer h.srv.mu.Unlock()
	147	+	if len(h.srv.annots) != 0 \|\| len(h.srv.summaries) != 0 {
	148	+	t.Error("sent without knowing the features")
	149	+	}
	150	+	}
	151	+
	152	+	// The host enforces the limits of the contract on what the guest sends: ten
	153	+	// per level, 4096 characters, titles of 255, valid numbers.
	154	+	func TestGuestAnnotationsAreBoundedByTheHost(t *testing.T) {
	155	+	var many []*execpb.Annotation
	156	+	for range 12 {
	157	+	many = append(many, &execpb.Annotation{Level: lWarn, Message: "w"})
	158	+	}
	159	+	many = append(many,
	160	+	&execpb.Annotation{Level: lFail, Message: strings.Repeat("é", 4097)}, // too long: dropped
	161	+	&execpb.Annotation{Level: lFail, Message: strings.Repeat("é", 4096), Title: strings.Repeat("t", 300)}, // kept, title cut
	162	+	&execpb.Annotation{Level: lFail, Message: ""}, // empty: dropped
	163	+	&execpb.Annotation{Level: execpb.AnnotationLevel_ANNOTATION_LEVEL_UNSPECIFIED, Message: "no level"}, // dropped
	164	+	&execpb.Annotation{Level: lFail, Message: "bad numbers", StartLine: -4, EndLine: -1, StartColumn: 5}, // numbers made valid
	165	+	&execpb.Annotation{Level: lFail, Message: "end before", StartLine: 9, EndLine: 3},
	166	+	)
	167	+	sc := func(_ testing.T, c execstream.Conn, _ execpb.Run, _ <-chan execpb.Cancel) {
	168	+	stepStart(c, 1, "Set up job")
	169	+	annFrame(c, 1, many...)
	170	+	stepDone(c, 1, "Set up job", cOK)
	171	+	jobDone(c, cOK, nil, false)
	172	+	}
	173	+	h := runResultsJob(t, sc, nil)
	174	+	s := h.srv
	175	+	s.mu.Lock()
	176	+	defer s.mu.Unlock()
	177	+	count := map[ermite.AnnotationLevel]int{}
	178	+	for _, a := range s.annots[1] {
	179	+	count[a.Level]++
	180	+	}
	181	+	if count[ermite.AnnotationLevel_ANNOTATION_LEVEL_WARNING] != 10 \|\| count[ermite.AnnotationLevel_ANNOTATION_LEVEL_FAILURE] != 3 {
	182	+	t.Fatalf("counts %v", count)
	183	+	}
	184	+	var cut, neg, end *ermite.Annotation
	185	+	for _, a := range s.annots[1] {
	186	+	switch a.Message {
	187	+	case strings.Repeat("é", 4096):
	188	+	cut = a
	189	+	case "bad numbers":
	190	+	neg = a
	191	+	case "end before":
	192	+	end = a
	193	+	}
	194	+	}
	195	+	if cut == nil \|\| len([]rune(cut.Title)) != 255 {
	196	+	t.Errorf("title: %v", cut)
	197	+	}
	198	+	if neg == nil \|\| neg.StartLine != 0 \|\| neg.EndLine != 0 \|\| neg.StartColumn != 5 {
	199	+	t.Errorf("%+v", neg)
	200	+	}
	201	+	if end == nil \|\| end.StartLine != 9 \|\| end.EndLine != 0 {
	202	+	t.Errorf("%+v", end)
	203	+	}
	204	+	if v := s.violations; len(v) != 0 {
	205	+	t.Error(v)
	206	+	}
	207	+	}
	208	+
	209	+	func TestGuestResultsForAnUnstartedStepFailTheJob(t *testing.T) {
	210	+	h := hostile(t, "annotations for step 9, which has not started", func(_ testing.T, c execstream.Conn, _ execpb.Run, _ <-chan execpb.Cancel) {
	211	+	stepStart(c, 1, "Set up job")
	212	+	annFrame(c, 9, &execpb.Annotation{Level: lWarn, Message: "x"})
	213	+	time.Sleep(50 * time.Millisecond)
	214	+	})
	215	+	_ = h
	216	+	}
	217	+
	218	+	// The annotations of a step that never completes (the executor is stopped) are
	219	+	// sent with CompleteJob.
	220	+	func TestAnnotationsOfAnUnfinishedStepGoWithCompleteJob(t *testing.T) {
	221	+	sc := func(_ testing.T, c execstream.Conn, _ execpb.Run, _ <-chan execpb.Cancel) {
	222	+	stepStart(c, 1, "Set up job")
	223	+	stepDone(c, 1, "Set up job", cOK)
	224	+	stepStart(c, 2, "Run it")
	225	+	annFrame(c, 2, &execpb.Annotation{Level: lWarn, Message: "last words"})
	226	+	// the executor dies here, without completing step 2 or the job
	227	+	}
	228	+	h := runResultsJob(t, sc, nil)
	229	+	h.srv.mu.Lock()
	230	+	defer h.srv.mu.Unlock()
	231	+	if len(h.srv.completes[0].Annotations) != 1 \|\| h.srv.completes[0].Annotations[0].Message != "last words" {
	232	+	t.Errorf("%+v", h.srv.completes[0].Annotations)
	233	+	}
	234	+	}
	235	+
	236	+	// UpdateJob that fails and is retried: the annotations are not duplicated,
	237	+	// since the list replaces the stored one.
	238	+	func TestAnnotationsSurviveARetriedUpdateJob(t *testing.T) {
	239	+	var fails atomic.Int32
	240	+	h := runResultsJob(t, resultsScript(nil), func(h *harness) {
	241	+	h.srv.updateHook = func(_ int, r *ermite.UpdateJobRequest) error {
	242	+	if len(r.Annotations) > 0 && fails.Add(1) <= 2 {
	243	+	return connect.NewError(connect.CodeUnavailable, nil)
	244	+	}
	245	+	return nil
	246	+	}
	247	+	})
	248	+	h.srv.mu.Lock()
	249	+	defer h.srv.mu.Unlock()
	250	+	if len(h.srv.annots[2]) != 2 {
	251	+	t.Errorf("%d annotations after retries", len(h.srv.annots[2]))
	252	+	}
	253	+	}
	254	+
	255	+	// A refused UpdateJob (invalid_argument) does not lose the step update: the
	256	+	// step is resent without its annotations.
	257	+	func TestRefusedAnnotationsDoNotLoseTheStep(t *testing.T) {
	258	+	h := runResultsJob(t, resultsScript(nil), func(h *harness) {
	259	+	h.srv.updateHook = func(_ int, r *ermite.UpdateJobRequest) error {
	260	+	if len(r.Annotations) > 0 {
	261	+	return connect.NewError(connect.CodeInvalidArgument, nil)
	262	+	}
	263	+	return nil
	264	+	}
	265	+	})
	266	+	h.srv.mu.Lock()
	267	+	defer h.srv.mu.Unlock()
	268	+	if u := h.srv.steps[2]; u == nil \|\| u.State != ermite.StepState_STEP_STATE_COMPLETED {
	269	+	t.Errorf("step 2: %v", u)
	270	+	}
	271	+	if len(h.srv.annots[2]) != 0 {
	272	+	t.Error("refused annotations stored")
	273	+	}
	274	+	if len(h.srv.completes[0].Annotations) != 0 {
	275	+	t.Error("a refused list is sent again with CompleteJob")
	276	+	}
	277	+	}
	278	+
	279	+	func TestSummaryOverOneMiBIsDropped(t *testing.T) {
	280	+	sc := func(_ testing.T, c execstream.Conn, _ execpb.Run, _ <-chan execpb.Cancel) {
	281	+	stepStart(c, 1, "Set up job")
	282	+	sumFrame(c, 1, strings.Repeat("x", 1<<20+1))
	283	+	sumFrame(c, 1, strings.Repeat("y", 1<<20))
	284	+	stepDone(c, 1, "Set up job", cOK)
	285	+	jobDone(c, cOK, nil, false)
	286	+	}
	287	+	h := runResultsJob(t, sc, nil)
	288	+	h.srv.mu.Lock()
	289	+	defer h.srv.mu.Unlock()
	290	+	if len(h.srv.summaries[1]) != 1<<20 \|\| h.srv.summaries[1][0] != 'y' {
	291	+	t.Errorf("summary of %d bytes", len(h.srv.summaries[1]))
	292	+	}
	293	+	}
	294	+
	295	+	// A message of the frame is bounded by execstream.MaxFrame: a summary of 1 MiB
	296	+	// fits.
	297	+	func TestSummaryFitsAFrame(t *testing.T) {
	298	+	f := &execpb.Frame{Body: &execpb.Frame_StepSummary{StepSummary: &execpb.StepSummary{Step: 1, Markdown: strings.Repeat("x", 1<<20)}}}
	299	+	c := execstream.NewConn(&discard{})
	300	+	if err := c.Send(f); err != nil {
	301	+	t.Fatal(err)
	302	+	}
	303	+	}
	304	+
	305	+	type discard struct{}
	306	+
	307	+	func (discard) Write(p []byte) (int, error) { return len(p), nil }
	308	+	func (discard) Read([]byte) (int, error) { select {} }

@@ -0,0 +1,308 @@
	1	+	// SPDX-License-Identifier: MIT
	2	+
	3	+	package host
	4	+
	5	+	import (
	6	+	"context"
	7	+	"errors"
	8	+	"strings"
	9	+	"sync/atomic"
	10	+	"testing"
	11	+	"time"
	12	+
	13	+	"connectrpc.com/connect"
	14	+
	15	+	ermite "bnema.dev/ermite-api/gen/ermite"
	16	+
	17	+	"bnema.dev/ermite-runner/internal/execstream"
	18	+	"bnema.dev/ermite-runner/internal/execstream/execpb"
	19	+	)
	20	+
	21	+	func annFrame(c execstream.Conn, step int, as ...execpb.Annotation) {
	22	+	send(c, &execpb.Frame{Body: &execpb.Frame_StepAnnotations{StepAnnotations: &execpb.StepAnnotations{Step: int32(step), Annotations: as}}})
	23	+	}
	24	+
	25	+	func sumFrame(c *execstream.Conn, step int, md string) {
	26	+	send(c, &execpb.Frame{Body: &execpb.Frame_StepSummary{StepSummary: &execpb.StepSummary{Step: int32(step), Markdown: md}}})
	27	+	}
	28	+
	29	+	func stepDoneOutcome(c *execstream.Conn, n int, name string, concl, outcome execpb.StepConclusion) {
	30	+	send(c, &execpb.Frame{Body: &execpb.Frame_StepCompleted{StepCompleted: &execpb.StepCompleted{Number: int32(n), Name: name, Conclusion: concl, Outcome: outcome, UnixNano: time.Now().UnixNano()}}})
	31	+	}
	32	+
	33	+	const (
	34	+	lWarn = execpb.AnnotationLevel_ANNOTATION_LEVEL_WARNING
	35	+	lFail = execpb.AnnotationLevel_ANNOTATION_LEVEL_FAILURE
	36	+	)
	37	+
	38	+	// resultsScript runs a job whose step 2 has annotations, a summary and an
	39	+	// outcome that differs from its conclusion (continue-on-error).
	40	+	func resultsScript(extra func(c *execstream.Conn)) script {
	41	+	return func(_ testing.T, c execstream.Conn, run execpb.Run, _ <-chan execpb.Cancel) {
	42	+	seenRun.Store(run)
	43	+	stepStart(c, 1, "Set up job")
	44	+	stepDone(c, 1, "Set up job", cOK)
	45	+	stepStart(c, 2, "Run it")
	46	+	logLineFrame(c, 2, "hi", false)
	47	+	annFrame(c, 2,
	48	+	&execpb.Annotation{Level: lWarn, Title: "T", Message: "careful", File: "a.go", StartLine: 3, EndLine: 4},
	49	+	&execpb.Annotation{Level: lFail, Message: "broken"})
	50	+	sumFrame(c, 2, "# Summary\n")
	51	+	stepDoneOutcome(c, 2, "Run it", cOK, cFail)
	52	+	if extra != nil {
	53	+	extra(c)
	54	+	}
	55	+	stepStart(c, 3, "Complete job")
	56	+	stepDone(c, 3, "Complete job", cOK)
	57	+	jobDone(c, cOK, nil, false)
	58	+	}
	59	+	}
	60	+
	61	+	// seenRun is the last Run frame the scripted executor received.
	62	+	var seenRun atomic.Pointer[execpb.Run]
	63	+
	64	+	func runResultsJob(t testing.T, sc script, tweak func(harness)) *harness {
	65	+	t.Helper()
	66	+	h := newHarness(t, sc)
	67	+	if tweak != nil {
	68	+	tweak(h)
	69	+	}
	70	+	h.srv.jobs = []*ermite.Job{testJob("j1")}
	71	+	stop := h.run(t)
	72	+	eventually(t, "CompleteJob", func() bool { return h.srv.completeCount() == 1 })
	73	+	if err := stop(); err != nil {
	74	+	t.Fatal(err)
	75	+	}
	76	+	return h
	77	+	}
	78	+
	79	+	func TestAnnotationsSummariesAndOutcomeAreForwarded(t *testing.T) {
	80	+	h := runResultsJob(t, resultsScript(nil), nil)
	81	+	if r := seenRun.Load(); r == nil \|\| !r.Job.SendAnnotations \|\| !r.Job.SendSummaries {
	82	+	t.Errorf("the executor was not told to send: %+v", r)
	83	+	}
	84	+	s := h.srv
	85	+	s.mu.Lock()
	86	+	defer s.mu.Unlock()
	87	+	// the annotations travel with the UpdateJob that completes the step
	88	+	var withAnn *ermite.UpdateJobRequest
	89	+	for _, u := range s.updates {
	90	+	if len(u.Annotations) > 0 {
	91	+	withAnn = u
	92	+	}
	93	+	}
	94	+	if withAnn == nil \|\| len(withAnn.Steps) != 1 \|\| withAnn.Steps[0].Number != 2 \|\| withAnn.Steps[0].State != ermite.StepState_STEP_STATE_COMPLETED {
	95	+	t.Fatalf("annotations did not ride the completion of step 2: %+v", s.updates)
	96	+	}
	97	+	got := s.annots[2]
	98	+	if len(got) != 2 \|\| got[0].Level != ermite.AnnotationLevel_ANNOTATION_LEVEL_WARNING \|\| got[0].Title != "T" \|\| got[0].File != "a.go" \|\|
	99	+	got[0].StartLine != 3 \|\| got[0].EndLine != 4 \|\| got[0].StepNumber != 2 \|\| got[1].Level != ermite.AnnotationLevel_ANNOTATION_LEVEL_FAILURE {
	100	+	t.Errorf("annotations %+v", got)
	101	+	}
	102	+	if s.summaries[2] != "# Summary\n" {
	103	+	t.Errorf("summaries %v", s.summaries)
	104	+	}
	105	+	u := s.steps[2]
	106	+	if u.Conclusion != ermite.CiConclusion_CI_CONCLUSION_SUCCESS \|\| u.Outcome != ermite.CiConclusion_CI_CONCLUSION_FAILURE {
	107	+	t.Errorf("conclusion %v outcome %v", u.Conclusion, u.Outcome)
	108	+	}
	109	+	if o := s.steps[3].Outcome; o != ermite.CiConclusion_CI_CONCLUSION_UNSPECIFIED {
	110	+	t.Errorf("a step that sends no outcome carries none: %v", o)
	111	+	}
	112	+	// nothing is left for CompleteJob when the UpdateJob went through
	113	+	if n := len(s.completes[0].Annotations); n != 0 {
	114	+	t.Errorf("CompleteJob repeats %d annotations", n)
	115	+	}
	116	+	}
	117	+
	118	+	// Without the server features, nothing is sent.
	119	+	func TestResultsAreNotSentWithoutTheFeatures(t *testing.T) {
	120	+	h := runResultsJob(t, resultsScript(nil), func(h *harness) {
	121	+	h.runner.Features = func(context.Context) ([]string, error) { return []string{"ci.runners"}, nil }
	122	+	})
	123	+	if r := seenRun.Load(); r == nil \|\| r.Job.SendAnnotations \|\| r.Job.SendSummaries {
	124	+	t.Errorf("the executor was told to send: %+v", r)
	125	+	}
	126	+	s := h.srv
	127	+	s.mu.Lock()
	128	+	defer s.mu.Unlock()
	129	+	if len(s.annots) != 0 \|\| len(s.summaries) != 0 {
	130	+	t.Errorf("annotations %v summaries %v", s.annots, s.summaries)
	131	+	}
	132	+	if len(s.completes[0].Annotations) != 0 {
	133	+	t.Error("CompleteJob carries annotations")
	134	+	}
	135	+	// the outcome is part of the StepUpdate, not of a feature
	136	+	if s.steps[2].Outcome != ermite.CiConclusion_CI_CONCLUSION_FAILURE {
	137	+	t.Error("outcome")
	138	+	}
	139	+	}
	140	+
	141	+	func TestFeaturesUnreachableMeansNothingIsSent(t *testing.T) {
	142	+	h := runResultsJob(t, resultsScript(nil), func(h *harness) {
	143	+	h.runner.Features = func(context.Context) ([]string, error) { return nil, errors.New("down") }
	144	+	})
	145	+	h.srv.mu.Lock()
	146	+	defer h.srv.mu.Unlock()
	147	+	if len(h.srv.annots) != 0 \|\| len(h.srv.summaries) != 0 {
	148	+	t.Error("sent without knowing the features")
	149	+	}
	150	+	}
	151	+
	152	+	// The host enforces the limits of the contract on what the guest sends: ten
	153	+	// per level, 4096 characters, titles of 255, valid numbers.
	154	+	func TestGuestAnnotationsAreBoundedByTheHost(t *testing.T) {
	155	+	var many []*execpb.Annotation
	156	+	for range 12 {
	157	+	many = append(many, &execpb.Annotation{Level: lWarn, Message: "w"})
	158	+	}
	159	+	many = append(many,
	160	+	&execpb.Annotation{Level: lFail, Message: strings.Repeat("é", 4097)}, // too long: dropped
	161	+	&execpb.Annotation{Level: lFail, Message: strings.Repeat("é", 4096), Title: strings.Repeat("t", 300)}, // kept, title cut
	162	+	&execpb.Annotation{Level: lFail, Message: ""}, // empty: dropped
	163	+	&execpb.Annotation{Level: execpb.AnnotationLevel_ANNOTATION_LEVEL_UNSPECIFIED, Message: "no level"}, // dropped
	164	+	&execpb.Annotation{Level: lFail, Message: "bad numbers", StartLine: -4, EndLine: -1, StartColumn: 5}, // numbers made valid
	165	+	&execpb.Annotation{Level: lFail, Message: "end before", StartLine: 9, EndLine: 3},
	166	+	)
	167	+	sc := func(_ testing.T, c execstream.Conn, _ execpb.Run, _ <-chan execpb.Cancel) {
	168	+	stepStart(c, 1, "Set up job")
	169	+	annFrame(c, 1, many...)
	170	+	stepDone(c, 1, "Set up job", cOK)
	171	+	jobDone(c, cOK, nil, false)
	172	+	}
	173	+	h := runResultsJob(t, sc, nil)
	174	+	s := h.srv
	175	+	s.mu.Lock()
	176	+	defer s.mu.Unlock()
	177	+	count := map[ermite.AnnotationLevel]int{}
	178	+	for _, a := range s.annots[1] {
	179	+	count[a.Level]++
	180	+	}
	181	+	if count[ermite.AnnotationLevel_ANNOTATION_LEVEL_WARNING] != 10 \|\| count[ermite.AnnotationLevel_ANNOTATION_LEVEL_FAILURE] != 3 {
	182	+	t.Fatalf("counts %v", count)
	183	+	}
	184	+	var cut, neg, end *ermite.Annotation
	185	+	for _, a := range s.annots[1] {
	186	+	switch a.Message {
	187	+	case strings.Repeat("é", 4096):
	188	+	cut = a
	189	+	case "bad numbers":
	190	+	neg = a
	191	+	case "end before":
	192	+	end = a
	193	+	}
	194	+	}
	195	+	if cut == nil \|\| len([]rune(cut.Title)) != 255 {
	196	+	t.Errorf("title: %v", cut)
	197	+	}
	198	+	if neg == nil \|\| neg.StartLine != 0 \|\| neg.EndLine != 0 \|\| neg.StartColumn != 5 {
	199	+	t.Errorf("%+v", neg)
	200	+	}
	201	+	if end == nil \|\| end.StartLine != 9 \|\| end.EndLine != 0 {
	202	+	t.Errorf("%+v", end)
	203	+	}
	204	+	if v := s.violations; len(v) != 0 {
	205	+	t.Error(v)
	206	+	}
	207	+	}
	208	+
	209	+	func TestGuestResultsForAnUnstartedStepFailTheJob(t *testing.T) {
	210	+	h := hostile(t, "annotations for step 9, which has not started", func(_ testing.T, c execstream.Conn, _ execpb.Run, _ <-chan execpb.Cancel) {
	211	+	stepStart(c, 1, "Set up job")
	212	+	annFrame(c, 9, &execpb.Annotation{Level: lWarn, Message: "x"})
	213	+	time.Sleep(50 * time.Millisecond)
	214	+	})
	215	+	_ = h
	216	+	}
	217	+
	218	+	// The annotations of a step that never completes (the executor is stopped) are
	219	+	// sent with CompleteJob.
	220	+	func TestAnnotationsOfAnUnfinishedStepGoWithCompleteJob(t *testing.T) {
	221	+	sc := func(_ testing.T, c execstream.Conn, _ execpb.Run, _ <-chan execpb.Cancel) {
	222	+	stepStart(c, 1, "Set up job")
	223	+	stepDone(c, 1, "Set up job", cOK)
	224	+	stepStart(c, 2, "Run it")
	225	+	annFrame(c, 2, &execpb.Annotation{Level: lWarn, Message: "last words"})
	226	+	// the executor dies here, without completing step 2 or the job
	227	+	}
	228	+	h := runResultsJob(t, sc, nil)
	229	+	h.srv.mu.Lock()
	230	+	defer h.srv.mu.Unlock()
	231	+	if len(h.srv.completes[0].Annotations) != 1 \|\| h.srv.completes[0].Annotations[0].Message != "last words" {
	232	+	t.Errorf("%+v", h.srv.completes[0].Annotations)
	233	+	}
	234	+	}
	235	+
	236	+	// UpdateJob that fails and is retried: the annotations are not duplicated,
	237	+	// since the list replaces the stored one.
	238	+	func TestAnnotationsSurviveARetriedUpdateJob(t *testing.T) {
	239	+	var fails atomic.Int32
	240	+	h := runResultsJob(t, resultsScript(nil), func(h *harness) {
	241	+	h.srv.updateHook = func(_ int, r *ermite.UpdateJobRequest) error {
	242	+	if len(r.Annotations) > 0 && fails.Add(1) <= 2 {
	243	+	return connect.NewError(connect.CodeUnavailable, nil)
	244	+	}
	245	+	return nil
	246	+	}
	247	+	})
	248	+	h.srv.mu.Lock()
	249	+	defer h.srv.mu.Unlock()
	250	+	if len(h.srv.annots[2]) != 2 {
	251	+	t.Errorf("%d annotations after retries", len(h.srv.annots[2]))
	252	+	}
	253	+	}
	254	+
	255	+	// A refused UpdateJob (invalid_argument) does not lose the step update: the
	256	+	// step is resent without its annotations.
	257	+	func TestRefusedAnnotationsDoNotLoseTheStep(t *testing.T) {
	258	+	h := runResultsJob(t, resultsScript(nil), func(h *harness) {
	259	+	h.srv.updateHook = func(_ int, r *ermite.UpdateJobRequest) error {
	260	+	if len(r.Annotations) > 0 {
	261	+	return connect.NewError(connect.CodeInvalidArgument, nil)
	262	+	}
	263	+	return nil
	264	+	}
	265	+	})
	266	+	h.srv.mu.Lock()
	267	+	defer h.srv.mu.Unlock()
	268	+	if u := h.srv.steps[2]; u == nil \|\| u.State != ermite.StepState_STEP_STATE_COMPLETED {
	269	+	t.Errorf("step 2: %v", u)
	270	+	}
	271	+	if len(h.srv.annots[2]) != 0 {
	272	+	t.Error("refused annotations stored")
	273	+	}
	274	+	if len(h.srv.completes[0].Annotations) != 0 {
	275	+	t.Error("a refused list is sent again with CompleteJob")
	276	+	}
	277	+	}
	278	+
	279	+	func TestSummaryOverOneMiBIsDropped(t *testing.T) {
	280	+	sc := func(_ testing.T, c execstream.Conn, _ execpb.Run, _ <-chan execpb.Cancel) {
	281	+	stepStart(c, 1, "Set up job")
	282	+	sumFrame(c, 1, strings.Repeat("x", 1<<20+1))
	283	+	sumFrame(c, 1, strings.Repeat("y", 1<<20))
	284	+	stepDone(c, 1, "Set up job", cOK)
	285	+	jobDone(c, cOK, nil, false)
	286	+	}
	287	+	h := runResultsJob(t, sc, nil)
	288	+	h.srv.mu.Lock()
	289	+	defer h.srv.mu.Unlock()
	290	+	if len(h.srv.summaries[1]) != 1<<20 \|\| h.srv.summaries[1][0] != 'y' {
	291	+	t.Errorf("summary of %d bytes", len(h.srv.summaries[1]))
	292	+	}
	293	+	}
	294	+
	295	+	// A message of the frame is bounded by execstream.MaxFrame: a summary of 1 MiB
	296	+	// fits.
	297	+	func TestSummaryFitsAFrame(t *testing.T) {
	298	+	f := &execpb.Frame{Body: &execpb.Frame_StepSummary{StepSummary: &execpb.StepSummary{Step: 1, Markdown: strings.Repeat("x", 1<<20)}}}
	299	+	c := execstream.NewConn(&discard{})
	300	+	if err := c.Send(f); err != nil {
	301	+	t.Fatal(err)
	302	+	}
	303	+	}
	304	+
	305	+	type discard struct{}
	306	+
	307	+	func (discard) Write(p []byte) (int, error) { return len(p), nil }
	308	+	func (discard) Read([]byte) (int, error) { select {} }

M internal/host/rpc.go +18 -0


@@ -106,6 +106,24 @@ func newClient(baseURL, token string, base http.RoundTripper, insecure bool) erm
106	106		return ermiteconnect.NewRunnerServiceClient(hc, strings.TrimRight(baseURL, "/")+APIPath)
107	107		}
108	108
	109	+	// ServerFeaturesFunc returns the function that reads the features of the
	110	+	// server with GetVersion, which needs no authentication. It is called once
	111	+	// per job.
	112	+	func (c Config) ServerFeaturesFunc(url string) ServerFeatures {
	113	+	hc := &http.Client{
	114	+	Transport: bearer{base: http.DefaultTransport, allowInsecure: c.InsecureHTTP},
	115	+	CheckRedirect: func(http.Request, []http.Request) error { return http.ErrUseLastResponse },
	116	+	}
	117	+	client := ermiteconnect.NewServerServiceClient(hc, strings.TrimRight(url, "/")+APIPath)
	118	+	return func(ctx context.Context) ([]string, error) {
	119	+	resp, err := client.GetVersion(ctx, connect.NewRequest(&ermite.GetVersionRequest{}))
	120	+	if err != nil {
	121	+	return nil, err
	122	+	}
	123	+	return resp.Msg.GetFeatures(), nil
	124	+	}
	125	+	}
	126	+
109	127		// APIPath is where ermite serves its API under the base URL, as in ermite's own
110	128		// client (internal/adapters/client.APIPath there).
111	129		const APIPath = "/-/api"


@@ -106,6 +106,24 @@ func newClient(baseURL, token string, base http.RoundTripper, insecure bool) erm
106	return ermiteconnect.NewRunnerServiceClient(hc, strings.TrimRight(baseURL, "/")+APIPath)	106		return ermiteconnect.NewRunnerServiceClient(hc, strings.TrimRight(baseURL, "/")+APIPath)
107	}	107		}
108		108
		109	+	// ServerFeaturesFunc returns the function that reads the features of the
		110	+	// server with GetVersion, which needs no authentication. It is called once
		111	+	// per job.
		112	+	func (c Config) ServerFeaturesFunc(url string) ServerFeatures {
		113	+	hc := &http.Client{
		114	+	Transport: bearer{base: http.DefaultTransport, allowInsecure: c.InsecureHTTP},
		115	+	CheckRedirect: func(http.Request, []http.Request) error { return http.ErrUseLastResponse },
		116	+	}
		117	+	client := ermiteconnect.NewServerServiceClient(hc, strings.TrimRight(url, "/")+APIPath)
		118	+	return func(ctx context.Context) ([]string, error) {
		119	+	resp, err := client.GetVersion(ctx, connect.NewRequest(&ermite.GetVersionRequest{}))
		120	+	if err != nil {
		121	+	return nil, err
		122	+	}
		123	+	return resp.Msg.GetFeatures(), nil
		124	+	}
		125	+	}
		126	+
109	// APIPath is where ermite serves its API under the base URL, as in ermite's own	127		// APIPath is where ermite serves its API under the base URL, as in ermite's own
110	// client (internal/adapters/client.APIPath there).	128		// client (internal/adapters/client.APIPath there).
111	const APIPath = "/-/api"	129		const APIPath = "/-/api"

M internal/host/runner.go +4 -0


@@ -32,6 +32,10 @@ type Runner struct {
32	32		// Job builds the client of a job from its job token.
33	33		Job func(jobToken string) ermiteconnect.RunnerServiceClient
34	34		Timing Timing
	35	+	// Features lists the features of the server (GetVersion). Annotations and
	36	+	// step summaries are sent only when it says "ci.annotations" and
	37	+	// "ci.summaries". Nil means neither is sent.
	38	+	Features ServerFeatures
35	39
36	40		// now returns the current time; tests replace it.
37	41		now func() time.Time


@@ -32,6 +32,10 @@ type Runner struct {
32	// Job builds the client of a job from its job token.	32		// Job builds the client of a job from its job token.
33	Job func(jobToken string) ermiteconnect.RunnerServiceClient	33		Job func(jobToken string) ermiteconnect.RunnerServiceClient
34	Timing Timing	34		Timing Timing
		35	+	// Features lists the features of the server (GetVersion). Annotations and
		36	+	// step summaries are sent only when it says "ci.annotations" and
		37	+	// "ci.summaries". Nil means neither is sent.
		38	+	Features ServerFeatures
35		39
36	// now returns the current time; tests replace it.	40		// now returns the current time; tests replace it.
37	now func() time.Time	41		now func() time.Time

M internal/host/shutdown_test.go +4 -0


@@ -27,6 +27,10 @@ func (c stubClient) UpdateJob(context.Context, connect.Request[ermite.UpdateJo
27	27		return connect.NewResponse(&ermite.UpdateJobResponse{}), nil
28	28		}
29	29
	30	+	func (c stubClient) UploadSummary(context.Context, connect.Request[ermite.UploadSummaryRequest]) (*connect.Response[ermite.UploadSummaryResponse], error) {
	31	+	return connect.NewResponse(&ermite.UploadSummaryResponse{}), nil
	32	+	}
	33	+
30	34		func (c stubClient) AppendLog(ctx context.Context, r connect.Request[ermite.AppendLogRequest]) (*connect.Response[ermite.AppendLogResponse], error) {
31	35		c.mu.Lock()
32	36		hang := c.hang


@@ -27,6 +27,10 @@ func (c stubClient) UpdateJob(context.Context, connect.Request[ermite.UpdateJo
27	return connect.NewResponse(&ermite.UpdateJobResponse{}), nil	27		return connect.NewResponse(&ermite.UpdateJobResponse{}), nil
28	}	28		}
29		29
		30	+	func (c stubClient) UploadSummary(context.Context, connect.Request[ermite.UploadSummaryRequest]) (*connect.Response[ermite.UploadSummaryResponse], error) {
		31	+	return connect.NewResponse(&ermite.UploadSummaryResponse{}), nil
		32	+	}
		33	+
30	func (c stubClient) AppendLog(ctx context.Context, r connect.Request[ermite.AppendLogRequest]) (*connect.Response[ermite.AppendLogResponse], error) {	34		func (c stubClient) AppendLog(ctx context.Context, r connect.Request[ermite.AppendLogRequest]) (*connect.Response[ermite.AppendLogResponse], error) {
31	c.mu.Lock()	35		c.mu.Lock()
32	hang := c.hang	36		hang := c.hang