feat(guest): run jobs with the executor as the runner user; the stub stays for runs without a job

commit: 90f106aa598e826213cab277a7d6e6714ddf4a22
author: brice · 9d ago · 2026-10-12 18:22 UTC
parent: aa5a961
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

5 files · +492 -4

M internal/executor/filecmd.go +5 -0


@@ -93,6 +93,8 @@ type Sink interface {
93	93		type FileCommands struct {
94	94		dir string
95	95		suffix string
	96	+	// Own, when set, gives a file or directory to the user of the steps.
	97	+	Own func(path string)
96	98		}
97	99
98	100		// NewFileCommands uses dir (RUNNER_TEMP/_runner_file_commands), creating it.

@@ -125,6 +127,9 @@ func (f *FileCommands) Init() (map[string]string, error) {
125	127		return nil, err
126	128		}
127	129		_ = fh.Close()
	130	+	if f.Own != nil {
	131	+	f.Own(p)
	132	+	}
128	133		env[envVarName(c.ContextName)] = p
129	134		}
130	135		return env, nil


@@ -93,6 +93,8 @@ type Sink interface {
93	type FileCommands struct {	93		type FileCommands struct {
94	dir string	94		dir string
95	suffix string	95		suffix string
		96	+	// Own, when set, gives a file or directory to the user of the steps.
		97	+	Own func(path string)
96	}	98		}
97		99
98	// NewFileCommands uses dir (RUNNER_TEMP/_runner_file_commands), creating it.	100		// NewFileCommands uses dir (RUNNER_TEMP/_runner_file_commands), creating it.

@@ -125,6 +127,9 @@ func (f *FileCommands) Init() (map[string]string, error) {
125	return nil, err	127		return nil, err
126	}	128		}
127	_ = fh.Close()	129		_ = fh.Close()
		130	+	if f.Own != nil {
		131	+	f.Own(p)
		132	+	}
128	env[envVarName(c.ContextName)] = p	133		env[envVarName(c.ContextName)] = p
129	}	134		}
130	return env, nil	135		return env, nil

M internal/executor/steps.go +32 -2


