feat(execstream): carry a job and its steps, logs and result on the frame stream

commit: aa5a961e600d37aa253759cc2e3fc0b901055600
author: brice · 10d ago · 2026-10-12 18:20 UTC
parent: e0de5de
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

4 files · +871 -48

M internal/execstream/exec.proto +79 -2


@@ -21,6 +21,10 @@ message Frame {
21	21		Output output = 3; // executor -> host
22	22		Exit exit = 4; // executor -> host, last frame
23	23		Cancel cancel = 5; // host -> executor
	24	+	StepStarted step_started = 6; // executor -> host
	25	+	StepCompleted step_completed = 7; // executor -> host
	26	+	Log log = 8; // executor -> host
	27	+	JobResult job_result = 9; // executor -> host, before Exit
24	28		}
25	29		}
26	30

@@ -30,11 +34,32 @@ message Hello {
30	34		string hostname = 3;
31	35		}
32	36
33		-	// Run asks the executor to run a script. CI-1 stub executor: the script is run
34		-	// with `sh -c`; an empty script means the built-in diagnostics script.
	37	+	// Run asks the executor to run a script or a job. Without job it is the stub
	38	+	// executor of sandbox-check: the script is run with `sh -c`; an empty script
	39	+	// means the built-in diagnostics script. With job the executor runs the job
	40	+	// like a GitHub Actions runner and reports it with StepStarted, Log,
	41	+	// StepCompleted and JobResult frames, then Exit.
35	42		message Run {
36	43		string script = 1;
37	44		map<string, string> env = 2;
	45	+	JobSpec job = 3;
	46	+	}
	47	+
	48	+	// JobSpec is the job to run: the fields of the job that ermite handed to the
	49	+	// host process, minus the tokens (the sandbox never sees them).
	50	+	message JobSpec {
	51	+	string workflow_path = 1;
	52	+	string workflow_yaml = 2;
	53	+	string job_key = 3;
	54	+	string matrix = 4; // JSON object
	55	+	string contexts = 5; // JSON object
	56	+	string repo = 6; // repository name
	57	+	int32 timeout_minutes = 7;
	58	+	string runner_name = 8;
	59	+	string runner_version = 9;
	60	+	// runner_environment is the runner.environment value ("self-hosted").
	61	+	string runner_environment = 10;
	62	+	bool debug = 11; // ACTIONS_STEP_DEBUG
38	63		}
39	64
40	65		enum StreamKind {

@@ -61,4 +86,56 @@ message Cancel {
61	86		// reason is for the job log and diagnostics ("cancelled by the user",
62	87		// "job timeout", ...); it never changes the behaviour.
63	88		string reason = 1;
	89	+	// kind tells the job executor how the job ends: cancelled, timed out, or
	90	+	// cancelled because the runner shuts down. The stub executor ignores it.
	91	+	CancelKind kind = 2;
	92	+	}
	93	+
	94	+	enum CancelKind {
	95	+	CANCEL_KIND_UNSPECIFIED = 0; // same as USER
	96	+	CANCEL_KIND_USER = 1;
	97	+	CANCEL_KIND_TIMEOUT = 2;
	98	+	CANCEL_KIND_SHUTDOWN = 3;
	99	+	}
	100	+
	101	+	// Steps are numbered like GitHub's log: "Set up job" is 1, the steps of the
	102	+	// workflow follow, "Complete job" is last.
	103	+	message StepStarted {
	104	+	int32 number = 1;
	105	+	string name = 2;
	106	+	int64 unix_nano = 3;
	107	+	}
	108	+
	109	+	message StepCompleted {
	110	+	int32 number = 1;
	111	+	string name = 2;
	112	+	StepConclusion conclusion = 3;
	113	+	int64 unix_nano = 4;
	114	+	}
	115	+
	116	+	enum StepConclusion {
	117	+	STEP_CONCLUSION_UNSPECIFIED = 0;
	118	+	STEP_CONCLUSION_SUCCESS = 1;
	119	+	STEP_CONCLUSION_FAILURE = 2;
	120	+	STEP_CONCLUSION_CANCELLED = 3;
	121	+	STEP_CONCLUSION_SKIPPED = 4;
	122	+	STEP_CONCLUSION_TIMED_OUT = 5; // jobs only
	123	+	}
	124	+
	125	+	// Log is one line of the job log, without terminator, valid UTF-8. A line of
	126	+	// more than 64 KiB is cut by the executor into pieces at character boundaries;
	127	+	// every piece but the first has continued set (the AppendLog rule).
	128	+	message Log {
	129	+	int32 step = 1;
	130	+	int64 unix_nano = 2;
	131	+	string text = 3;
	132	+	bool continued = 4;
	133	+	}
	134	+
	135	+	// JobResult is the outcome of the job; Exit follows.
	136	+	message JobResult {
	137	+	StepConclusion conclusion = 1;
	138	+	map<string, string> outputs = 2;
	139	+	string error = 3;
	140	+	bool runner_shutdown = 4;
64	141		}


@@ -21,6 +21,10 @@ message Frame {
21		Output output = 3; // executor -> host	21		Output output = 3; // executor -> host
22		Exit exit = 4; // executor -> host, last frame	22		Exit exit = 4; // executor -> host, last frame
23		Cancel cancel = 5; // host -> executor	23		Cancel cancel = 5; // host -> executor
			24	+	StepStarted step_started = 6; // executor -> host
			25	+	StepCompleted step_completed = 7; // executor -> host
			26	+	Log log = 8; // executor -> host
			27	+	JobResult job_result = 9; // executor -> host, before Exit
24		}	28		}
25		}	29		}
26			30