@@ -52,8 +52,15 @@ type Config struct {
52	52
53	53		Runner RunnerInfo
54	54		RunnerVersion string
55		-	// Layout overrides the directory layout (tests); nil means the hosted layout.
	55	+	// Layout overrides the directory layout (tests); nil means the hosted layout
	56	+	// under Home.
56	57		Layout *Layout
	58	+	// Home is the home directory of the user of the steps; empty means
	59	+	// /home/runner.
	60	+	Home string
	61	+	// ToolCache is RUNNER_TOOL_CACHE; empty means DefaultToolCache. The official
	62	+	// runner reads it from its own environment (HostContext.cs:411).
	63	+	ToolCache string
57	64		// BaseEnv is the environment steps inherit; nil means os.Environ().
58	65		BaseEnv []string
59	66		// Debug turns ##[debug] lines on (ACTIONS_STEP_DEBUG).

@@ -324,9 +331,17 @@ func (j job) setUp() ([]workflow.Step, error) {
324	331		if err != nil {
325	332		return nil, err
326	333		}
327		-	if cfg.Layout != nil {
	334	+	switch {
	335	+	case cfg.Layout != nil:
328	336		state.WithLayout(*cfg.Layout)
329	337		layout = *cfg.Layout
	338	+	case (cfg.Home != "" && cfg.Home != layout.Home) \|\| cfg.ToolCache != "":
	339	+	home := cfg.Home
	340	+	if home == "" {
	341	+	home = layout.Home
	342	+	}
	343	+	layout = NewLayout(home, filepath.Base(layout.Workspace), cfg.ToolCache)
	344	+	state.WithLayout(layout)
330	345		}
331	346		j.state = state
332	347

@@ -355,7 +370,11 @@ func (j job) setUp() ([]workflow.Step, error) {
355	370		if err != nil {
356	371		return nil, err
357	372		}
	373	+	fc.Own = j.own
358	374		j.fc = fc
	375	+	for _, d := range []string{layout.WorkRoot, layout.RunnerWorkspace, layout.Workspace, layout.Temp, fc.dir} {
	376	+	j.own(d)
	377	+	}
359	378
360	379		j.baseEnv = cfg.BaseEnv
361	380		if j.baseEnv == nil {

@@ -411,6 +430,15 @@ func (j job) setUp() ([]workflow.Step, error) {
411	430		return jb.Steps, nil
412	431		}
413	432
	433	+	// own gives a file or directory created by the executor to the user the steps
	434	+	// run as, so that scripts can write to RUNNER_TEMP, the workspace and the file
	435	+	// command files. Best effort: without Credential nothing is done.
	436	+	func (j *job) own(path string) {
	437	+	if c := j.cfg.Credential; c != nil {
	438	+	_ = os.Lchown(path, int(c.Uid), int(c.Gid))
	439	+	}
	440	+	}
	441	+
414	442		func clearDir(dir string) {
415	443		ents, err := os.ReadDir(dir)
416	444		if err != nil {

@@ -730,8 +758,10 @@ func (j job) execute(s stepRun, stepCancel <-chan struct{}) (result stepResult
730	758		if ev := j.state.EventJSON(); ev != "" {
731	759		dir := filepath.Join(j.state.Layout().Temp, "_github_workflow")
732	760		if err := os.MkdirAll(dir, 0o755); err == nil {
	761	+	j.own(dir)
733	762		p := filepath.Join(dir, "event.json")
734	763		if err := os.WriteFile(p, []byte(ev), 0o644); err == nil {
	764	+	j.own(p)
735	765		j.state.SetGitHub("event_path", p)
736	766		}
737	767		}

 	Runner        RunnerInfo
 	RunnerVersion string
-	// Layout overrides the directory layout (tests); nil means the hosted layout.
+	// Layout overrides the directory layout (tests); nil means the hosted layout
+	// under Home.
 	Layout *Layout
+	// Home is the home directory of the user of the steps; empty means
+	// /home/runner.
+	Home string
+	// ToolCache is RUNNER_TOOL_CACHE; empty means DefaultToolCache. The official
+	// runner reads it from its own environment (HostContext.cs:411).
+	ToolCache string
 	// BaseEnv is the environment steps inherit; nil means os.Environ().
 	BaseEnv []string
 	// Debug turns ##[debug] lines on (ACTIONS_STEP_DEBUG).
 	if err != nil {
 		return nil, err
 	}
-	if cfg.Layout != nil {
+	switch {
+	case cfg.Layout != nil:
 		state.WithLayout(*cfg.Layout)
 		layout = *cfg.Layout
+	case (cfg.Home != "" && cfg.Home != layout.Home) || cfg.ToolCache != "":
+		home := cfg.Home
+		if home == "" {
+			home = layout.Home
+		}
+		layout = NewLayout(home, filepath.Base(layout.Workspace), cfg.ToolCache)
+		state.WithLayout(layout)
 	}
 	j.state = state
 	if err != nil {
 		return nil, err
 	}
+	fc.Own = j.own
 	j.fc = fc
+	for _, d := range []string{layout.WorkRoot, layout.RunnerWorkspace, layout.Workspace, layout.Temp, fc.dir} {
+		j.own(d)
+	}
 	j.baseEnv = cfg.BaseEnv
 	if j.baseEnv == nil {
 	return jb.Steps, nil
 }
+// own gives a file or directory created by the executor to the user the steps
+// run as, so that scripts can write to RUNNER_TEMP, the workspace and the file
+// command files. Best effort: without Credential nothing is done.
+func (j *job) own(path string) {
+	if c := j.cfg.Credential; c != nil {
+		_ = os.Lchown(path, int(c.Uid), int(c.Gid))
+	}
+}
 func clearDir(dir string) {
 	ents, err := os.ReadDir(dir)
 	if err != nil {
 		if ev := j.state.EventJSON(); ev != "" {
 			dir := filepath.Join(j.state.Layout().Temp, "_github_workflow")
 			if err := os.MkdirAll(dir, 0o755); err == nil {
+				j.own(dir)
 				p := filepath.Join(dir, "event.json")
 				if err := os.WriteFile(p, []byte(ev), 0o644); err == nil {
+					j.own(p)
 					j.state.SetGitHub("event_path", p)
 				}
 			}

M internal/guest/exec.go +15 -2


@@ -38,6 +38,9 @@ type ExecConfig struct {
38	38		Version string
39	39		// Shell runs the script as `Shell -c script`; default /bin/sh.
40	40		Shell string
	41	+	// StepUser is the account the steps of a job run as when the executor is
	42	+	// root; default "runner".
	43	+	StepUser string
41	44		// ConnectTimeout bounds the connection attempts (default 30 s).
42	45		ConnectTimeout time.Duration
43	46		// RunTimeout bounds the wait for the host's Run frame (default 60 s).

@@ -57,6 +60,9 @@ func (c *ExecConfig) defaults() {
57	60		if c.Shell == "" {
58	61		c.Shell = "/bin/sh"
59	62		}
	63	+	if c.StepUser == "" {
	64	+	c.StepUser = "runner"
	65	+	}
60	66		if c.ConnectTimeout == 0 {
61	67		c.ConnectTimeout = 30 * time.Second
62	68		}

@@ -77,8 +83,9 @@ func (c *ExecConfig) defaults() {
77	83		}
78	84		}
79	85
80		-	// Exec is `ermite-runner exec`, the CI-1 stub executor: it connects back to the
81		-	// host, receives one Run, runs the script with the shell, streams stdout and
	86	+	// Exec is `ermite-runner exec`: it connects back to the host and receives one
	87	+	// Run. A Run with a job is executed by package executor (see runJob); a Run
	88	+	// without one is the stub of sandbox-check, which runs the script with the shell, streams stdout and
82	89		// stderr as Output frames, and ends with an Exit frame. It returns the exit code
83	90		// it reported. A Cancel frame, or the end of ctx, stops the script's process
84	91		// group with escalating signals (see ExecConfig).

@@ -202,6 +209,12 @@ func serve(ctx context.Context, rw io.ReadWriteCloser, cfg ExecConfig) (int, err
202	209		return 1, ctx.Err()
203	210		}
204	211
	212	+	if run.GetJob() != nil {
	213	+	code := runJob(ctx, conn, cfg, run.GetJob())
	214	+	return sendExit(conn, code, nil)
	215	+	}
	216	+
	217	+	// Without a job this is the stub executor of sandbox-check.
205	218		script := run.GetScript()
206	219		if script == "" {
207	220		script = DefaultScript

 	Version string
 	// Shell runs the script as `Shell -c script`; default /bin/sh.
 	Shell string
+	// StepUser is the account the steps of a job run as when the executor is
+	// root; default "runner".
+	StepUser string
 	// ConnectTimeout bounds the connection attempts (default 30 s).
 	ConnectTimeout time.Duration
 	// RunTimeout bounds the wait for the host's Run frame (default 60 s).
 	if c.Shell == "" {
 		c.Shell = "/bin/sh"
 	}
+	if c.StepUser == "" {
+		c.StepUser = "runner"
+	}
 	if c.ConnectTimeout == 0 {
 		c.ConnectTimeout = 30 * time.Second
 	}
 	}
 }
-// Exec is `ermite-runner exec`, the CI-1 stub executor: it connects back to the
+// Exec is `ermite-runner exec`: it connects back to the host and receives one
-// host, receives one Run, runs the script with the shell, streams stdout and
+// Run. A Run with a job is executed by package executor (see runJob); a Run
+// without one is the stub of sandbox-check, which runs the script with the shell, streams stdout and
 // stderr as Output frames, and ends with an Exit frame. It returns the exit code
 // it reported. A Cancel frame, or the end of ctx, stops the script's process
 // group with escalating signals (see ExecConfig).
 		return 1, ctx.Err()
 	}
+	if run.GetJob() != nil {
+		code := runJob(ctx, conn, cfg, run.GetJob())
+		return sendExit(conn, code, nil)
+	}
+	// Without a job this is the stub executor of sandbox-check.
 	script := run.GetScript()
 	if script == "" {
 		script = DefaultScript

A internal/guest/jobexec.go +224 -0

@@ -0,0 +1,224 @@
	1	+	// SPDX-License-Identifier: MIT
	2	+
	3	+	package guest
	4	+
	5	+	import (
	6	+	"context"
	7	+	"os"
	8	+	"os/user"
	9	+	"strconv"
	10	+	"strings"
	11	+	"syscall"
	12	+	"time"
	13	+	"unicode/utf8"
	14	+
	15	+	"bnema.dev/ermite-runner/internal/execstream"
	16	+	"bnema.dev/ermite-runner/internal/execstream/execpb"
	17	+	"bnema.dev/ermite-runner/internal/executor"
	18	+	)
	19	+
	20	+	// MaxLogLine is the largest piece of a log line that fits an AppendLog line
	21	+	// (64 KiB); longer lines are cut by SplitLine.
	22	+	const MaxLogLine = 64 * 1024
	23	+
	24	+	// SplitLine cuts text into pieces of at most max bytes at UTF-8 character
	25	+	// boundaries. The result has one piece for a short line, and always at least
	26	+	// one (an empty line is one empty piece). Invalid UTF-8 in text is replaced
	27	+	// with U+FFFD first, because the protocol only carries valid text.
	28	+	func SplitLine(text string, max int) []string {
	29	+	if !utf8.ValidString(text) {
	30	+	text = strings.ToValidUTF8(text, "\ufffd")
	31	+	}
	32	+	if len(text) <= max {
	33	+	return []string{text}
	34	+	}
	35	+	var out []string
	36	+	for len(text) > max {
	37	+	cut := max
	38	+	for cut > 0 && !utf8.RuneStart(text[cut]) {
	39	+	cut--
	40	+	}
	41	+	if cut == 0 { // cannot happen with max >= utf8.UTFMax
	42	+	cut = max
	43	+	}
	44	+	out = append(out, text[:cut])
	45	+	text = text[cut:]
	46	+	}
	47	+	return append(out, text)
	48	+	}
	49	+
	50	+	// frameReporter is the executor.Reporter that writes frames on the stream.
	51	+	type frameReporter struct {
	52	+	conn *execstream.Conn
	53	+	// err keeps the first send error; the job goes on without a host.
	54	+	failed chan struct{}
	55	+	}
	56	+
	57	+	func ts(t time.Time) int64 { return t.UnixNano() }
	58	+
	59	+	func conclusionProto(c executor.Conclusion) execpb.StepConclusion {
	60	+	switch c {
	61	+	case executor.ConclusionSuccess:
	62	+	return execpb.StepConclusion_STEP_CONCLUSION_SUCCESS
	63	+	case executor.ConclusionFailure:
	64	+	return execpb.StepConclusion_STEP_CONCLUSION_FAILURE
	65	+	case executor.ConclusionCancelled:
	66	+	return execpb.StepConclusion_STEP_CONCLUSION_CANCELLED
	67	+	case executor.ConclusionSkipped:
	68	+	return execpb.StepConclusion_STEP_CONCLUSION_SKIPPED
	69	+	case executor.ConclusionTimedOut:
	70	+	return execpb.StepConclusion_STEP_CONCLUSION_TIMED_OUT
	71	+	}
	72	+	return execpb.StepConclusion_STEP_CONCLUSION_UNSPECIFIED
	73	+	}
	74	+
	75	+	func (r frameReporter) send(f execpb.Frame) {
	76	+	if err := r.conn.Send(f); err != nil {
	77	+	select {
	78	+	case <-r.failed:
	79	+	default:
	80	+	close(r.failed)
	81	+	}
	82	+	}
	83	+	}
	84	+
	85	+	func (r *frameReporter) StepStarted(n int, name string, at time.Time) {
	86	+	r.send(&execpb.Frame{Body: &execpb.Frame_StepStarted{StepStarted: &execpb.StepStarted{Number: int32(n), Name: name, UnixNano: ts(at)}}})
	87	+	}
	88	+
	89	+	func (r *frameReporter) Log(n int, at time.Time, line string) {
	90	+	for i, piece := range SplitLine(line, MaxLogLine) {
	91	+	r.send(&execpb.Frame{Body: &execpb.Frame_Log{Log: &execpb.Log{Step: int32(n), UnixNano: ts(at), Text: piece, Continued: i > 0}}})
	92	+	}
	93	+	}
	94	+
	95	+	func (r *frameReporter) StepCompleted(n int, name string, c executor.Conclusion, at time.Time) {
	96	+	r.send(&execpb.Frame{Body: &execpb.Frame_StepCompleted{StepCompleted: &execpb.StepCompleted{Number: int32(n), Name: name, Conclusion: conclusionProto(c), UnixNano: ts(at)}}})
	97	+	}
	98	+
	99	+	func (r *frameReporter) JobCompleted(res executor.Result) {
	100	+	r.send(&execpb.Frame{Body: &execpb.Frame_JobResult{JobResult: &execpb.JobResult{
	101	+	Conclusion: conclusionProto(res.Conclusion), Outputs: res.Outputs, Error: res.Error, RunnerShutdown: res.RunnerShutdown,
	102	+	}}})
	103	+	}
	104	+
	105	+	// StepUser is the account the steps run as: uid, gid and supplementary groups,
	106	+	// like the hosted runners' "runner" user. Nil credential means the steps run as
	107	+	// the executor itself.
	108	+	type StepUser struct {
	109	+	Credential *syscall.Credential
	110	+	Name, Home string
	111	+	}
	112	+
	113	+	// LookupStepUser finds the user the steps run as. When the executor is not
	114	+	// root it cannot switch user: the steps run as the current user. When it is
	115	+	// root and name does not exist, the steps run as root (an image without the
	116	+	// user is a test image).
	117	+	func LookupStepUser(name string) StepUser {
	118	+	if os.Geteuid() != 0 \|\| name == "" {
	119	+	u := StepUser{Home: os.Getenv("HOME")}
	120	+	if cu, err := user.Current(); err == nil {
	121	+	u.Name = cu.Username
	122	+	if u.Home == "" {
	123	+	u.Home = cu.HomeDir
	124	+	}
	125	+	}
	126	+	return u
	127	+	}
	128	+	u, err := user.Lookup(name)
	129	+	if err != nil {
	130	+	return StepUser{Name: "root", Home: "/root"}
	131	+	}
	132	+	uid, _ := strconv.ParseUint(u.Uid, 10, 32)
	133	+	gid, _ := strconv.ParseUint(u.Gid, 10, 32)
	134	+	cred := &syscall.Credential{Uid: uint32(uid), Gid: uint32(gid)}
	135	+	if ids, err := u.GroupIds(); err == nil {
	136	+	for _, g := range ids {
	137	+	if n, err := strconv.ParseUint(g, 10, 32); err == nil {
	138	+	cred.Groups = append(cred.Groups, uint32(n))
	139	+	}
	140	+	}
	141	+	}
	142	+	return StepUser{Credential: cred, Name: u.Username, Home: u.HomeDir}
	143	+	}
	144	+
	145	+	// stepEnv is the environment the steps inherit: the executor's, without the
	146	+	// variables of the stream, with HOME, USER and LOGNAME of the step user.
	147	+	func stepEnv(environ []string, u StepUser) []string {
	148	+	var out []string
	149	+	for _, kv := range environ {
	150	+	k, _, _ := strings.Cut(kv, "=")
	151	+	switch {
	152	+	case strings.HasPrefix(k, "ERMITE_"), k == "RUNNER_TOOL_CACHE", k == "HOME", k == "USER", k == "LOGNAME":
	153	+	continue
	154	+	}
	155	+	out = append(out, kv)
	156	+	}
	157	+	if u.Home != "" {
	158	+	out = append(out, "HOME="+u.Home)
	159	+	}
	160	+	if u.Name != "" {
	161	+	out = append(out, "USER="+u.Name, "LOGNAME="+u.Name)
	162	+	}
	163	+	return out
	164	+	}
	165	+
	166	+	func cancelKind(k execpb.CancelKind) executor.CancelReason {
	167	+	switch k {
	168	+	case execpb.CancelKind_CANCEL_KIND_TIMEOUT:
	169	+	return executor.CancelTimeout
	170	+	case execpb.CancelKind_CANCEL_KIND_SHUTDOWN:
	171	+	return executor.CancelShutdown
	172	+	}
	173	+	return executor.CancelUser
	174	+	}
	175	+
	176	+	// runJob runs the job of a Run frame with the executor and reports on conn.
	177	+	// It returns the exit code to report.
	178	+	func runJob(ctx context.Context, conn execstream.Conn, cfg ExecConfig, spec execpb.JobSpec) int {
	179	+	su := LookupStepUser(cfg.StepUser)
	180	+	canceller := executor.NewCanceller()
	181	+	rep := &frameReporter{conn: conn, failed: make(chan struct{})}
	182	+
	183	+	// Frames from the host while the job runs: a Cancel stops the job; a lost
	184	+	// host does too, like a runner shutdown.
	185	+	go func() {
	186	+	for {
	187	+	f, err := conn.Recv()
	188	+	if err != nil {
	189	+	canceller.Cancel(executor.CancelShutdown)
	190	+	return
	191	+	}
	192	+	if c := f.GetCancel(); c != nil {
	193	+	canceller.Cancel(cancelKind(c.GetKind()))
	194	+	}
	195	+	}
	196	+	}()
	197	+	// The executor's own context (SIGTERM forwarded by init: the sandbox is
	198	+	// being stopped) is a shutdown.
	199	+	stop := context.AfterFunc(ctx, func() { canceller.Cancel(executor.CancelShutdown) })
	200	+	defer stop()
	201	+	go func() {
	202	+	select {
	203	+	case <-rep.failed:
	204	+	canceller.Cancel(executor.CancelShutdown)
	205	+	case <-canceller.Done():
	206	+	}
	207	+	}()
	208	+
	209	+	env := spec.GetRunnerEnvironment()
	210	+	res := executor.Run(canceller, executor.Config{
	211	+	WorkflowPath: spec.GetWorkflowPath(), WorkflowYAML: []byte(spec.GetWorkflowYaml()), JobKey: spec.GetJobKey(),
	212	+	ContextsJSON: spec.GetContexts(), MatrixJSON: spec.GetMatrix(), Repo: spec.GetRepo(),
	213	+	TimeoutMinutes: int(spec.GetTimeoutMinutes()),
	214	+	Runner: executor.RunnerInfo{Name: spec.GetRunnerName(), Environment: env},
	215	+	RunnerVersion: spec.GetRunnerVersion(),
	216	+	Home: su.Home, ToolCache: os.Getenv("RUNNER_TOOL_CACHE"), BaseEnv: stepEnv(os.Environ(), su), Credential: su.Credential,
	217	+	Debug: spec.GetDebug(), Reporter: rep,
	218	+	TermAfter: cfg.TermAfter, KillAfter: cfg.KillAfter, WaitDelay: cfg.WaitDelay, Signal: cfg.signal,
	219	+	})
	220	+	if res.Conclusion == executor.ConclusionSuccess {
	221	+	return 0
	222	+	}
	223	+	return 1
	224	+	}

@@ -0,0 +1,224 @@
	1	+	// SPDX-License-Identifier: MIT
	2	+
	3	+	package guest
	4	+
	5	+	import (
	6	+	"context"
	7	+	"os"
	8	+	"os/user"
	9	+	"strconv"
	10	+	"strings"
	11	+	"syscall"
	12	+	"time"
	13	+	"unicode/utf8"
	14	+
	15	+	"bnema.dev/ermite-runner/internal/execstream"
	16	+	"bnema.dev/ermite-runner/internal/execstream/execpb"
	17	+	"bnema.dev/ermite-runner/internal/executor"
	18	+	)
	19	+
	20	+	// MaxLogLine is the largest piece of a log line that fits an AppendLog line
	21	+	// (64 KiB); longer lines are cut by SplitLine.
	22	+	const MaxLogLine = 64 * 1024
	23	+
	24	+	// SplitLine cuts text into pieces of at most max bytes at UTF-8 character
	25	+	// boundaries. The result has one piece for a short line, and always at least
	26	+	// one (an empty line is one empty piece). Invalid UTF-8 in text is replaced
	27	+	// with U+FFFD first, because the protocol only carries valid text.
	28	+	func SplitLine(text string, max int) []string {
	29	+	if !utf8.ValidString(text) {
	30	+	text = strings.ToValidUTF8(text, "\ufffd")
	31	+	}
	32	+	if len(text) <= max {
	33	+	return []string{text}
	34	+	}
	35	+	var out []string
	36	+	for len(text) > max {
	37	+	cut := max
	38	+	for cut > 0 && !utf8.RuneStart(text[cut]) {
	39	+	cut--
	40	+	}
	41	+	if cut == 0 { // cannot happen with max >= utf8.UTFMax
	42	+	cut = max
	43	+	}
	44	+	out = append(out, text[:cut])
	45	+	text = text[cut:]
	46	+	}
	47	+	return append(out, text)
	48	+	}
	49	+
	50	+	// frameReporter is the executor.Reporter that writes frames on the stream.
	51	+	type frameReporter struct {
	52	+	conn *execstream.Conn
	53	+	// err keeps the first send error; the job goes on without a host.
	54	+	failed chan struct{}
	55	+	}
	56	+
	57	+	func ts(t time.Time) int64 { return t.UnixNano() }
	58	+
	59	+	func conclusionProto(c executor.Conclusion) execpb.StepConclusion {
	60	+	switch c {
	61	+	case executor.ConclusionSuccess:
	62	+	return execpb.StepConclusion_STEP_CONCLUSION_SUCCESS
	63	+	case executor.ConclusionFailure:
	64	+	return execpb.StepConclusion_STEP_CONCLUSION_FAILURE
	65	+	case executor.ConclusionCancelled:
	66	+	return execpb.StepConclusion_STEP_CONCLUSION_CANCELLED
	67	+	case executor.ConclusionSkipped:
	68	+	return execpb.StepConclusion_STEP_CONCLUSION_SKIPPED
	69	+	case executor.ConclusionTimedOut:
	70	+	return execpb.StepConclusion_STEP_CONCLUSION_TIMED_OUT
	71	+	}
	72	+	return execpb.StepConclusion_STEP_CONCLUSION_UNSPECIFIED
	73	+	}
	74	+
	75	+	func (r frameReporter) send(f execpb.Frame) {
	76	+	if err := r.conn.Send(f); err != nil {
	77	+	select {
	78	+	case <-r.failed:
	79	+	default:
	80	+	close(r.failed)
	81	+	}
	82	+	}
	83	+	}
	84	+
	85	+	func (r *frameReporter) StepStarted(n int, name string, at time.Time) {
	86	+	r.send(&execpb.Frame{Body: &execpb.Frame_StepStarted{StepStarted: &execpb.StepStarted{Number: int32(n), Name: name, UnixNano: ts(at)}}})
	87	+	}
	88	+
	89	+	func (r *frameReporter) Log(n int, at time.Time, line string) {
	90	+	for i, piece := range SplitLine(line, MaxLogLine) {
	91	+	r.send(&execpb.Frame{Body: &execpb.Frame_Log{Log: &execpb.Log{Step: int32(n), UnixNano: ts(at), Text: piece, Continued: i > 0}}})
	92	+	}
	93	+	}
	94	+
	95	+	func (r *frameReporter) StepCompleted(n int, name string, c executor.Conclusion, at time.Time) {
	96	+	r.send(&execpb.Frame{Body: &execpb.Frame_StepCompleted{StepCompleted: &execpb.StepCompleted{Number: int32(n), Name: name, Conclusion: conclusionProto(c), UnixNano: ts(at)}}})
	97	+	}
	98	+
	99	+	func (r *frameReporter) JobCompleted(res executor.Result) {
	100	+	r.send(&execpb.Frame{Body: &execpb.Frame_JobResult{JobResult: &execpb.JobResult{
	101	+	Conclusion: conclusionProto(res.Conclusion), Outputs: res.Outputs, Error: res.Error, RunnerShutdown: res.RunnerShutdown,
	102	+	}}})
	103	+	}
	104	+
	105	+	// StepUser is the account the steps run as: uid, gid and supplementary groups,
	106	+	// like the hosted runners' "runner" user. Nil credential means the steps run as
	107	+	// the executor itself.
	108	+	type StepUser struct {
	109	+	Credential *syscall.Credential
	110	+	Name, Home string
	111	+	}
	112	+
	113	+	// LookupStepUser finds the user the steps run as. When the executor is not
	114	+	// root it cannot switch user: the steps run as the current user. When it is
	115	+	// root and name does not exist, the steps run as root (an image without the
	116	+	// user is a test image).
	117	+	func LookupStepUser(name string) StepUser {
	118	+	if os.Geteuid() != 0 \|\| name == "" {
	119	+	u := StepUser{Home: os.Getenv("HOME")}
	120	+	if cu, err := user.Current(); err == nil {
	121	+	u.Name = cu.Username
	122	+	if u.Home == "" {
	123	+	u.Home = cu.HomeDir
	124	+	}
	125	+	}
	126	+	return u
	127	+	}
	128	+	u, err := user.Lookup(name)
	129	+	if err != nil {
	130	+	return StepUser{Name: "root", Home: "/root"}
	131	+	}
	132	+	uid, _ := strconv.ParseUint(u.Uid, 10, 32)
	133	+	gid, _ := strconv.ParseUint(u.Gid, 10, 32)
	134	+	cred := &syscall.Credential{Uid: uint32(uid), Gid: uint32(gid)}
	135	+	if ids, err := u.GroupIds(); err == nil {
	136	+	for _, g := range ids {
	137	+	if n, err := strconv.ParseUint(g, 10, 32); err == nil {
	138	+	cred.Groups = append(cred.Groups, uint32(n))
	139	+	}
	140	+	}
	141	+	}
	142	+	return StepUser{Credential: cred, Name: u.Username, Home: u.HomeDir}
	143	+	}
	144	+
	145	+	// stepEnv is the environment the steps inherit: the executor's, without the
	146	+	// variables of the stream, with HOME, USER and LOGNAME of the step user.
	147	+	func stepEnv(environ []string, u StepUser) []string {
	148	+	var out []string
	149	+	for _, kv := range environ {
	150	+	k, _, _ := strings.Cut(kv, "=")
	151	+	switch {
	152	+	case strings.HasPrefix(k, "ERMITE_"), k == "RUNNER_TOOL_CACHE", k == "HOME", k == "USER", k == "LOGNAME":
	153	+	continue
	154	+	}
	155	+	out = append(out, kv)
	156	+	}
	157	+	if u.Home != "" {
	158	+	out = append(out, "HOME="+u.Home)
	159	+	}
	160	+	if u.Name != "" {
	161	+	out = append(out, "USER="+u.Name, "LOGNAME="+u.Name)
	162	+	}
	163	+	return out
	164	+	}
	165	+
	166	+	func cancelKind(k execpb.CancelKind) executor.CancelReason {
	167	+	switch k {
	168	+	case execpb.CancelKind_CANCEL_KIND_TIMEOUT:
	169	+	return executor.CancelTimeout
	170	+	case execpb.CancelKind_CANCEL_KIND_SHUTDOWN:
	171	+	return executor.CancelShutdown
	172	+	}
	173	+	return executor.CancelUser
	174	+	}
	175	+
	176	+	// runJob runs the job of a Run frame with the executor and reports on conn.
	177	+	// It returns the exit code to report.
	178	+	func runJob(ctx context.Context, conn execstream.Conn, cfg ExecConfig, spec execpb.JobSpec) int {
	179	+	su := LookupStepUser(cfg.StepUser)
	180	+	canceller := executor.NewCanceller()
	181	+	rep := &frameReporter{conn: conn, failed: make(chan struct{})}
	182	+
	183	+	// Frames from the host while the job runs: a Cancel stops the job; a lost
	184	+	// host does too, like a runner shutdown.
	185	+	go func() {
	186	+	for {
	187	+	f, err := conn.Recv()
	188	+	if err != nil {
	189	+	canceller.Cancel(executor.CancelShutdown)
	190	+	return
	191	+	}
	192	+	if c := f.GetCancel(); c != nil {
	193	+	canceller.Cancel(cancelKind(c.GetKind()))
	194	+	}
	195	+	}
	196	+	}()
	197	+	// The executor's own context (SIGTERM forwarded by init: the sandbox is
	198	+	// being stopped) is a shutdown.
	199	+	stop := context.AfterFunc(ctx, func() { canceller.Cancel(executor.CancelShutdown) })
	200	+	defer stop()
	201	+	go func() {
	202	+	select {
	203	+	case <-rep.failed:
	204	+	canceller.Cancel(executor.CancelShutdown)
	205	+	case <-canceller.Done():
	206	+	}
	207	+	}()
	208	+
	209	+	env := spec.GetRunnerEnvironment()
	210	+	res := executor.Run(canceller, executor.Config{
	211	+	WorkflowPath: spec.GetWorkflowPath(), WorkflowYAML: []byte(spec.GetWorkflowYaml()), JobKey: spec.GetJobKey(),
	212	+	ContextsJSON: spec.GetContexts(), MatrixJSON: spec.GetMatrix(), Repo: spec.GetRepo(),
	213	+	TimeoutMinutes: int(spec.GetTimeoutMinutes()),
	214	+	Runner: executor.RunnerInfo{Name: spec.GetRunnerName(), Environment: env},
	215	+	RunnerVersion: spec.GetRunnerVersion(),
	216	+	Home: su.Home, ToolCache: os.Getenv("RUNNER_TOOL_CACHE"), BaseEnv: stepEnv(os.Environ(), su), Credential: su.Credential,
	217	+	Debug: spec.GetDebug(), Reporter: rep,
	218	+	TermAfter: cfg.TermAfter, KillAfter: cfg.KillAfter, WaitDelay: cfg.WaitDelay, Signal: cfg.signal,
	219	+	})
	220	+	if res.Conclusion == executor.ConclusionSuccess {
	221	+	return 0
	222	+	}
	223	+	return 1
	224	+	}

A internal/guest/jobexec_test.go +216 -0

@@ -0,0 +1,216 @@
	1	+	// SPDX-License-Identifier: MIT
	2	+
	3	+	package guest
	4	+
	5	+	import (
	6	+	"context"
	7	+	"net"
	8	+	"os/exec"
	9	+	"path/filepath"
	10	+	"strings"
	11	+	"sync"
	12	+	"testing"
	13	+	"time"
	14	+	"unicode/utf8"
	15	+
	16	+	"bnema.dev/ermite-runner/internal/execstream"
	17	+	"bnema.dev/ermite-runner/internal/execstream/execpb"
	18	+	)
	19	+
	20	+	func TestSplitLine(t *testing.T) {
	21	+	if got := SplitLine("", 10); len(got) != 1 \|\| got[0] != "" {
	22	+	t.Fatalf("empty: %q", got)
	23	+	}
	24	+	if got := SplitLine("short", 10); len(got) != 1 \|\| got[0] != "short" {
	25	+	t.Fatalf("%q", got)
	26	+	}
	27	+	// Cuts at character boundaries and joins back exactly.
	28	+	in := strings.Repeat("é€😀a", 40) // 2, 3, 4, 1 bytes
	29	+	for _, max := range []int{4, 5, 7, 10, 33} {
	30	+	parts := SplitLine(in, max)
	31	+	var b strings.Builder
	32	+	for _, p := range parts {
	33	+	if len(p) > max \|\| !utf8.ValidString(p) \|\| p == "" {
	34	+	t.Fatalf("max %d: bad piece %q", max, p)
	35	+	}
	36	+	b.WriteString(p)
	37	+	}
	38	+	if b.String() != in {
	39	+	t.Fatalf("max %d: pieces do not rebuild the line", max)
	40	+	}
	41	+	}
	42	+	// Invalid UTF-8 is replaced before cutting.
	43	+	if got := SplitLine("a\xffb", 10); got[0] != "a\ufffdb" {
	44	+	t.Fatalf("%q", got)
	45	+	}
	46	+	big := SplitLine(strings.Repeat("x", 3*MaxLogLine+5), MaxLogLine)
	47	+	if len(big) != 4 \|\| len(big[0]) != MaxLogLine \|\| len(big[3]) != 5 {
	48	+	t.Fatalf("%d pieces", len(big))
	49	+	}
	50	+	}
	51	+
	52	+	// runJobSession starts Exec on a unix socket and drives it with a job.
	53	+	func runJobSession(t testing.T, yaml string, cancelAfter time.Duration, tweak func(ExecConfig)) (steps []string, logs []execpb.Log, res execpb.JobResult, exit *execpb.Exit) {
	54	+	t.Helper()
	55	+	sock := filepath.Join(t.TempDir(), "s")
	56	+	ln, err := net.Listen("unix", sock)
	57	+	if err != nil {
	58	+	t.Skip(err)
	59	+	}
	60	+	defer func() { _ = ln.Close() }()
	61	+	home := t.TempDir()
	62	+	t.Setenv("HOME", home)
	63	+	t.Setenv("RUNNER_TOOL_CACHE", filepath.Join(home, "toolcache"))
	64	+	t.Setenv("ERMITE_CONNECT", "unix:"+sock)
	65	+	done := make(chan struct{})
	66	+	go func() {
	67	+	defer close(done)
	68	+	cfg := ExecConfig{Target: Target{"unix", sock}, Sandbox: "container", Version: "test", RunTimeout: 5 * time.Second,
	69	+	TermAfter: 200 * time.Millisecond, KillAfter: 200 * time.Millisecond, WaitDelay: 300 * time.Millisecond}
	70	+	if tweak != nil {
	71	+	tweak(&cfg)
	72	+	}
	73	+	_, _ = Exec(context.Background(), cfg)
	74	+	}()
	75	+	conn, err := ln.Accept()
	76	+	if err != nil {
	77	+	t.Fatal(err)
	78	+	}
	79	+	defer func() { _ = conn.Close() }()
	80	+	var cancel chan *execpb.Cancel
	81	+	if cancelAfter > 0 {
	82	+	cancel = make(chan *execpb.Cancel, 1)
	83	+	time.AfterFunc(cancelAfter, func() { cancel <- &execpb.Cancel{Reason: "test", Kind: execpb.CancelKind_CANCEL_KIND_USER} })
	84	+	}
	85	+	var mu sync.Mutex
	86	+	_, exit, err = execstream.DriveJob(conn, &execpb.Run{Job: &execpb.JobSpec{
	87	+	WorkflowPath: ".github/workflows/ci.yml", WorkflowYaml: yaml, JobKey: "j", Repo: "hello", RunnerName: "r", RunnerVersion: "v",
	88	+	Contexts: `{"github":{"repository":"octo/hello","event_name":"push","job":"j"}}`, Matrix: "{}",
	89	+	}}, cancel, execstream.JobHandler{
	90	+	StepStarted: func(s *execpb.StepStarted) {
	91	+	mu.Lock()
	92	+	steps = append(steps, s.Name)
	93	+	mu.Unlock()
	94	+	},
	95	+	Log: func(l *execpb.Log) { logs = append(logs, l) },
	96	+	JobResult: func(r *execpb.JobResult) { res = r },
	97	+	})
	98	+	if err != nil {
	99	+	t.Fatal(err)
	100	+	}
	101	+	<-done
	102	+	return
	103	+	}
	104	+
	105	+	func requireBash(t *testing.T) {
	106	+	t.Helper()
	107	+	if _, err := lookPath("bash"); err != nil {
	108	+	t.Skip("bash not installed")
	109	+	}
	110	+	}
	111	+
	112	+	func TestExecRunsAJob(t *testing.T) {
	113	+	requireBash(t)
	114	+	steps, logs, res, exit := runJobSession(t, "on: push\njobs:\n j:\n runs-on: x\n outputs:\n o: ${{ steps.s.outputs.v }}\n steps:\n - id: s\n run: \|\n echo hello-from-job\n echo v=42 >> \"$GITHUB_OUTPUT\"\n", 0, nil)
	115	+	if res == nil \|\| res.Conclusion != execpb.StepConclusion_STEP_CONCLUSION_SUCCESS \|\| res.Outputs["o"] != "42" \|\| exit.Code != 0 {
	116	+	t.Fatalf("res=%v exit=%v", res, exit)
	117	+	}
	118	+	if strings.Join(steps, "\|") != "Set up job\|Run echo hello-from-job\|Complete job" {
	119	+	t.Fatalf("steps %v", steps)
	120	+	}
	121	+	found := false
	122	+	for _, l := range logs {
	123	+	if l.Text == "hello-from-job" && l.Step == 2 && l.UnixNano > 0 && !l.Continued {
	124	+	found = true
	125	+	}
	126	+	}
	127	+	if !found {
	128	+	t.Fatalf("log line missing: %d lines", len(logs))
	129	+	}
	130	+	}
	131	+
	132	+	func TestExecJobFailureExitsNonZero(t *testing.T) {
	133	+	requireBash(t)
	134	+	_, _, res, exit := runJobSession(t, "on: push\njobs:\n j:\n runs-on: x\n steps:\n - run: exit 4\n", 0, nil)
	135	+	if res.Conclusion != execpb.StepConclusion_STEP_CONCLUSION_FAILURE \|\| exit.Code == 0 {
	136	+	t.Fatalf("res=%v exit=%v", res, exit)
	137	+	}
	138	+	}
	139	+
	140	+	func TestExecJobLongLinesAreContinued(t *testing.T) {
	141	+	requireBash(t)
	142	+	_, logs, res, _ := runJobSession(t, "on: push\njobs:\n j:\n runs-on: x\n steps:\n - run: head -c 150000 /dev/zero \| tr '\\0' y; echo\n", 0, nil)
	143	+	if res.Conclusion != execpb.StepConclusion_STEP_CONCLUSION_SUCCESS {
	144	+	t.Fatalf("%v", res)
	145	+	}
	146	+	var total int
	147	+	var pieces int
	148	+	for _, l := range logs {
	149	+	if strings.HasPrefix(l.Text, "yyyy") {
	150	+	pieces++
	151	+	total += len(l.Text)
	152	+	if len(l.Text) > MaxLogLine \|\| (pieces > 1) != l.Continued {
	153	+	t.Fatalf("piece %d: len %d continued %v", pieces, len(l.Text), l.Continued)
	154	+	}
	155	+	}
	156	+	}
	157	+	if pieces != 3 \|\| total != 150000 {
	158	+	t.Fatalf("pieces=%d total=%d", pieces, total)
	159	+	}
	160	+	}
	161	+
	162	+	func TestExecJobCancel(t *testing.T) {
	163	+	requireBash(t)
	164	+	start := time.Now()
	165	+	_, _, res, _ := runJobSession(t, "on: push\njobs:\n j:\n runs-on: x\n steps:\n - run: \|\n echo ready\n sleep 30\n", 700*time.Millisecond, nil)
	166	+	if res.Conclusion != execpb.StepConclusion_STEP_CONCLUSION_CANCELLED \|\| res.RunnerShutdown {
	167	+	t.Fatalf("%v", res)
	168	+	}
	169	+	if time.Since(start) > 8*time.Second {
	170	+	t.Fatal("cancel too slow")
	171	+	}
	172	+	}
	173	+
	174	+	func TestExecJobHostGoneIsAShutdown(t *testing.T) {
	175	+	requireBash(t)
	176	+	sock := filepath.Join(t.TempDir(), "s")
	177	+	ln, err := net.Listen("unix", sock)
	178	+	if err != nil {
	179	+	t.Skip(err)
	180	+	}
	181	+	defer func() { _ = ln.Close() }()
	182	+	t.Setenv("HOME", t.TempDir())
	183	+	t.Setenv("RUNNER_TOOL_CACHE", filepath.Join(t.TempDir(), "toolcache"))
	184	+	done := make(chan int, 1)
	185	+	go func() {
	186	+	c, _ := Exec(context.Background(), ExecConfig{Target: Target{"unix", sock}, RunTimeout: 5 * time.Second,
	187	+	TermAfter: 100 * time.Millisecond, KillAfter: 100 * time.Millisecond, WaitDelay: 200 * time.Millisecond})
	188	+	done <- c
	189	+	}()
	190	+	conn, _ := ln.Accept()
	191	+	c := execstream.NewConn(conn)
	192	+	if _, err := c.Recv(); err != nil {
	193	+	t.Fatal(err)
	194	+	}
	195	+	_ = c.Send(&execpb.Frame{Body: &execpb.Frame_Run{Run: &execpb.Run{Job: &execpb.JobSpec{
	196	+	WorkflowYaml: "on: push\njobs:\n j:\n runs-on: x\n steps:\n - run: sleep 60\n", JobKey: "j", Repo: "hello",
	197	+	Contexts: `{"github":{"repository":"o/hello"}}`,
	198	+	}}}})
	199	+	time.Sleep(700 * time.Millisecond)
	200	+	_ = conn.Close()
	201	+	select {
	202	+	case <-done:
	203	+	case <-time.After(10 * time.Second):
	204	+	t.Fatal("the executor kept running without a host")
	205	+	}
	206	+	}
	207	+
	208	+	func TestStepEnv(t *testing.T) {
	209	+	got := stepEnv([]string{"PATH=/bin", "HOME=/root", "ERMITE_CONNECT=x", "USER=root", "RUNNER_TOOL_CACHE=/x", "FOO=1"}, StepUser{Name: "runner", Home: "/home/runner"})
	210	+	want := "PATH=/bin FOO=1 HOME=/home/runner USER=runner LOGNAME=runner"
	211	+	if strings.Join(got, " ") != want {
	212	+	t.Fatalf("%v", got)
	213	+	}
	214	+	}
	215	+
	216	+	func lookPath(name string) (string, error) { return exec.LookPath(name) }

@@ -0,0 +1,216 @@
	1	+	// SPDX-License-Identifier: MIT
	2	+
	3	+	package guest
	4	+
	5	+	import (
	6	+	"context"
	7	+	"net"
	8	+	"os/exec"
	9	+	"path/filepath"
	10	+	"strings"
	11	+	"sync"
	12	+	"testing"
	13	+	"time"
	14	+	"unicode/utf8"
	15	+
	16	+	"bnema.dev/ermite-runner/internal/execstream"
	17	+	"bnema.dev/ermite-runner/internal/execstream/execpb"
	18	+	)
	19	+
	20	+	func TestSplitLine(t *testing.T) {
	21	+	if got := SplitLine("", 10); len(got) != 1 \|\| got[0] != "" {
	22	+	t.Fatalf("empty: %q", got)
	23	+	}
	24	+	if got := SplitLine("short", 10); len(got) != 1 \|\| got[0] != "short" {
	25	+	t.Fatalf("%q", got)
	26	+	}
	27	+	// Cuts at character boundaries and joins back exactly.
	28	+	in := strings.Repeat("é€😀a", 40) // 2, 3, 4, 1 bytes
	29	+	for _, max := range []int{4, 5, 7, 10, 33} {
	30	+	parts := SplitLine(in, max)
	31	+	var b strings.Builder
	32	+	for _, p := range parts {
	33	+	if len(p) > max \|\| !utf8.ValidString(p) \|\| p == "" {
	34	+	t.Fatalf("max %d: bad piece %q", max, p)
	35	+	}
	36	+	b.WriteString(p)
	37	+	}
	38	+	if b.String() != in {
	39	+	t.Fatalf("max %d: pieces do not rebuild the line", max)
	40	+	}
	41	+	}
	42	+	// Invalid UTF-8 is replaced before cutting.
	43	+	if got := SplitLine("a\xffb", 10); got[0] != "a\ufffdb" {
	44	+	t.Fatalf("%q", got)
	45	+	}
	46	+	big := SplitLine(strings.Repeat("x", 3*MaxLogLine+5), MaxLogLine)
	47	+	if len(big) != 4 \|\| len(big[0]) != MaxLogLine \|\| len(big[3]) != 5 {
	48	+	t.Fatalf("%d pieces", len(big))
	49	+	}
	50	+	}
	51	+
	52	+	// runJobSession starts Exec on a unix socket and drives it with a job.
	53	+	func runJobSession(t testing.T, yaml string, cancelAfter time.Duration, tweak func(ExecConfig)) (steps []string, logs []execpb.Log, res execpb.JobResult, exit *execpb.Exit) {
	54	+	t.Helper()
	55	+	sock := filepath.Join(t.TempDir(), "s")
	56	+	ln, err := net.Listen("unix", sock)
	57	+	if err != nil {
	58	+	t.Skip(err)
	59	+	}
	60	+	defer func() { _ = ln.Close() }()
	61	+	home := t.TempDir()
	62	+	t.Setenv("HOME", home)
	63	+	t.Setenv("RUNNER_TOOL_CACHE", filepath.Join(home, "toolcache"))
	64	+	t.Setenv("ERMITE_CONNECT", "unix:"+sock)
	65	+	done := make(chan struct{})
	66	+	go func() {
	67	+	defer close(done)
	68	+	cfg := ExecConfig{Target: Target{"unix", sock}, Sandbox: "container", Version: "test", RunTimeout: 5 * time.Second,
	69	+	TermAfter: 200 * time.Millisecond, KillAfter: 200 * time.Millisecond, WaitDelay: 300 * time.Millisecond}
	70	+	if tweak != nil {
	71	+	tweak(&cfg)
	72	+	}
	73	+	_, _ = Exec(context.Background(), cfg)
	74	+	}()
	75	+	conn, err := ln.Accept()
	76	+	if err != nil {
	77	+	t.Fatal(err)
	78	+	}
	79	+	defer func() { _ = conn.Close() }()
	80	+	var cancel chan *execpb.Cancel
	81	+	if cancelAfter > 0 {
	82	+	cancel = make(chan *execpb.Cancel, 1)
	83	+	time.AfterFunc(cancelAfter, func() { cancel <- &execpb.Cancel{Reason: "test", Kind: execpb.CancelKind_CANCEL_KIND_USER} })
	84	+	}
	85	+	var mu sync.Mutex
	86	+	_, exit, err = execstream.DriveJob(conn, &execpb.Run{Job: &execpb.JobSpec{
	87	+	WorkflowPath: ".github/workflows/ci.yml", WorkflowYaml: yaml, JobKey: "j", Repo: "hello", RunnerName: "r", RunnerVersion: "v",
	88	+	Contexts: `{"github":{"repository":"octo/hello","event_name":"push","job":"j"}}`, Matrix: "{}",
	89	+	}}, cancel, execstream.JobHandler{
	90	+	StepStarted: func(s *execpb.StepStarted) {
	91	+	mu.Lock()
	92	+	steps = append(steps, s.Name)
	93	+	mu.Unlock()
	94	+	},
	95	+	Log: func(l *execpb.Log) { logs = append(logs, l) },
	96	+	JobResult: func(r *execpb.JobResult) { res = r },
	97	+	})
	98	+	if err != nil {
	99	+	t.Fatal(err)
	100	+	}
	101	+	<-done
	102	+	return
	103	+	}
	104	+
	105	+	func requireBash(t *testing.T) {
	106	+	t.Helper()
	107	+	if _, err := lookPath("bash"); err != nil {
	108	+	t.Skip("bash not installed")
	109	+	}
	110	+	}
	111	+
	112	+	func TestExecRunsAJob(t *testing.T) {
	113	+	requireBash(t)
	114	+	steps, logs, res, exit := runJobSession(t, "on: push\njobs:\n j:\n runs-on: x\n outputs:\n o: ${{ steps.s.outputs.v }}\n steps:\n - id: s\n run: \|\n echo hello-from-job\n echo v=42 >> \"$GITHUB_OUTPUT\"\n", 0, nil)
	115	+	if res == nil \|\| res.Conclusion != execpb.StepConclusion_STEP_CONCLUSION_SUCCESS \|\| res.Outputs["o"] != "42" \|\| exit.Code != 0 {
	116	+	t.Fatalf("res=%v exit=%v", res, exit)
	117	+	}
	118	+	if strings.Join(steps, "\|") != "Set up job\|Run echo hello-from-job\|Complete job" {
	119	+	t.Fatalf("steps %v", steps)
	120	+	}
	121	+	found := false
	122	+	for _, l := range logs {
	123	+	if l.Text == "hello-from-job" && l.Step == 2 && l.UnixNano > 0 && !l.Continued {
	124	+	found = true
	125	+	}
	126	+	}
	127	+	if !found {
	128	+	t.Fatalf("log line missing: %d lines", len(logs))
	129	+	}
	130	+	}
	131	+
	132	+	func TestExecJobFailureExitsNonZero(t *testing.T) {
	133	+	requireBash(t)
	134	+	_, _, res, exit := runJobSession(t, "on: push\njobs:\n j:\n runs-on: x\n steps:\n - run: exit 4\n", 0, nil)
	135	+	if res.Conclusion != execpb.StepConclusion_STEP_CONCLUSION_FAILURE \|\| exit.Code == 0 {
	136	+	t.Fatalf("res=%v exit=%v", res, exit)
	137	+	}
	138	+	}
	139	+
	140	+	func TestExecJobLongLinesAreContinued(t *testing.T) {
	141	+	requireBash(t)
	142	+	_, logs, res, _ := runJobSession(t, "on: push\njobs:\n j:\n runs-on: x\n steps:\n - run: head -c 150000 /dev/zero \| tr '\\0' y; echo\n", 0, nil)
	143	+	if res.Conclusion != execpb.StepConclusion_STEP_CONCLUSION_SUCCESS {
	144	+	t.Fatalf("%v", res)
	145	+	}
	146	+	var total int
	147	+	var pieces int
	148	+	for _, l := range logs {
	149	+	if strings.HasPrefix(l.Text, "yyyy") {
	150	+	pieces++
	151	+	total += len(l.Text)
	152	+	if len(l.Text) > MaxLogLine \|\| (pieces > 1) != l.Continued {
	153	+	t.Fatalf("piece %d: len %d continued %v", pieces, len(l.Text), l.Continued)
	154	+	}
	155	+	}
	156	+	}
	157	+	if pieces != 3 \|\| total != 150000 {
	158	+	t.Fatalf("pieces=%d total=%d", pieces, total)
	159	+	}
	160	+	}
	161	+
	162	+	func TestExecJobCancel(t *testing.T) {
	163	+	requireBash(t)
	164	+	start := time.Now()
	165	+	_, _, res, _ := runJobSession(t, "on: push\njobs:\n j:\n runs-on: x\n steps:\n - run: \|\n echo ready\n sleep 30\n", 700*time.Millisecond, nil)
	166	+	if res.Conclusion != execpb.StepConclusion_STEP_CONCLUSION_CANCELLED \|\| res.RunnerShutdown {
	167	+	t.Fatalf("%v", res)
	168	+	}
	169	+	if time.Since(start) > 8*time.Second {
	170	+	t.Fatal("cancel too slow")
	171	+	}
	172	+	}
	173	+
	174	+	func TestExecJobHostGoneIsAShutdown(t *testing.T) {
	175	+	requireBash(t)
	176	+	sock := filepath.Join(t.TempDir(), "s")
	177	+	ln, err := net.Listen("unix", sock)
	178	+	if err != nil {
	179	+	t.Skip(err)
	180	+	}
	181	+	defer func() { _ = ln.Close() }()
	182	+	t.Setenv("HOME", t.TempDir())
	183	+	t.Setenv("RUNNER_TOOL_CACHE", filepath.Join(t.TempDir(), "toolcache"))
	184	+	done := make(chan int, 1)
	185	+	go func() {
	186	+	c, _ := Exec(context.Background(), ExecConfig{Target: Target{"unix", sock}, RunTimeout: 5 * time.Second,
	187	+	TermAfter: 100 * time.Millisecond, KillAfter: 100 * time.Millisecond, WaitDelay: 200 * time.Millisecond})
	188	+	done <- c
	189	+	}()
	190	+	conn, _ := ln.Accept()
	191	+	c := execstream.NewConn(conn)
	192	+	if _, err := c.Recv(); err != nil {
	193	+	t.Fatal(err)
	194	+	}
	195	+	_ = c.Send(&execpb.Frame{Body: &execpb.Frame_Run{Run: &execpb.Run{Job: &execpb.JobSpec{
	196	+	WorkflowYaml: "on: push\njobs:\n j:\n runs-on: x\n steps:\n - run: sleep 60\n", JobKey: "j", Repo: "hello",
	197	+	Contexts: `{"github":{"repository":"o/hello"}}`,
	198	+	}}}})
	199	+	time.Sleep(700 * time.Millisecond)
	200	+	_ = conn.Close()
	201	+	select {
	202	+	case <-done:
	203	+	case <-time.After(10 * time.Second):
	204	+	t.Fatal("the executor kept running without a host")
	205	+	}
	206	+	}
	207	+
	208	+	func TestStepEnv(t *testing.T) {
	209	+	got := stepEnv([]string{"PATH=/bin", "HOME=/root", "ERMITE_CONNECT=x", "USER=root", "RUNNER_TOOL_CACHE=/x", "FOO=1"}, StepUser{Name: "runner", Home: "/home/runner"})
	210	+	want := "PATH=/bin FOO=1 HOME=/home/runner USER=runner LOGNAME=runner"
	211	+	if strings.Join(got, " ") != want {
	212	+	t.Fatalf("%v", got)
	213	+	}
	214	+	}
	215	+
	216	+	func lookPath(name string) (string, error) { return exec.LookPath(name) }