@@ -30,11 +34,32 @@ message Hello {
30		string hostname = 3;	34		string hostname = 3;
31		}	35		}
32			36
33	-	// Run asks the executor to run a script. CI-1 stub executor: the script is run	37	+	// Run asks the executor to run a script or a job. Without job it is the stub
34	-	// with `sh -c`; an empty script means the built-in diagnostics script.	38	+	// executor of sandbox-check: the script is run with `sh -c`; an empty script
			39	+	// means the built-in diagnostics script. With job the executor runs the job
			40	+	// like a GitHub Actions runner and reports it with StepStarted, Log,
			41	+	// StepCompleted and JobResult frames, then Exit.
35		message Run {	42		message Run {
36		string script = 1;	43		string script = 1;
37		map<string, string> env = 2;	44		map<string, string> env = 2;
			45	+	JobSpec job = 3;
			46	+	}
			47	+
			48	+	// JobSpec is the job to run: the fields of the job that ermite handed to the
			49	+	// host process, minus the tokens (the sandbox never sees them).
			50	+	message JobSpec {
			51	+	string workflow_path = 1;
			52	+	string workflow_yaml = 2;
			53	+	string job_key = 3;
			54	+	string matrix = 4; // JSON object
			55	+	string contexts = 5; // JSON object
			56	+	string repo = 6; // repository name
			57	+	int32 timeout_minutes = 7;
			58	+	string runner_name = 8;
			59	+	string runner_version = 9;
			60	+	// runner_environment is the runner.environment value ("self-hosted").
			61	+	string runner_environment = 10;
			62	+	bool debug = 11; // ACTIONS_STEP_DEBUG
38		}	63		}
39			64
40		enum StreamKind {	65		enum StreamKind {

@@ -61,4 +86,56 @@ message Cancel {
61		// reason is for the job log and diagnostics ("cancelled by the user",	86		// reason is for the job log and diagnostics ("cancelled by the user",
62		// "job timeout", ...); it never changes the behaviour.	87		// "job timeout", ...); it never changes the behaviour.
63		string reason = 1;	88		string reason = 1;
			89	+	// kind tells the job executor how the job ends: cancelled, timed out, or
			90	+	// cancelled because the runner shuts down. The stub executor ignores it.
			91	+	CancelKind kind = 2;
			92	+	}
			93	+
			94	+	enum CancelKind {
			95	+	CANCEL_KIND_UNSPECIFIED = 0; // same as USER
			96	+	CANCEL_KIND_USER = 1;
			97	+	CANCEL_KIND_TIMEOUT = 2;
			98	+	CANCEL_KIND_SHUTDOWN = 3;
			99	+	}
			100	+
			101	+	// Steps are numbered like GitHub's log: "Set up job" is 1, the steps of the
			102	+	// workflow follow, "Complete job" is last.
			103	+	message StepStarted {
			104	+	int32 number = 1;
			105	+	string name = 2;
			106	+	int64 unix_nano = 3;
			107	+	}
			108	+
			109	+	message StepCompleted {
			110	+	int32 number = 1;
			111	+	string name = 2;
			112	+	StepConclusion conclusion = 3;
			113	+	int64 unix_nano = 4;
			114	+	}
			115	+
			116	+	enum StepConclusion {
			117	+	STEP_CONCLUSION_UNSPECIFIED = 0;
			118	+	STEP_CONCLUSION_SUCCESS = 1;
			119	+	STEP_CONCLUSION_FAILURE = 2;
			120	+	STEP_CONCLUSION_CANCELLED = 3;
			121	+	STEP_CONCLUSION_SKIPPED = 4;
			122	+	STEP_CONCLUSION_TIMED_OUT = 5; // jobs only
			123	+	}
			124	+
			125	+	// Log is one line of the job log, without terminator, valid UTF-8. A line of
			126	+	// more than 64 KiB is cut by the executor into pieces at character boundaries;
			127	+	// every piece but the first has continued set (the AppendLog rule).
			128	+	message Log {
			129	+	int32 step = 1;
			130	+	int64 unix_nano = 2;
			131	+	string text = 3;
			132	+	bool continued = 4;
			133	+	}
			134	+
			135	+	// JobResult is the outcome of the job; Exit follows.
			136	+	message JobResult {
			137	+	StepConclusion conclusion = 1;
			138	+	map<string, string> outputs = 2;
			139	+	string error = 3;
			140	+	bool runner_shutdown = 4;
64		}	141		}

M internal/execstream/execpb/exec.pb.go +717 -44 generated file: collapsed

load diff

M internal/execstream/frame.go +39 -2


@@ -100,6 +100,24 @@ func (c Conn) Recv() (execpb.Frame, error) {
100	100		// a Cancel; the caller bounds that wait by stopping the sandbox, which closes
101	101		// the stream and makes Drive return an error.
102	102		func Drive(rw io.ReadWriter, run execpb.Run, cancel <-chan execpb.Cancel, onOutput func(execpb.Output)) (execpb.Hello, *execpb.Exit, error) {
	103	+	return DriveJob(rw, run, cancel, JobHandler{Output: onOutput})
	104	+	}
	105	+
	106	+	// JobHandler receives the frames of a session. Nil functions ignore their
	107	+	// frames. The functions are called from the goroutine of DriveJob, in the
	108	+	// order of the stream.
	109	+	type JobHandler struct {
	110	+	Hello func(*execpb.Hello)
	111	+	Output func(*execpb.Output)
	112	+	StepStarted func(*execpb.StepStarted)
	113	+	StepCompleted func(*execpb.StepCompleted)
	114	+	Log func(*execpb.Log)
	115	+	JobResult func(*execpb.JobResult)
	116	+	}
	117	+
	118	+	// DriveJob is Drive for a session that carries a job: besides Output frames it
	119	+	// delivers the step, log and result frames to h.
	120	+	func DriveJob(rw io.ReadWriter, run execpb.Run, cancel <-chan execpb.Cancel, h JobHandler) (execpb.Hello, execpb.Exit, error) {
103	121		c := NewConn(rw)
104	122		first, err := c.Recv()
105	123		if err != nil {

@@ -109,6 +127,9 @@ func Drive(rw io.ReadWriter, run execpb.Run, cancel <-chan execpb.Cancel, onOu
109	127		if hello == nil {
110	128		return nil, nil, fmt.Errorf("execstream: first frame is not hello")
111	129		}
	130	+	if h.Hello != nil {
	131	+	h.Hello(hello)
	132	+	}
112	133		if err := c.Send(&execpb.Frame{Body: &execpb.Frame_Run{Run: run}}); err != nil {
113	134		return hello, nil, fmt.Errorf("execstream: sending run: %w", err)
114	135		}

@@ -137,8 +158,24 @@ func Drive(rw io.ReadWriter, run execpb.Run, cancel <-chan execpb.Cancel, onOu
137	158		}
138	159		switch b := f.Body.(type) {
139	160		case *execpb.Frame_Output:
140		-	if onOutput != nil {
141		-	onOutput(b.Output)
	161	+	if h.Output != nil {
	162	+	h.Output(b.Output)
	163	+	}
	164	+	case *execpb.Frame_StepStarted:
	165	+	if h.StepStarted != nil {
	166	+	h.StepStarted(b.StepStarted)
	167	+	}
	168	+	case *execpb.Frame_StepCompleted:
	169	+	if h.StepCompleted != nil {
	170	+	h.StepCompleted(b.StepCompleted)
	171	+	}
	172	+	case *execpb.Frame_Log:
	173	+	if h.Log != nil {
	174	+	h.Log(b.Log)
	175	+	}
	176	+	case *execpb.Frame_JobResult:
	177	+	if h.JobResult != nil {
	178	+	h.JobResult(b.JobResult)
142	179		}
143	180		case *execpb.Frame_Exit:
144	181		return hello, b.Exit, nil


@@ -100,6 +100,24 @@ func (c Conn) Recv() (execpb.Frame, error) {
100		// a Cancel; the caller bounds that wait by stopping the sandbox, which closes	100		// a Cancel; the caller bounds that wait by stopping the sandbox, which closes
101		// the stream and makes Drive return an error.	101		// the stream and makes Drive return an error.
102		func Drive(rw io.ReadWriter, run execpb.Run, cancel <-chan execpb.Cancel, onOutput func(execpb.Output)) (execpb.Hello, *execpb.Exit, error) {	102		func Drive(rw io.ReadWriter, run execpb.Run, cancel <-chan execpb.Cancel, onOutput func(execpb.Output)) (execpb.Hello, *execpb.Exit, error) {
			103	+	return DriveJob(rw, run, cancel, JobHandler{Output: onOutput})
			104	+	}
			105	+
			106	+	// JobHandler receives the frames of a session. Nil functions ignore their
			107	+	// frames. The functions are called from the goroutine of DriveJob, in the
			108	+	// order of the stream.
			109	+	type JobHandler struct {
			110	+	Hello func(*execpb.Hello)
			111	+	Output func(*execpb.Output)
			112	+	StepStarted func(*execpb.StepStarted)
			113	+	StepCompleted func(*execpb.StepCompleted)
			114	+	Log func(*execpb.Log)
			115	+	JobResult func(*execpb.JobResult)
			116	+	}
			117	+
			118	+	// DriveJob is Drive for a session that carries a job: besides Output frames it
			119	+	// delivers the step, log and result frames to h.
			120	+	func DriveJob(rw io.ReadWriter, run execpb.Run, cancel <-chan execpb.Cancel, h JobHandler) (execpb.Hello, execpb.Exit, error) {
103		c := NewConn(rw)	121		c := NewConn(rw)
104		first, err := c.Recv()	122		first, err := c.Recv()
105		if err != nil {	123		if err != nil {

@@ -109,6 +127,9 @@ func Drive(rw io.ReadWriter, run execpb.Run, cancel <-chan execpb.Cancel, onOu
109		if hello == nil {	127		if hello == nil {
110		return nil, nil, fmt.Errorf("execstream: first frame is not hello")	128		return nil, nil, fmt.Errorf("execstream: first frame is not hello")
111		}	129		}
			130	+	if h.Hello != nil {
			131	+	h.Hello(hello)
			132	+	}
112		if err := c.Send(&execpb.Frame{Body: &execpb.Frame_Run{Run: run}}); err != nil {	133		if err := c.Send(&execpb.Frame{Body: &execpb.Frame_Run{Run: run}}); err != nil {
113		return hello, nil, fmt.Errorf("execstream: sending run: %w", err)	134		return hello, nil, fmt.Errorf("execstream: sending run: %w", err)
114		}	135		}

@@ -137,8 +158,24 @@ func Drive(rw io.ReadWriter, run execpb.Run, cancel <-chan execpb.Cancel, onOu
137		}	158		}
138		switch b := f.Body.(type) {	159		switch b := f.Body.(type) {
139		case *execpb.Frame_Output:	160		case *execpb.Frame_Output:
140	-	if onOutput != nil {	161	+	if h.Output != nil {
141	-	onOutput(b.Output)	162	+	h.Output(b.Output)
			163	+	}
			164	+	case *execpb.Frame_StepStarted:
			165	+	if h.StepStarted != nil {
			166	+	h.StepStarted(b.StepStarted)
			167	+	}
			168	+	case *execpb.Frame_StepCompleted:
			169	+	if h.StepCompleted != nil {
			170	+	h.StepCompleted(b.StepCompleted)
			171	+	}
			172	+	case *execpb.Frame_Log:
			173	+	if h.Log != nil {
			174	+	h.Log(b.Log)
			175	+	}
			176	+	case *execpb.Frame_JobResult:
			177	+	if h.JobResult != nil {
			178	+	h.JobResult(b.JobResult)
142		}	179		}
143		case *execpb.Frame_Exit:	180		case *execpb.Frame_Exit:
144		return hello, b.Exit, nil	181		return hello, b.Exit, nil

M internal/execstream/frame_test.go +36 -0


@@ -8,6 +8,7 @@ import (
8	8		"errors"
9	9		"io"
10	10		"net"
	11	+	"strings"
11	12		"testing"
12	13		"testing/iotest"
13	14		"time"

@@ -136,3 +137,38 @@ func TestDriveHelloMissing(t *testing.T) {
136	137		t.Fatal("want error when the first frame is not hello")
137	138		}
138	139		}
	140	+
	141	+	func TestDriveJobDeliversJobFrames(t *testing.T) {
	142	+	host, guest := net.Pipe()
	143	+	go func() {
	144	+	g := NewConn(guest)
	145	+	_ = g.Send(&execpb.Frame{Body: &execpb.Frame_Hello{Hello: &execpb.Hello{Version: "v"}}})
	146	+	r, _ := g.Recv()
	147	+	if r.GetRun().GetJob().GetJobKey() != "build" {
	148	+	t.Errorf("run: %v", r)
	149	+	}
	150	+	for _, b := range []*execpb.Frame{
	151	+	{Body: &execpb.Frame_StepStarted{StepStarted: &execpb.StepStarted{Number: 1, Name: "Set up job"}}},
	152	+	{Body: &execpb.Frame_Log{Log: &execpb.Log{Step: 1, Text: "hello", UnixNano: 5}}},
	153	+	{Body: &execpb.Frame_StepCompleted{StepCompleted: &execpb.StepCompleted{Number: 1, Conclusion: execpb.StepConclusion_STEP_CONCLUSION_SUCCESS}}},
	154	+	{Body: &execpb.Frame_JobResult{JobResult: &execpb.JobResult{Conclusion: execpb.StepConclusion_STEP_CONCLUSION_SUCCESS, Outputs: map[string]string{"a": "b"}}}},
	155	+	{Body: &execpb.Frame_Exit{Exit: &execpb.Exit{}}},
	156	+	} {
	157	+	_ = g.Send(b)
	158	+	}
	159	+	}()
	160	+	var got []string
	161	+	_, exit, err := DriveJob(host, &execpb.Run{Job: &execpb.JobSpec{JobKey: "build"}}, nil, JobHandler{
	162	+	StepStarted: func(s *execpb.StepStarted) { got = append(got, "start:"+s.Name) },
	163	+	Log: func(l *execpb.Log) { got = append(got, "log:"+l.Text) },
	164	+	StepCompleted: func(s *execpb.StepCompleted) { got = append(got, "done:"+s.Conclusion.String()) },
	165	+	JobResult: func(r *execpb.JobResult) { got = append(got, "result:"+r.Outputs["a"]) },
	166	+	})
	167	+	if err != nil \|\| exit == nil {
	168	+	t.Fatal(err)
	169	+	}
	170	+	want := "start:Set up job log:hello done:STEP_CONCLUSION_SUCCESS result:b"
	171	+	if strings.Join(got, " ") != want {
	172	+	t.Fatalf("%v", got)
	173	+	}
	174	+	}


@@ -8,6 +8,7 @@ import (
8	"errors"	8		"errors"
9	"io"	9		"io"
10	"net"	10		"net"
		11	+	"strings"
11	"testing"	12		"testing"
12	"testing/iotest"	13		"testing/iotest"
13	"time"	14		"time"

@@ -136,3 +137,38 @@ func TestDriveHelloMissing(t *testing.T) {
136	t.Fatal("want error when the first frame is not hello")	137		t.Fatal("want error when the first frame is not hello")
137	}	138		}
138	}	139		}
		140	+
		141	+	func TestDriveJobDeliversJobFrames(t *testing.T) {
		142	+	host, guest := net.Pipe()
		143	+	go func() {
		144	+	g := NewConn(guest)
		145	+	_ = g.Send(&execpb.Frame{Body: &execpb.Frame_Hello{Hello: &execpb.Hello{Version: "v"}}})
		146	+	r, _ := g.Recv()
		147	+	if r.GetRun().GetJob().GetJobKey() != "build" {
		148	+	t.Errorf("run: %v", r)
		149	+	}
		150	+	for _, b := range []*execpb.Frame{
		151	+	{Body: &execpb.Frame_StepStarted{StepStarted: &execpb.StepStarted{Number: 1, Name: "Set up job"}}},
		152	+	{Body: &execpb.Frame_Log{Log: &execpb.Log{Step: 1, Text: "hello", UnixNano: 5}}},
		153	+	{Body: &execpb.Frame_StepCompleted{StepCompleted: &execpb.StepCompleted{Number: 1, Conclusion: execpb.StepConclusion_STEP_CONCLUSION_SUCCESS}}},
		154	+	{Body: &execpb.Frame_JobResult{JobResult: &execpb.JobResult{Conclusion: execpb.StepConclusion_STEP_CONCLUSION_SUCCESS, Outputs: map[string]string{"a": "b"}}}},
		155	+	{Body: &execpb.Frame_Exit{Exit: &execpb.Exit{}}},
		156	+	} {
		157	+	_ = g.Send(b)
		158	+	}
		159	+	}()
		160	+	var got []string
		161	+	_, exit, err := DriveJob(host, &execpb.Run{Job: &execpb.JobSpec{JobKey: "build"}}, nil, JobHandler{
		162	+	StepStarted: func(s *execpb.StepStarted) { got = append(got, "start:"+s.Name) },
		163	+	Log: func(l *execpb.Log) { got = append(got, "log:"+l.Text) },
		164	+	StepCompleted: func(s *execpb.StepCompleted) { got = append(got, "done:"+s.Conclusion.String()) },
		165	+	JobResult: func(r *execpb.JobResult) { got = append(got, "result:"+r.Outputs["a"]) },
		166	+	})
		167	+	if err != nil \|\| exit == nil {
		168	+	t.Fatal(err)
		169	+	}
		170	+	want := "start:Set up job log:hello done:STEP_CONCLUSION_SUCCESS result:b"
		171	+	if strings.Join(got, " ") != want {
		172	+	t.Fatalf("%v", got)
		173	+	}
		174	+	}